1107 lines
41 KiB
Nim
1107 lines
41 KiB
Nim
# beacon_chain
|
|
# Copyright (c) 2021-2024 Status Research & Development GmbH
|
|
# Licensed and distributed under either of
|
|
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
|
|
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
|
|
# at your option. This file may not be copied, modified, or distributed except according to those terms.
|
|
|
|
{.push raises: [].}
|
|
|
|
import
|
|
chronicles,
|
|
".."/validators/activity_metrics,
|
|
".."/spec/forks,
|
|
"."/[common, api, fallback_service]
|
|
|
|
const
|
|
ServiceName = "block_service"
|
|
BlockPollInterval = attestationSlotOffset.nanoseconds div 4
|
|
BlockPollOffset1 = TimeDiff(nanoseconds: BlockPollInterval)
|
|
BlockPollOffset2 = TimeDiff(nanoseconds: BlockPollInterval * 2)
|
|
BlockPollOffset3 = TimeDiff(nanoseconds: BlockPollInterval * 3)
|
|
|
|
logScope: service = ServiceName
|
|
|
|
type
|
|
PreparedBeaconBlock = object
|
|
blockRoot*: Eth2Digest
|
|
data*: ForkedBeaconBlock
|
|
kzgProofsOpt*: Opt[deneb.KzgProofs]
|
|
blobsOpt*: Opt[deneb.Blobs]
|
|
|
|
PreparedBlindedBeaconBlock = object
|
|
blockRoot*: Eth2Digest
|
|
data*: ForkedBlindedBeaconBlock
|
|
|
|
func shortLog(v: Opt[UInt256]): auto =
|
|
if v.isNone(): "<not available>" else: toString(v.get, 10)
|
|
|
|
func shortLog(v: ForkedMaybeBlindedBeaconBlock): auto =
|
|
withForkyMaybeBlindedBlck(v):
|
|
when consensusFork == ConsensusFork.Electra:
|
|
debugRaiseAssert ""
|
|
shortLog(default(phase0.BeaconBlock))
|
|
elif consensusFork < ConsensusFork.Deneb:
|
|
shortLog(forkyMaybeBlindedBlck)
|
|
else:
|
|
when isBlinded:
|
|
shortLog(forkyMaybeBlindedBlck)
|
|
else:
|
|
shortLog(forkyMaybeBlindedBlck.`block`)
|
|
|
|
proc proposeBlock(vc: ValidatorClientRef, slot: Slot,
|
|
proposerKey: ValidatorPubKey) {.async.}
|
|
|
|
proc produceBlock(
|
|
vc: ValidatorClientRef,
|
|
currentSlot, slot: Slot,
|
|
randao_reveal: ValidatorSig,
|
|
graffiti: GraffitiBytes,
|
|
validator: AttachedValidator
|
|
): Future[Opt[PreparedBeaconBlock]] {.async.} =
|
|
logScope:
|
|
slot = slot
|
|
wall_slot = currentSlot
|
|
validator = validatorLog(validator)
|
|
let
|
|
produceBlockResponse =
|
|
try:
|
|
await vc.produceBlockV2(slot, randao_reveal, graffiti,
|
|
ApiStrategyKind.Best)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to retrieve block data", reason = exc.getFailureReason()
|
|
return Opt.none(PreparedBeaconBlock)
|
|
except CancelledError as exc:
|
|
debug "Block data production has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while getting block data",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return Opt.none(PreparedBeaconBlock)
|
|
case produceBlockResponse.kind
|
|
of ConsensusFork.Phase0:
|
|
let blck = produceBlockResponse.phase0Data
|
|
return Opt.some(PreparedBeaconBlock(blockRoot: hash_tree_root(blck),
|
|
data: ForkedBeaconBlock.init(blck),
|
|
kzgProofsOpt: Opt.none(deneb.KzgProofs),
|
|
blobsOpt: Opt.none(deneb.Blobs)))
|
|
of ConsensusFork.Altair:
|
|
let blck = produceBlockResponse.altairData
|
|
return Opt.some(PreparedBeaconBlock(blockRoot: hash_tree_root(blck),
|
|
data: ForkedBeaconBlock.init(blck),
|
|
kzgProofsOpt: Opt.none(deneb.KzgProofs),
|
|
blobsOpt: Opt.none(deneb.Blobs)))
|
|
of ConsensusFork.Bellatrix:
|
|
let blck = produceBlockResponse.bellatrixData
|
|
return Opt.some(PreparedBeaconBlock(blockRoot: hash_tree_root(blck),
|
|
data: ForkedBeaconBlock.init(blck),
|
|
kzgProofsOpt: Opt.none(deneb.KzgProofs),
|
|
blobsOpt: Opt.none(deneb.Blobs)))
|
|
of ConsensusFork.Capella:
|
|
let blck = produceBlockResponse.capellaData
|
|
return Opt.some(PreparedBeaconBlock(blockRoot: hash_tree_root(blck),
|
|
data: ForkedBeaconBlock.init(blck),
|
|
kzgProofsOpt: Opt.none(deneb.KzgProofs),
|
|
blobsOpt: Opt.none(deneb.Blobs)))
|
|
of ConsensusFork.Deneb:
|
|
let
|
|
blck = produceBlockResponse.denebData.`block`
|
|
kzgProofs = produceBlockResponse.denebData.kzg_proofs
|
|
blobs = produceBlockResponse.denebData.blobs
|
|
return Opt.some(PreparedBeaconBlock(blockRoot: hash_tree_root(blck),
|
|
data: ForkedBeaconBlock.init(blck),
|
|
kzgProofsOpt: Opt.some(kzgProofs),
|
|
blobsOpt: Opt.some(blobs)))
|
|
of ConsensusFork.Electra:
|
|
let
|
|
blck = produceBlockResponse.electraData.`block`
|
|
kzgProofs = produceBlockResponse.electraData.kzg_proofs
|
|
blobs = produceBlockResponse.electraData.blobs
|
|
return Opt.some(PreparedBeaconBlock(blockRoot: hash_tree_root(blck),
|
|
data: ForkedBeaconBlock.init(blck),
|
|
kzgProofsOpt: Opt.some(kzgProofs),
|
|
blobsOpt: Opt.some(blobs)))
|
|
|
|
proc produceBlindedBlock(
|
|
vc: ValidatorClientRef,
|
|
currentSlot, slot: Slot,
|
|
randao_reveal: ValidatorSig,
|
|
graffiti: GraffitiBytes,
|
|
validator: AttachedValidator
|
|
): Future[Opt[PreparedBlindedBeaconBlock]] {.async.} =
|
|
logScope:
|
|
slot = slot
|
|
wall_slot = currentSlot
|
|
validator = validatorLog(validator)
|
|
let
|
|
beaconBlock =
|
|
try:
|
|
await vc.produceBlindedBlock(slot, randao_reveal, graffiti,
|
|
ApiStrategyKind.Best)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to retrieve blinded block data", error_msg = exc.msg,
|
|
reason = exc.getFailureReason()
|
|
return Opt.none(PreparedBlindedBeaconBlock)
|
|
except CancelledError as exc:
|
|
debug "Blinded block data production has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while getting blinded block data",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return Opt.none(PreparedBlindedBeaconBlock)
|
|
blockRoot = withBlck(beaconBlock): hash_tree_root(forkyBlck)
|
|
|
|
return Opt.some(
|
|
PreparedBlindedBeaconBlock(blockRoot: blockRoot, data: beaconBlock))
|
|
|
|
proc lazyWait[T](fut: Future[T]) {.async.} =
|
|
try:
|
|
discard await fut
|
|
except CatchableError:
|
|
discard
|
|
|
|
proc prepareRandao(vc: ValidatorClientRef, slot: Slot,
|
|
proposerKey: ValidatorPubKey) {.async.} =
|
|
if slot == GENESIS_SLOT:
|
|
return
|
|
|
|
let
|
|
destSlot = slot - 1'u64
|
|
destOffset = TimeDiff(nanoseconds: NANOSECONDS_PER_SLOT.int64 div 2)
|
|
deadline = destSlot.start_beacon_time() + destOffset
|
|
epoch = slot.epoch()
|
|
# We going to wait to T - (T / 4 * 2), where T is proposer's
|
|
# duty slot.
|
|
currentSlot = (await vc.checkedWaitForSlot(destSlot, destOffset,
|
|
false)).valueOr:
|
|
debug "Unable to perform RANDAO signature preparation because of " &
|
|
"system time failure"
|
|
return
|
|
validator =
|
|
vc.getValidatorForDuties(proposerKey, slot, true).valueOr: return
|
|
|
|
if currentSlot <= destSlot:
|
|
# We do not need result, because we want it to be cached.
|
|
let
|
|
start = Moment.now()
|
|
genesisRoot = vc.beaconGenesis.genesis_validators_root
|
|
fork = vc.forkAtEpoch(epoch)
|
|
rsig = await validator.getEpochSignature(fork, genesisRoot, epoch)
|
|
timeElapsed = Moment.now() - start
|
|
if rsig.isErr():
|
|
debug "Unable to prepare RANDAO signature", epoch = epoch,
|
|
validator = validatorLog(validator), elapsed_time = timeElapsed,
|
|
current_slot = currentSlot, destination_slot = destSlot,
|
|
delay = vc.getDelay(deadline)
|
|
else:
|
|
debug "RANDAO signature has been prepared", epoch = epoch,
|
|
validator = validatorLog(validator), elapsed_time = timeElapsed,
|
|
current_slot = currentSlot, destination_slot = destSlot,
|
|
delay = vc.getDelay(deadline)
|
|
else:
|
|
debug "RANDAO signature preparation timed out", epoch = epoch,
|
|
validator = validatorLog(validator),
|
|
current_slot = currentSlot, destination_slot = destSlot,
|
|
delay = vc.getDelay(deadline)
|
|
|
|
proc spawnProposalTask(vc: ValidatorClientRef,
|
|
duty: RestProposerDuty): ProposerTask =
|
|
ProposerTask(
|
|
randaoFut: prepareRandao(vc, duty.slot, duty.pubkey),
|
|
proposeFut: proposeBlock(vc, duty.slot, duty.pubkey),
|
|
duty: duty
|
|
)
|
|
|
|
proc isProduceBlockV3Supported(vc: ValidatorClientRef): bool =
|
|
let
|
|
# Both `statuses` and `roles` should be set to values which are used in
|
|
# api.produceBlockV3() call.
|
|
statuses = ViableNodeStatus
|
|
roles = {BeaconNodeRole.BlockProposalData}
|
|
|
|
for node in vc.filterNodes(statuses, roles):
|
|
if RestBeaconNodeFeature.NoProduceBlockV3 notin node.features:
|
|
return true
|
|
false
|
|
|
|
proc publishBlockV3(vc: ValidatorClientRef, currentSlot, slot: Slot,
|
|
fork: Fork, randaoReveal: ValidatorSig,
|
|
validator: AttachedValidator): Future[bool] {.async.} =
|
|
let
|
|
genesisRoot = vc.beaconGenesis.genesis_validators_root
|
|
graffiti =
|
|
if vc.config.graffiti.isSome():
|
|
vc.config.graffiti.get()
|
|
else:
|
|
defaultGraffitiBytes()
|
|
vindex = validator.index.get()
|
|
|
|
logScope:
|
|
validator = validatorLog(validator)
|
|
validator_index = vindex
|
|
slot = slot
|
|
wall_slot = currentSlot
|
|
|
|
let
|
|
maybeBlock =
|
|
try:
|
|
await vc.produceBlockV3(slot, randao_reveal, graffiti,
|
|
ApiStrategyKind.Best)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to retrieve block data", reason = exc.getFailureReason()
|
|
return false
|
|
except CancelledError as exc:
|
|
debug "Block data production has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while getting block data",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return false
|
|
|
|
withForkyMaybeBlindedBlck(maybeBlock):
|
|
when isBlinded:
|
|
let
|
|
blockRoot = hash_tree_root(forkyMaybeBlindedBlck)
|
|
|
|
debug "Block produced",
|
|
block_type = "blinded",
|
|
block_root = shortLog(blockRoot),
|
|
blck = shortLog(maybeBlock),
|
|
execution_value = shortLog(maybeBlock.executionValue),
|
|
consensus_value = shortLog(maybeBlock.consensusValue)
|
|
|
|
let
|
|
signingRoot =
|
|
compute_block_signing_root(fork, genesisRoot, slot, blockRoot)
|
|
notSlashable = vc.attachedValidators[]
|
|
.slashingProtection
|
|
.registerBlock(vindex, validator.pubkey, slot, signingRoot)
|
|
|
|
logScope:
|
|
blck = shortLog(forkyMaybeBlindedBlck)
|
|
block_root = shortLog(blockRoot)
|
|
signing_root = shortLog(signingRoot)
|
|
|
|
if notSlashable.isErr():
|
|
warn "Slashing protection activated for blinded block proposal"
|
|
return false
|
|
|
|
let signature =
|
|
try:
|
|
let res = await validator.getBlockSignature(fork, genesisRoot,
|
|
slot, blockRoot,
|
|
maybeBlock)
|
|
if res.isErr():
|
|
warn "Unable to sign blinded block proposal using remote signer",
|
|
reason = res.error()
|
|
return false
|
|
res.get()
|
|
except CancelledError as exc:
|
|
debug "Blinded block signature process has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while signing blinded block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return false
|
|
|
|
let
|
|
signedBlock =
|
|
ForkedSignedBlindedBeaconBlock.init(forkyMaybeBlindedBlck,
|
|
blockRoot, signature)
|
|
res =
|
|
try:
|
|
debug "Sending blinded block"
|
|
await vc.publishBlindedBlock(signedBlock, ApiStrategyKind.First)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to publish blinded block",
|
|
reason = exc.getFailureReason()
|
|
return false
|
|
except CancelledError as exc:
|
|
debug "Blinded block publication has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while publishing blinded " &
|
|
"block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return false
|
|
|
|
if res:
|
|
let delay = vc.getDelay(slot.block_deadline())
|
|
beacon_blocks_sent.inc()
|
|
beacon_blocks_sent_delay.observe(delay.toFloatSeconds())
|
|
notice "Blinded block published", delay = delay
|
|
true
|
|
else:
|
|
warn "Blinded block was not accepted by beacon node"
|
|
false
|
|
else:
|
|
debugRaiseAssert "electra htr"
|
|
let
|
|
blockRoot = hash_tree_root(
|
|
when consensusFork == ConsensusFork.Electra:
|
|
default(Attestation)
|
|
elif consensusFork < ConsensusFork.Deneb:
|
|
forkyMaybeBlindedBlck
|
|
else:
|
|
forkyMaybeBlindedBlck.`block`
|
|
)
|
|
|
|
debug "Block produced",
|
|
block_type = "non-blinded",
|
|
block_root = shortLog(blockRoot),
|
|
blck = shortLog(maybeBlock),
|
|
execution_value = shortLog(maybeBlock.executionValue),
|
|
consensus_value = shortLog(maybeBlock.consensusValue)
|
|
|
|
let
|
|
signingRoot =
|
|
compute_block_signing_root(fork, genesisRoot, slot, blockRoot)
|
|
notSlashable = vc.attachedValidators[]
|
|
.slashingProtection
|
|
.registerBlock(vindex, validator.pubkey, slot, signingRoot)
|
|
|
|
debugRaiseAssert "electra logging"
|
|
logScope:
|
|
blck = shortLog(
|
|
when consensusFork == ConsensusFork.Electra:
|
|
default(phase0.BeaconBlock)
|
|
elif consensusFork < ConsensusFork.Deneb:
|
|
forkyMaybeBlindedBlck
|
|
else:
|
|
forkyMaybeBlindedBlck.`block`
|
|
)
|
|
block_root = shortLog(blockRoot)
|
|
signing_root = shortLog(signingRoot)
|
|
|
|
if notSlashable.isErr():
|
|
warn "Slashing protection activated for block proposal"
|
|
return false
|
|
|
|
let
|
|
signature =
|
|
try:
|
|
let res = await validator.getBlockSignature(
|
|
fork, genesisRoot, slot, blockRoot, maybeBlock)
|
|
if res.isErr():
|
|
warn "Unable to sign block proposal using remote signer",
|
|
reason = res.error()
|
|
return false
|
|
res.get()
|
|
except CancelledError as exc:
|
|
debug "Block signature process has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while signing block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return false
|
|
|
|
signedBlockContents =
|
|
RestPublishedSignedBlockContents.init(
|
|
forkyMaybeBlindedBlck, blockRoot, signature)
|
|
|
|
res =
|
|
try:
|
|
debug "Sending block"
|
|
await vc.publishBlock(signedBlockContents, ApiStrategyKind.First)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to publish block", reason = exc.getFailureReason()
|
|
return false
|
|
except CancelledError as exc:
|
|
debug "Block publication has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while publishing block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return false
|
|
|
|
if res:
|
|
let delay = vc.getDelay(slot.block_deadline())
|
|
beacon_blocks_sent.inc()
|
|
beacon_blocks_sent_delay.observe(delay.toFloatSeconds())
|
|
notice "Block published", delay = delay
|
|
true
|
|
else:
|
|
warn "Block was not accepted by beacon node"
|
|
false
|
|
|
|
proc publishBlockV2(vc: ValidatorClientRef, currentSlot, slot: Slot,
|
|
fork: Fork, randaoReveal: ValidatorSig,
|
|
validator: AttachedValidator) {.async.} =
|
|
let
|
|
genesisRoot = vc.beaconGenesis.genesis_validators_root
|
|
graffiti = vc.getGraffitiBytes(validator)
|
|
vindex = validator.index.get()
|
|
|
|
logScope:
|
|
validator = validatorLog(validator)
|
|
validator_index = vindex
|
|
slot = slot
|
|
wall_slot = currentSlot
|
|
|
|
var beaconBlocks =
|
|
block:
|
|
let blindedBlockFut =
|
|
if vc.config.payloadBuilderEnable:
|
|
vc.produceBlindedBlock(currentSlot, slot, randaoReveal, graffiti,
|
|
validator)
|
|
else:
|
|
nil
|
|
let normalBlockFut = vc.produceBlock(currentSlot, slot, randaoReveal,
|
|
graffiti, validator)
|
|
let blindedBlock =
|
|
if isNil(blindedBlockFut):
|
|
Opt.none(PreparedBlindedBeaconBlock)
|
|
else:
|
|
try:
|
|
await blindedBlockFut
|
|
except CancelledError as exc:
|
|
if not(normalBlockFut.finished()):
|
|
await normalBlockFut.cancelAndWait()
|
|
raise exc
|
|
except CatchableError:
|
|
# This should not happen, because all the exceptions are handled.
|
|
Opt.none(PreparedBlindedBeaconBlock)
|
|
|
|
let normalBlock =
|
|
if blindedBlock.isNone():
|
|
try:
|
|
await normalBlockFut
|
|
except CancelledError as exc:
|
|
raise exc
|
|
except CatchableError:
|
|
# This should not happen, because all the exceptions are handled.
|
|
Opt.none(PreparedBeaconBlock)
|
|
else:
|
|
if not(normalBlockFut.finished()):
|
|
asyncSpawn lazyWait(normalBlockFut)
|
|
Opt.none(PreparedBeaconBlock)
|
|
|
|
if blindedBlock.isNone() and normalBlock.isNone():
|
|
return
|
|
|
|
(blindedBlock: blindedBlock, normalBlock: normalBlock)
|
|
|
|
if beaconBlocks.blindedBlock.isSome():
|
|
let
|
|
preparedBlock = beaconBlocks.blindedBlock.get()
|
|
signingRoot = compute_block_signing_root(fork, genesisRoot, slot,
|
|
preparedBlock.blockRoot)
|
|
notSlashable = vc.attachedValidators[]
|
|
.slashingProtection
|
|
.registerBlock(vindex, validator.pubkey, slot, signingRoot)
|
|
|
|
logScope:
|
|
blck = shortLog(preparedBlock.data)
|
|
block_root = shortLog(preparedBlock.blockRoot)
|
|
signing_root = shortLog(signingRoot)
|
|
|
|
if notSlashable.isOk():
|
|
let
|
|
signature =
|
|
try:
|
|
let res = await validator.getBlockSignature(fork, genesisRoot,
|
|
slot,
|
|
preparedBlock.blockRoot,
|
|
preparedBlock.data)
|
|
if res.isErr():
|
|
warn "Unable to sign blinded block proposal using remote signer",
|
|
reason = res.error()
|
|
return
|
|
res.get()
|
|
except CancelledError as exc:
|
|
debug "Blinded block signature process has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while signing blinded block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return
|
|
|
|
logScope:
|
|
signature = shortLog(signature)
|
|
|
|
let
|
|
signedBlock = ForkedSignedBlindedBeaconBlock.init(preparedBlock.data,
|
|
preparedBlock.blockRoot, signature)
|
|
res =
|
|
try:
|
|
debug "Sending blinded block"
|
|
await vc.publishBlindedBlock(signedBlock, ApiStrategyKind.First)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to publish blinded block",
|
|
reason = exc.getFailureReason()
|
|
return
|
|
except CancelledError as exc:
|
|
debug "Blinded block publication has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while publishing blinded block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return
|
|
|
|
if res:
|
|
let delay = vc.getDelay(slot.block_deadline())
|
|
beacon_blocks_sent.inc()
|
|
beacon_blocks_sent_delay.observe(delay.toFloatSeconds())
|
|
notice "Blinded block published", delay = delay
|
|
else:
|
|
warn "Blinded block was not accepted by beacon node"
|
|
else:
|
|
warn "Slashing protection activated for blinded block proposal"
|
|
else:
|
|
let
|
|
preparedBlock = beaconBlocks.normalBlock.get()
|
|
signingRoot = compute_block_signing_root(fork, genesisRoot, slot,
|
|
preparedBlock.blockRoot)
|
|
notSlashable = vc.attachedValidators[]
|
|
.slashingProtection
|
|
.registerBlock(vindex, validator.pubkey, slot, signingRoot)
|
|
|
|
logScope:
|
|
blck = shortLog(preparedBlock.data)
|
|
block_root = shortLog(preparedBlock.blockRoot)
|
|
signing_root = shortLog(signingRoot)
|
|
|
|
if notSlashable.isOk():
|
|
let
|
|
signature =
|
|
try:
|
|
let res = await validator.getBlockSignature(fork,
|
|
genesisRoot, slot,
|
|
preparedBlock.blockRoot,
|
|
preparedBlock.data)
|
|
if res.isErr():
|
|
warn "Unable to sign block proposal using remote signer",
|
|
reason = res.error()
|
|
return
|
|
res.get()
|
|
except CancelledError as exc:
|
|
debug "Block signature process has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while signing block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return
|
|
|
|
signedBlockContents =
|
|
case preparedBlock.data.kind
|
|
of ConsensusFork.Phase0:
|
|
RestPublishedSignedBlockContents(kind: ConsensusFork.Phase0,
|
|
phase0Data: phase0.SignedBeaconBlock(
|
|
message: preparedBlock.data.phase0Data,
|
|
root: preparedBlock.blockRoot,
|
|
signature: signature))
|
|
of ConsensusFork.Altair:
|
|
RestPublishedSignedBlockContents(kind: ConsensusFork.Altair,
|
|
altairData: altair.SignedBeaconBlock(
|
|
message: preparedBlock.data.altairData,
|
|
root: preparedBlock.blockRoot,
|
|
signature: signature))
|
|
of ConsensusFork.Bellatrix:
|
|
RestPublishedSignedBlockContents(kind: ConsensusFork.Bellatrix,
|
|
bellatrixData: bellatrix.SignedBeaconBlock(
|
|
message: preparedBlock.data.bellatrixData,
|
|
root: preparedBlock.blockRoot,
|
|
signature: signature))
|
|
of ConsensusFork.Capella:
|
|
RestPublishedSignedBlockContents(kind: ConsensusFork.Capella,
|
|
capellaData: capella.SignedBeaconBlock(
|
|
message: preparedBlock.data.capellaData,
|
|
root: preparedBlock.blockRoot,
|
|
signature: signature))
|
|
of ConsensusFork.Deneb:
|
|
RestPublishedSignedBlockContents(kind: ConsensusFork.Deneb,
|
|
denebData: DenebSignedBlockContents(
|
|
signed_block: deneb.SignedBeaconBlock(
|
|
message: preparedBlock.data.denebData,
|
|
root: preparedBlock.blockRoot,
|
|
signature: signature),
|
|
kzg_proofs: preparedBlock.kzgProofsOpt.get,
|
|
blobs: preparedBlock.blobsOpt.get))
|
|
of ConsensusFork.Electra:
|
|
debugRaiseAssert ""
|
|
default(RestPublishedSignedBlockContents)
|
|
|
|
res =
|
|
try:
|
|
debug "Sending block"
|
|
await vc.publishBlock(signedBlockContents, ApiStrategyKind.First)
|
|
except ValidatorApiError as exc:
|
|
warn "Unable to publish block", reason = exc.getFailureReason()
|
|
return
|
|
except CancelledError as exc:
|
|
debug "Block publication has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while publishing block",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return
|
|
|
|
if res:
|
|
let delay = vc.getDelay(slot.block_deadline())
|
|
beacon_blocks_sent.inc()
|
|
beacon_blocks_sent_delay.observe(delay.toFloatSeconds())
|
|
notice "Block published", delay = delay
|
|
else:
|
|
warn "Block was not accepted by beacon node"
|
|
else:
|
|
warn "Slashing protection activated for block proposal"
|
|
|
|
proc publishBlock(vc: ValidatorClientRef, currentSlot, slot: Slot,
|
|
validator: AttachedValidator) {.async.} =
|
|
let
|
|
genesisRoot = vc.beaconGenesis.genesis_validators_root
|
|
graffiti = vc.getGraffitiBytes(validator)
|
|
fork = vc.forkAtEpoch(slot.epoch)
|
|
vindex = validator.index.get()
|
|
|
|
logScope:
|
|
validator = validatorLog(validator)
|
|
validator_index = vindex
|
|
slot = slot
|
|
wall_slot = currentSlot
|
|
|
|
debug "Publishing block", delay = vc.getDelay(slot.block_deadline()),
|
|
genesis_root = genesisRoot,
|
|
graffiti = graffiti, fork = fork
|
|
let
|
|
randaoReveal =
|
|
try:
|
|
(await validator.getEpochSignature(fork, genesisRoot,
|
|
slot.epoch())).valueOr:
|
|
warn "Unable to generate RANDAO reveal using remote signer",
|
|
reason = error
|
|
return
|
|
except CancelledError as exc:
|
|
debug "RANDAO reveal production has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
error "An unexpected error occurred while receiving RANDAO data",
|
|
error_name = exc.name, error_msg = exc.msg
|
|
return
|
|
|
|
# TODO (cheatfate): This branch should be removed as soon as `produceBlockV2`
|
|
# call will be fully deprecated.
|
|
if vc.isProduceBlockV3Supported():
|
|
# We call `V3` first, if call fails and `isProduceBlockV3Supported()`
|
|
# did not find any nodes which support `V3` we try to call `V2`.
|
|
let res =
|
|
await vc.publishBlockV3(currentSlot, slot, fork, randaoReveal, validator)
|
|
if not(res) and not(vc.isProduceBlockV3Supported()):
|
|
notice "Block production using V3 failed, trying V2"
|
|
await vc.publishBlockV2(currentSlot, slot, fork, randaoReveal, validator)
|
|
else:
|
|
await vc.publishBlockV2(currentSlot, slot, fork, randaoReveal, validator)
|
|
|
|
proc proposeBlock(vc: ValidatorClientRef, slot: Slot,
|
|
proposerKey: ValidatorPubKey) {.async.} =
|
|
let
|
|
currentSlot = (await vc.checkedWaitForSlot(slot, ZeroTimeDiff,
|
|
false)).valueOr:
|
|
error "Unable to perform block production because of system time"
|
|
return
|
|
|
|
if currentSlot > slot:
|
|
warn "Skip block production for expired slot",
|
|
current_slot = currentSlot, duties_slot = slot
|
|
return
|
|
|
|
let validator = vc.getValidatorForDuties(proposerKey, slot).valueOr: return
|
|
|
|
try:
|
|
await vc.publishBlock(currentSlot, slot, validator)
|
|
except CancelledError as exc:
|
|
debug "Block proposing process was interrupted",
|
|
slot = slot, validator = validatorLog(validator)
|
|
raise exc
|
|
except CatchableError:
|
|
error "Unexpected error encountered while proposing block",
|
|
slot = slot, validator = validatorLog(validator)
|
|
|
|
proc contains(data: openArray[RestProposerDuty], task: ProposerTask): bool =
|
|
for item in data:
|
|
if (item.pubkey == task.duty.pubkey) and (item.slot == task.duty.slot):
|
|
return true
|
|
false
|
|
|
|
proc contains(data: openArray[ProposerTask], duty: RestProposerDuty): bool =
|
|
for item in data:
|
|
if (item.duty.pubkey == duty.pubkey) and (item.duty.slot == duty.slot):
|
|
return true
|
|
false
|
|
|
|
proc checkDuty(duty: RestProposerDuty, epoch: Epoch, slot: Slot): bool =
|
|
let lastSlot = start_slot(epoch + 1'u64)
|
|
if duty.slot >= slot:
|
|
if duty.slot < lastSlot:
|
|
true
|
|
else:
|
|
warn "Block proposal duty is in the far future, ignoring",
|
|
duty_slot = duty.slot, pubkey = shortLog(duty.pubkey),
|
|
wall_slot = slot, last_slot_in_epoch = (lastSlot - 1'u64)
|
|
false
|
|
else:
|
|
warn "Block proposal duty is in the past, ignoring", duty_slot = duty.slot,
|
|
pubkey = shortLog(duty.pubkey), wall_slot = slot
|
|
false
|
|
|
|
proc addOrReplaceProposers*(vc: ValidatorClientRef, epoch: Epoch,
|
|
dependentRoot: Eth2Digest,
|
|
duties: openArray[RestProposerDuty]) =
|
|
let
|
|
default = ProposedData(epoch: FAR_FUTURE_EPOCH)
|
|
currentSlot = vc.getCurrentSlot().get(Slot(0))
|
|
epochDuties = vc.proposers.getOrDefault(epoch, default)
|
|
|
|
if not(epochDuties.isDefault()):
|
|
if epochDuties.dependentRoot != dependentRoot:
|
|
warn "Proposer duties re-organization", duties_count = len(duties),
|
|
wall_slot = currentSlot, epoch = epoch,
|
|
prior_dependent_root = epochDuties.dependentRoot,
|
|
dependent_root = dependentRoot
|
|
let tasks =
|
|
block:
|
|
var res: seq[ProposerTask]
|
|
var hashset = initHashSet[Slot]()
|
|
|
|
for task in epochDuties.duties:
|
|
if task notin duties:
|
|
# Task is no more relevant, so cancel it.
|
|
debug "Cancelling running proposal duty tasks",
|
|
slot = task.duty.slot,
|
|
pubkey = shortLog(task.duty.pubkey)
|
|
task.proposeFut.cancelSoon()
|
|
task.randaoFut.cancelSoon()
|
|
else:
|
|
# If task is already running for proper slot, we keep it alive.
|
|
debug "Keep running previous proposal duty tasks",
|
|
slot = task.duty.slot,
|
|
pubkey = shortLog(task.duty.pubkey)
|
|
res.add(task)
|
|
|
|
for duty in duties:
|
|
if duty notin res:
|
|
info "Received new proposer duty", slot = duty.slot,
|
|
pubkey = shortLog(duty.pubkey)
|
|
if checkDuty(duty, epoch, currentSlot):
|
|
let task = vc.spawnProposalTask(duty)
|
|
if duty.slot in hashset:
|
|
error "Multiple block proposers for this slot, " &
|
|
"producing blocks for all proposers", slot = duty.slot
|
|
else:
|
|
hashset.incl(duty.slot)
|
|
res.add(task)
|
|
res
|
|
vc.proposers[epoch] = ProposedData.init(epoch, dependentRoot, tasks)
|
|
else:
|
|
debug "New block proposal duties received",
|
|
dependent_root = dependentRoot, duties_count = len(duties),
|
|
wall_slot = currentSlot, epoch = epoch
|
|
# Spawn new proposer tasks and modify proposers map.
|
|
let tasks =
|
|
block:
|
|
var hashset = initHashSet[Slot]()
|
|
var res: seq[ProposerTask]
|
|
for duty in duties:
|
|
info "Received new proposer duty", slot = duty.slot,
|
|
pubkey = shortLog(duty.pubkey)
|
|
if checkDuty(duty, epoch, currentSlot):
|
|
let task = vc.spawnProposalTask(duty)
|
|
if duty.slot in hashset:
|
|
error "Multiple block proposers for this slot, " &
|
|
"producing blocks for all proposers", slot = duty.slot
|
|
else:
|
|
hashset.incl(duty.slot)
|
|
res.add(task)
|
|
res
|
|
vc.proposers[epoch] = ProposedData.init(epoch, dependentRoot, tasks)
|
|
|
|
proc pollForEvents(service: BlockServiceRef, node: BeaconNodeServerRef,
|
|
response: RestHttpResponseRef) {.async.} =
|
|
let vc = service.client
|
|
|
|
logScope:
|
|
node = node
|
|
|
|
while true:
|
|
let events =
|
|
try:
|
|
await response.getServerSentEvents()
|
|
except RestError as exc:
|
|
debug "Unable to receive server-sent event", reason = $exc.msg
|
|
return
|
|
except CancelledError as exc:
|
|
raise exc
|
|
except CatchableError as exc:
|
|
warn "Got an unexpected error, " &
|
|
"while reading server-sent event stream", reason = $exc.msg
|
|
return
|
|
|
|
for event in events:
|
|
case event.name
|
|
of "data":
|
|
let blck = EventBeaconBlockObject.decodeString(event.data).valueOr:
|
|
debug "Got invalid block event format", reason = error
|
|
return
|
|
vc.registerBlock(blck, node)
|
|
of "event":
|
|
if event.data != "block":
|
|
debug "Got unexpected event name field", event_name = event.name,
|
|
event_data = event.data
|
|
else:
|
|
debug "Got some unexpected event field", event_name = event.name
|
|
|
|
if len(events) == 0:
|
|
break
|
|
|
|
proc runBlockEventMonitor(service: BlockServiceRef,
|
|
node: BeaconNodeServerRef) {.async.} =
|
|
let
|
|
vc = service.client
|
|
roles = {BeaconNodeRole.BlockProposalData}
|
|
statuses = {RestBeaconNodeStatus.Synced}
|
|
|
|
logScope:
|
|
node = node
|
|
|
|
while true:
|
|
while node.status notin statuses:
|
|
await vc.waitNodes(nil, statuses, roles, true)
|
|
|
|
let response =
|
|
block:
|
|
var resp: HttpClientResponseRef
|
|
try:
|
|
resp = await node.client.subscribeEventStream({EventTopic.Block})
|
|
if resp.status == 200:
|
|
Opt.some(resp)
|
|
else:
|
|
let body = await resp.getBodyBytes()
|
|
await resp.closeWait()
|
|
let
|
|
plain = RestPlainResponse(status: resp.status,
|
|
contentType: resp.contentType, data: body)
|
|
reason = plain.getErrorMessage()
|
|
debug "Unable to to obtain events stream", code = resp.status,
|
|
reason = reason
|
|
Opt.none(HttpClientResponseRef)
|
|
except RestError as exc:
|
|
if not(isNil(resp)): await resp.closeWait()
|
|
debug "Unable to obtain events stream", reason = $exc.msg
|
|
Opt.none(HttpClientResponseRef)
|
|
except CancelledError as exc:
|
|
if not(isNil(resp)): await resp.closeWait()
|
|
debug "Block monitoring loop has been interrupted"
|
|
raise exc
|
|
except CatchableError as exc:
|
|
if not(isNil(resp)): await resp.closeWait()
|
|
warn "Got an unexpected error while trying to establish event stream",
|
|
reason = $exc.msg
|
|
Opt.none(HttpClientResponseRef)
|
|
|
|
if response.isSome():
|
|
debug "Block monitoring connection has been established"
|
|
try:
|
|
await service.pollForEvents(node, response.get())
|
|
except CancelledError as exc:
|
|
raise exc
|
|
except CatchableError as exc:
|
|
warn "Got an unexpected error while receiving block events",
|
|
reason = $exc.msg
|
|
finally:
|
|
debug "Block monitoring connection has been lost"
|
|
await response.get().closeWait()
|
|
|
|
proc pollForBlockHeaders(service: BlockServiceRef, node: BeaconNodeServerRef,
|
|
slot: Slot, waitTime: Duration,
|
|
index: int): Future[bool] {.async.} =
|
|
let vc = service.client
|
|
|
|
logScope:
|
|
node = node
|
|
slot = slot
|
|
wait_time = waitTime
|
|
schedule_index = index
|
|
|
|
trace "Polling for block header"
|
|
|
|
let bres =
|
|
try:
|
|
await sleepAsync(waitTime)
|
|
await node.client.getBlockHeader(BlockIdent.init(slot))
|
|
except RestError as exc:
|
|
debug "Unable to obtain block header",
|
|
reason = $exc.msg, error = $exc.name
|
|
return false
|
|
except RestResponseError as exc:
|
|
debug "Got an error while trying to obtain block header",
|
|
reason = exc.message, status = exc.status
|
|
return false
|
|
except CancelledError as exc:
|
|
raise exc
|
|
except CatchableError as exc:
|
|
warn "Unexpected error encountered while receiving block header",
|
|
reason = $exc.msg, error = $exc.name
|
|
return false
|
|
|
|
if bres.isNone():
|
|
trace "Beacon node does not yet have block"
|
|
return false
|
|
|
|
let blockHeader = bres.get()
|
|
|
|
let eventBlock = EventBeaconBlockObject(
|
|
slot: blockHeader.data.header.message.slot,
|
|
block_root: blockHeader.data.root,
|
|
optimistic: blockHeader.execution_optimistic
|
|
)
|
|
vc.registerBlock(eventBlock, node)
|
|
return true
|
|
|
|
proc runBlockPollMonitor(service: BlockServiceRef,
|
|
node: BeaconNodeServerRef) {.async.} =
|
|
let
|
|
vc = service.client
|
|
roles = {BeaconNodeRole.BlockProposalData}
|
|
statuses = {RestBeaconNodeStatus.Synced}
|
|
|
|
logScope:
|
|
node = node
|
|
|
|
while true:
|
|
let currentSlot =
|
|
block:
|
|
let res = await vc.checkedWaitForNextSlot(ZeroTimeDiff, false)
|
|
if res.isNone(): continue
|
|
res.geT()
|
|
|
|
while node.status notin statuses:
|
|
await vc.waitNodes(nil, statuses, roles, true)
|
|
|
|
let
|
|
currentTime = vc.beaconClock.now()
|
|
afterSlot = currentTime.slotOrZero()
|
|
|
|
if currentTime > afterSlot.attestation_deadline():
|
|
# Attestation time already, lets wait for next slot.
|
|
continue
|
|
|
|
let
|
|
pollTime1 = afterSlot.start_beacon_time() + BlockPollOffset1
|
|
pollTime2 = afterSlot.start_beacon_time() + BlockPollOffset2
|
|
pollTime3 = afterSlot.start_beacon_time() + BlockPollOffset3
|
|
|
|
var pendingTasks =
|
|
block:
|
|
var res: seq[FutureBase]
|
|
if currentTime <= pollTime1:
|
|
let stime = nanoseconds((pollTime1 - currentTime).nanoseconds)
|
|
res.add(FutureBase(
|
|
service.pollForBlockHeaders(node, afterSlot, stime, 0)))
|
|
if currentTime <= pollTime2:
|
|
let stime = nanoseconds((pollTime2 - currentTime).nanoseconds)
|
|
res.add(FutureBase(
|
|
service.pollForBlockHeaders(node, afterSlot, stime, 1)))
|
|
if currentTime <= pollTime3:
|
|
let stime = nanoseconds((pollTime3 - currentTime).nanoseconds)
|
|
res.add(FutureBase(
|
|
service.pollForBlockHeaders(node, afterSlot, stime, 2)))
|
|
res
|
|
try:
|
|
while true:
|
|
let completedFuture = await race(pendingTasks)
|
|
let blockReceived =
|
|
block:
|
|
var res = false
|
|
for future in pendingTasks:
|
|
if not(future.completed()): continue
|
|
if not(cast[Future[bool]](future).read()): continue
|
|
res = true
|
|
break
|
|
res
|
|
if blockReceived:
|
|
let pending =
|
|
pendingTasks.filterIt(not(it.finished())).mapIt(it.cancelAndWait())
|
|
# We use `noCancel` here because its cleanup and we have `break`
|
|
# after it.
|
|
await noCancel allFutures(pending)
|
|
break
|
|
pendingTasks.keepItIf(it != completedFuture)
|
|
if len(pendingTasks) == 0: break
|
|
except CancelledError as exc:
|
|
let pending =
|
|
pendingTasks.filterIt(not(it.finished())).mapIt(it.cancelAndWait())
|
|
await noCancel allFutures(pending)
|
|
raise exc
|
|
except CatchableError as exc:
|
|
warn "An unexpected error occurred while running block monitoring",
|
|
reason = $exc.msg, error = $exc.name
|
|
|
|
proc runBlockMonitor(service: BlockServiceRef) {.async.} =
|
|
let
|
|
vc = service.client
|
|
blockNodes = vc.filterNodes(ResolvedBeaconNodeStatuses,
|
|
{BeaconNodeRole.BlockProposalData})
|
|
let pendingTasks =
|
|
case vc.config.monitoringType
|
|
of BlockMonitoringType.Disabled:
|
|
debug "Block monitoring disabled"
|
|
@[newFuture[void]("block.monitor.disabled")]
|
|
of BlockMonitoringType.Poll:
|
|
var res: seq[Future[void]]
|
|
for node in blockNodes:
|
|
res.add(service.runBlockPollMonitor(node))
|
|
res
|
|
of BlockMonitoringType.Event:
|
|
var res: seq[Future[void]]
|
|
for node in blockNodes:
|
|
res.add(service.runBlockEventMonitor(node))
|
|
res
|
|
|
|
try:
|
|
await allFutures(pendingTasks)
|
|
except CancelledError as exc:
|
|
let pending =
|
|
pendingTasks.filterIt(not(it.finished())).mapIt(it.cancelAndWait())
|
|
await noCancel allFutures(pending)
|
|
raise exc
|
|
except CatchableError as exc:
|
|
warn "An unexpected error occurred while running block monitoring",
|
|
reason = $exc.msg, error = $exc.name
|
|
return
|
|
|
|
proc mainLoop(service: BlockServiceRef) {.async.} =
|
|
let vc = service.client
|
|
service.state = ServiceState.Running
|
|
debug "Service started"
|
|
let future = service.runBlockMonitor()
|
|
try:
|
|
# Future is not going to be completed, so the only way to exit, is to
|
|
# cancel it.
|
|
await future
|
|
except CancelledError:
|
|
debug "Service interrupted"
|
|
except CatchableError as exc:
|
|
error "Service crashed with unexpected error", err_name = exc.name,
|
|
err_msg = exc.msg
|
|
|
|
# We going to cleanup all the pending proposer tasks.
|
|
var res: seq[FutureBase]
|
|
for epoch, data in vc.proposers.pairs():
|
|
for duty in data.duties.items():
|
|
if not(duty.proposeFut.finished()):
|
|
res.add(duty.proposeFut.cancelAndWait())
|
|
if not(duty.randaoFut.finished()):
|
|
res.add(duty.randaoFut.cancelAndWait())
|
|
await noCancel allFutures(res)
|
|
|
|
proc init*(t: typedesc[BlockServiceRef],
|
|
vc: ValidatorClientRef): Future[BlockServiceRef] {.async.} =
|
|
logScope: service = ServiceName
|
|
var res = BlockServiceRef(name: ServiceName, client: vc,
|
|
state: ServiceState.Initialized)
|
|
debug "Initializing service"
|
|
return res
|
|
|
|
proc start*(service: BlockServiceRef) =
|
|
service.lifeFut = mainLoop(service)
|