2021-03-11 10:10:57 +00:00
|
|
|
# beacon_chain
|
2022-01-18 13:36:52 +00:00
|
|
|
# Copyright (c) 2018-2022 Status Research & Development GmbH
|
2021-03-11 10:10:57 +00:00
|
|
|
# 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.
|
|
|
|
|
2021-05-28 16:34:00 +00:00
|
|
|
{.push raises: [Defect].}
|
|
|
|
|
2021-03-11 10:10:57 +00:00
|
|
|
import
|
|
|
|
std/math,
|
|
|
|
stew/results,
|
|
|
|
chronicles, chronos, metrics,
|
2022-01-18 13:36:52 +00:00
|
|
|
../spec/datatypes/[phase0, altair],
|
2021-12-06 09:49:01 +00:00
|
|
|
../spec/[forks, signatures_batch],
|
|
|
|
../consensus_object_pools/[
|
|
|
|
attestation_pool, block_clearance, blockchain_dag, block_quarantine,
|
|
|
|
spec_cache],
|
2021-03-11 10:10:57 +00:00
|
|
|
./consensus_manager,
|
2021-10-19 14:09:26 +00:00
|
|
|
".."/[beacon_clock],
|
2021-08-18 18:57:58 +00:00
|
|
|
../sszdump
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2021-12-06 09:49:01 +00:00
|
|
|
export sszdump, signatures_batch
|
2021-07-15 19:01:07 +00:00
|
|
|
|
2021-05-28 16:34:00 +00:00
|
|
|
# Block Processor
|
2021-03-11 10:10:57 +00:00
|
|
|
# ------------------------------------------------------------------------------
|
2021-05-28 16:34:00 +00:00
|
|
|
# The block processor moves blocks from "Incoming" to "Consensus verified"
|
2021-03-11 10:10:57 +00:00
|
|
|
|
|
|
|
declareHistogram beacon_store_block_duration_seconds,
|
|
|
|
"storeBlock() duration", buckets = [0.25, 0.5, 1, 2, 4, 8, Inf]
|
|
|
|
|
|
|
|
type
|
|
|
|
BlockEntry* = object
|
2021-07-15 19:01:07 +00:00
|
|
|
blck*: ForkedSignedBeaconBlock
|
2021-05-28 16:34:00 +00:00
|
|
|
resfut*: Future[Result[void, BlockError]]
|
|
|
|
queueTick*: Moment # Moment when block was enqueued
|
|
|
|
validationDur*: Duration # Time it took to perform gossip validation
|
2021-12-20 19:20:31 +00:00
|
|
|
src*: MsgSource
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2021-05-28 16:34:00 +00:00
|
|
|
BlockProcessor* = object
|
|
|
|
## This manages the processing of blocks from different sources
|
2021-03-11 10:10:57 +00:00
|
|
|
## Blocks and attestations are enqueued in a gossip-validated state
|
|
|
|
##
|
|
|
|
## from:
|
|
|
|
## - Gossip (when synced)
|
|
|
|
## - SyncManager (during sync)
|
|
|
|
## - RequestManager (missing ancestor blocks)
|
|
|
|
##
|
|
|
|
## are then consensus-verified and added to:
|
|
|
|
## - the blockchain DAG
|
|
|
|
## - database
|
|
|
|
## - attestation pool
|
|
|
|
## - fork choice
|
2021-12-06 09:49:01 +00:00
|
|
|
##
|
|
|
|
## The processor will also reinsert blocks from the quarantine, should a
|
|
|
|
## parent be found.
|
2021-03-11 10:10:57 +00:00
|
|
|
|
|
|
|
# Config
|
|
|
|
# ----------------------------------------------------------------
|
|
|
|
dumpEnabled: bool
|
|
|
|
dumpDirInvalid: string
|
|
|
|
dumpDirIncoming: string
|
|
|
|
|
|
|
|
# Producers
|
|
|
|
# ----------------------------------------------------------------
|
2022-01-26 12:20:08 +00:00
|
|
|
blockQueue: AsyncQueue[BlockEntry]
|
2021-03-11 10:10:57 +00:00
|
|
|
|
|
|
|
# Consumer
|
|
|
|
# ----------------------------------------------------------------
|
|
|
|
consensusManager: ref ConsensusManager
|
|
|
|
## Blockchain DAG, AttestationPool and Quarantine
|
2022-01-26 12:20:08 +00:00
|
|
|
validatorMonitor: ref ValidatorMonitor
|
2021-08-19 10:45:31 +00:00
|
|
|
getBeaconTime: GetBeaconTimeFn
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2021-12-06 09:49:01 +00:00
|
|
|
verifier: BatchVerifier
|
|
|
|
|
2021-03-11 10:10:57 +00:00
|
|
|
# Initialization
|
|
|
|
# ------------------------------------------------------------------------------
|
|
|
|
|
2021-05-28 16:34:00 +00:00
|
|
|
proc new*(T: type BlockProcessor,
|
2021-04-06 11:59:11 +00:00
|
|
|
dumpEnabled: bool,
|
|
|
|
dumpDirInvalid, dumpDirIncoming: string,
|
2021-12-06 09:49:01 +00:00
|
|
|
rng: ref BrHmacDrbgContext, taskpool: TaskPoolPtr,
|
2021-03-11 10:10:57 +00:00
|
|
|
consensusManager: ref ConsensusManager,
|
2021-12-20 19:20:31 +00:00
|
|
|
validatorMonitor: ref ValidatorMonitor,
|
2021-08-19 10:45:31 +00:00
|
|
|
getBeaconTime: GetBeaconTimeFn): ref BlockProcessor =
|
2021-05-28 16:34:00 +00:00
|
|
|
(ref BlockProcessor)(
|
2021-04-06 11:59:11 +00:00
|
|
|
dumpEnabled: dumpEnabled,
|
|
|
|
dumpDirInvalid: dumpDirInvalid,
|
|
|
|
dumpDirIncoming: dumpDirIncoming,
|
2021-12-06 09:49:01 +00:00
|
|
|
blockQueue: newAsyncQueue[BlockEntry](),
|
2021-04-06 11:59:11 +00:00
|
|
|
consensusManager: consensusManager,
|
2021-12-20 19:20:31 +00:00
|
|
|
validatorMonitor: validatorMonitor,
|
2021-12-06 09:49:01 +00:00
|
|
|
getBeaconTime: getBeaconTime,
|
|
|
|
verifier: BatchVerifier(rng: rng, taskpool: taskpool)
|
|
|
|
)
|
2021-03-11 10:10:57 +00:00
|
|
|
|
|
|
|
# Sync callbacks
|
|
|
|
# ------------------------------------------------------------------------------
|
|
|
|
|
2022-02-17 11:53:55 +00:00
|
|
|
func hasBlocks*(self: BlockProcessor): bool =
|
2021-12-06 09:49:01 +00:00
|
|
|
self.blockQueue.len() > 0
|
2021-04-26 20:39:44 +00:00
|
|
|
|
2021-03-11 10:10:57 +00:00
|
|
|
# Storage
|
|
|
|
# ------------------------------------------------------------------------------
|
|
|
|
|
2021-11-05 15:39:47 +00:00
|
|
|
proc dumpInvalidBlock*(
|
2021-12-06 09:49:01 +00:00
|
|
|
self: BlockProcessor, signedBlock: ForkySignedBeaconBlock) =
|
2021-11-05 15:39:47 +00:00
|
|
|
if self.dumpEnabled:
|
|
|
|
dump(self.dumpDirInvalid, signedBlock)
|
|
|
|
|
2022-01-26 12:20:08 +00:00
|
|
|
proc dumpBlock[T](
|
2021-07-15 19:01:07 +00:00
|
|
|
self: BlockProcessor,
|
2021-11-05 07:34:34 +00:00
|
|
|
signedBlock: ForkySignedBeaconBlock,
|
2021-12-06 09:49:01 +00:00
|
|
|
res: Result[T, BlockError]) =
|
2021-03-11 10:10:57 +00:00
|
|
|
if self.dumpEnabled and res.isErr:
|
2021-12-06 09:49:01 +00:00
|
|
|
case res.error
|
|
|
|
of BlockError.Invalid:
|
2021-11-05 15:39:47 +00:00
|
|
|
self.dumpInvalidBlock(signedBlock)
|
2021-12-06 09:49:01 +00:00
|
|
|
of BlockError.MissingParent:
|
2021-11-05 15:39:47 +00:00
|
|
|
dump(self.dumpDirIncoming, signedBlock)
|
2021-03-11 10:10:57 +00:00
|
|
|
else:
|
|
|
|
discard
|
|
|
|
|
2022-01-26 12:20:08 +00:00
|
|
|
proc storeBackfillBlock(
|
|
|
|
self: var BlockProcessor,
|
|
|
|
signedBlock: ForkySignedBeaconBlock): Result[void, BlockError] =
|
|
|
|
|
|
|
|
# The block is certainly not missing any more
|
|
|
|
self.consensusManager.quarantine[].missing.del(signedBlock.root)
|
|
|
|
|
|
|
|
let res = self.consensusManager.dag.addBackfillBlock(signedBlock)
|
|
|
|
|
|
|
|
if res.isErr():
|
|
|
|
case res.error
|
|
|
|
of BlockError.MissingParent:
|
|
|
|
if signedBlock.message.parent_root in
|
|
|
|
self.consensusManager.quarantine[].unviable:
|
|
|
|
# DAG doesn't know about unviable ancestor blocks - we do! Translate
|
|
|
|
# this to the appropriate error so that sync etc doesn't retry the block
|
|
|
|
self.consensusManager.quarantine[].addUnviable(signedBlock.root)
|
|
|
|
|
|
|
|
return err(BlockError.UnviableFork)
|
|
|
|
of BlockError.UnviableFork:
|
|
|
|
# Track unviables so that descendants can be discarded properly
|
|
|
|
self.consensusManager.quarantine[].addUnviable(signedBlock.root)
|
|
|
|
else: discard
|
|
|
|
|
|
|
|
res
|
|
|
|
|
2021-10-19 15:20:55 +00:00
|
|
|
proc storeBlock*(
|
2021-07-15 19:01:07 +00:00
|
|
|
self: var BlockProcessor,
|
2021-12-20 19:20:31 +00:00
|
|
|
src: MsgSource, wallTime: BeaconTime,
|
|
|
|
signedBlock: ForkySignedBeaconBlock, queueTick: Moment = Moment.now(),
|
2021-12-02 18:34:12 +00:00
|
|
|
validationDur = Duration()): Result[BlockRef, BlockError] =
|
2021-12-20 19:20:31 +00:00
|
|
|
## storeBlock is the main entry point for unvalidated blocks - all untrusted
|
|
|
|
## blocks, regardless of origin, pass through here. When storing a block,
|
|
|
|
## we will add it to the dag and pass it to all block consumers that need
|
|
|
|
## to know about it, such as the fork choice and the monitoring
|
2021-03-11 10:10:57 +00:00
|
|
|
let
|
|
|
|
attestationPool = self.consensusManager.attestationPool
|
2021-12-02 18:34:12 +00:00
|
|
|
startTick = Moment.now()
|
2021-12-20 19:20:31 +00:00
|
|
|
wallSlot = wallTime.slotOrZero()
|
|
|
|
vm = self.validatorMonitor
|
2021-12-06 09:49:01 +00:00
|
|
|
dag = self.consensusManager.dag
|
|
|
|
|
|
|
|
# The block is certainly not missing any more
|
|
|
|
self.consensusManager.quarantine[].missing.del(signedBlock.root)
|
|
|
|
|
|
|
|
# We'll also remove the block as an orphan: it's unlikely the parent is
|
|
|
|
# missing if we get this far - should that be the case, the block will
|
|
|
|
# be re-added later
|
|
|
|
self.consensusManager.quarantine[].removeOrphan(signedBlock)
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2021-07-15 19:01:07 +00:00
|
|
|
type Trusted = typeof signedBlock.asTrusted()
|
Backfill support for ChainDAG (#3171)
In the ChainDAG, 3 block pointers are kept: genesis, tail and head. This
PR adds one more block pointer: the backfill block which represents the
block that has been backfilled so far.
When doing a checkpoint sync, a random block is given as starting point
- this is the tail block, and we require that the tail block has a
corresponding state.
When backfilling, we end up with blocks without corresponding states,
hence we cannot use `tail` as a backfill pointer - there is no state.
Nonetheless, we need to keep track of where we are in the backfill
process between restarts, such that we can answer GetBeaconBlocksByRange
requests.
This PR adds the basic support for backfill handling - it needs to be
integrated with backfill sync, and the REST API needs to be adjusted to
take advantage of the new backfilled blocks when responding to certain
requests.
Future work will also enable moving the tail in either direction:
* pruning means moving the tail forward in time and removing states
* backwards means recreating past states from genesis, such that
intermediate states are recreated step by step all the way to the tail -
at that point, tail, genesis and backfill will match up.
* backfilling is done when backfill != genesis - later, this will be the
WSS checkpoint instead
2021-12-13 13:36:06 +00:00
|
|
|
let blck = dag.addHeadBlock(self.verifier, signedBlock) do (
|
2021-07-15 19:01:07 +00:00
|
|
|
blckRef: BlockRef, trustedBlock: Trusted, epochRef: EpochRef):
|
2021-03-11 10:10:57 +00:00
|
|
|
# Callback add to fork choice if valid
|
|
|
|
attestationPool[].addForkChoice(
|
2021-12-21 18:56:08 +00:00
|
|
|
epochRef, blckRef, trustedBlock.message, wallTime)
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2021-12-20 19:20:31 +00:00
|
|
|
vm[].registerBeaconBlock(
|
|
|
|
src, wallTime, trustedBlock.message)
|
|
|
|
|
|
|
|
for attestation in trustedBlock.message.body.attestations:
|
2022-01-08 23:28:49 +00:00
|
|
|
for validator_index in get_attesting_indices(
|
|
|
|
epochRef, attestation.data.slot,
|
|
|
|
CommitteeIndex.init(attestation.data.index).expect(
|
|
|
|
"index has been checked"),
|
|
|
|
attestation.aggregation_bits):
|
|
|
|
vm[].registerAttestationInBlock(attestation.data, validator_index,
|
2021-12-20 19:20:31 +00:00
|
|
|
trustedBlock.message)
|
|
|
|
|
2022-03-16 07:20:40 +00:00
|
|
|
withState(dag[].clearanceState):
|
2021-12-20 19:20:31 +00:00
|
|
|
when stateFork >= BeaconStateFork.Altair and
|
|
|
|
Trusted isnot phase0.TrustedSignedBeaconBlock: # altair+
|
|
|
|
for i in trustedBlock.message.body.sync_aggregate.sync_committee_bits.oneIndices():
|
|
|
|
vm[].registerSyncAggregateInBlock(
|
|
|
|
trustedBlock.message.slot, trustedBlock.root,
|
|
|
|
state.data.current_sync_committee.pubkeys.data[i])
|
|
|
|
|
2021-03-11 10:10:57 +00:00
|
|
|
self.dumpBlock(signedBlock, blck)
|
|
|
|
|
|
|
|
# There can be a scenario where we receive a block we already received.
|
|
|
|
# However this block was before the last finalized epoch and so its parent
|
|
|
|
# was pruned from the ForkChoice.
|
2021-12-06 09:49:01 +00:00
|
|
|
if blck.isErr():
|
2022-01-26 12:20:08 +00:00
|
|
|
case blck.error()
|
|
|
|
of BlockError.MissingParent:
|
|
|
|
if signedBlock.message.parent_root in
|
|
|
|
self.consensusManager.quarantine[].unviable:
|
|
|
|
# DAG doesn't know about unviable ancestor blocks - we do! Translate
|
|
|
|
# this to the appropriate error so that sync etc doesn't retry the block
|
|
|
|
self.consensusManager.quarantine[].addUnviable(signedBlock.root)
|
|
|
|
|
|
|
|
return err(BlockError.UnviableFork)
|
|
|
|
|
|
|
|
if not self.consensusManager.quarantine[].addOrphan(
|
|
|
|
dag.finalizedHead.slot, ForkedSignedBeaconBlock.init(signedBlock)):
|
2021-12-06 09:49:01 +00:00
|
|
|
debug "Block quarantine full",
|
|
|
|
blockRoot = shortLog(signedBlock.root),
|
|
|
|
blck = shortLog(signedBlock.message),
|
|
|
|
signature = shortLog(signedBlock.signature)
|
2022-01-26 12:20:08 +00:00
|
|
|
of BlockError.UnviableFork:
|
|
|
|
# Track unviables so that descendants can be discarded properly
|
|
|
|
self.consensusManager.quarantine[].addUnviable(signedBlock.root)
|
|
|
|
else: discard
|
2021-12-06 09:49:01 +00:00
|
|
|
|
|
|
|
return blck
|
2021-12-02 18:34:12 +00:00
|
|
|
|
|
|
|
let storeBlockTick = Moment.now()
|
|
|
|
|
|
|
|
# Eagerly update head: the incoming block "should" get selected
|
2021-12-21 18:56:08 +00:00
|
|
|
self.consensusManager[].updateHead(wallTime.slotOrZero)
|
2021-12-02 18:34:12 +00:00
|
|
|
|
|
|
|
let
|
|
|
|
updateHeadTick = Moment.now()
|
|
|
|
queueDur = startTick - queueTick
|
|
|
|
storeBlockDur = storeBlockTick - startTick
|
|
|
|
updateHeadDur = updateHeadTick - storeBlockTick
|
|
|
|
|
|
|
|
beacon_store_block_duration_seconds.observe(storeBlockDur.toFloatSeconds())
|
|
|
|
|
|
|
|
debug "Block processed",
|
|
|
|
localHeadSlot = self.consensusManager.dag.head.slot,
|
|
|
|
blockSlot = blck.get().slot,
|
|
|
|
validationDur, queueDur, storeBlockDur, updateHeadDur
|
|
|
|
|
2021-12-06 09:49:01 +00:00
|
|
|
for quarantined in self.consensusManager.quarantine[].pop(blck.get().root):
|
|
|
|
# Process the blocks that had the newly accepted block as parent
|
2021-12-20 19:20:31 +00:00
|
|
|
self.addBlock(MsgSource.gossip, quarantined)
|
2021-12-06 09:49:01 +00:00
|
|
|
|
|
|
|
blck
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2022-01-26 12:20:08 +00:00
|
|
|
# Enqueue
|
|
|
|
# ------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
proc addBlock*(
|
|
|
|
self: var BlockProcessor, src: MsgSource, blck: ForkedSignedBeaconBlock,
|
|
|
|
resfut: Future[Result[void, BlockError]] = nil,
|
|
|
|
validationDur = Duration()) =
|
|
|
|
## Enqueue a Gossip-validated block for consensus verification
|
|
|
|
# Backpressure:
|
|
|
|
# There is no backpressure here - producers must wait for `resfut` to
|
|
|
|
# constrain their own processing
|
|
|
|
# Producers:
|
|
|
|
# - Gossip (when synced)
|
|
|
|
# - SyncManager (during sync)
|
|
|
|
# - RequestManager (missing ancestor blocks)
|
|
|
|
|
|
|
|
withBlck(blck):
|
|
|
|
if blck.message.slot <= self.consensusManager.dag.finalizedHead.slot:
|
|
|
|
# let backfill blocks skip the queue - these are always "fast" to process
|
|
|
|
# because there are no state rewinds to deal with
|
|
|
|
let res = self.storeBackfillBlock(blck)
|
|
|
|
|
|
|
|
if resFut != nil:
|
|
|
|
resFut.complete(res)
|
|
|
|
return
|
|
|
|
|
|
|
|
try:
|
|
|
|
self.blockQueue.addLastNoWait(BlockEntry(
|
|
|
|
blck: blck,
|
|
|
|
resfut: resfut, queueTick: Moment.now(),
|
|
|
|
validationDur: validationDur,
|
|
|
|
src: src))
|
|
|
|
except AsyncQueueFullError:
|
|
|
|
raiseAssert "unbounded queue"
|
|
|
|
|
2021-03-11 10:10:57 +00:00
|
|
|
# Event Loop
|
|
|
|
# ------------------------------------------------------------------------------
|
|
|
|
|
2021-05-28 16:34:00 +00:00
|
|
|
proc processBlock(self: var BlockProcessor, entry: BlockEntry) =
|
2021-03-11 10:10:57 +00:00
|
|
|
logScope:
|
2021-05-28 16:34:00 +00:00
|
|
|
blockRoot = shortLog(entry.blck.root)
|
2021-03-11 10:10:57 +00:00
|
|
|
|
|
|
|
let
|
2021-08-19 10:45:31 +00:00
|
|
|
wallTime = self.getBeaconTime()
|
2021-03-11 10:10:57 +00:00
|
|
|
(afterGenesis, wallSlot) = wallTime.toSlot()
|
|
|
|
|
|
|
|
if not afterGenesis:
|
|
|
|
error "Processing block before genesis, clock turned back?"
|
|
|
|
quit 1
|
|
|
|
|
|
|
|
let
|
2021-12-02 18:34:12 +00:00
|
|
|
res = withBlck(entry.blck):
|
2021-12-20 19:20:31 +00:00
|
|
|
self.storeBlock(entry.src, wallTime, blck, entry.queueTick, entry.validationDur)
|
2021-05-28 16:34:00 +00:00
|
|
|
|
2021-12-16 14:57:16 +00:00
|
|
|
if entry.resfut != nil:
|
|
|
|
entry.resfut.complete(
|
|
|
|
if res.isOk(): Result[void, BlockError].ok()
|
|
|
|
else: Result[void, BlockError].err(res.error()))
|
2021-03-11 10:10:57 +00:00
|
|
|
|
2021-05-28 16:34:00 +00:00
|
|
|
proc runQueueProcessingLoop*(self: ref BlockProcessor) {.async.} =
|
2021-03-11 10:10:57 +00:00
|
|
|
while true:
|
2021-04-26 20:39:44 +00:00
|
|
|
# Cooperative concurrency: one block per loop iteration - because
|
2021-03-11 10:10:57 +00:00
|
|
|
# we run both networking and CPU-heavy things like block processing
|
|
|
|
# on the same thread, we need to make sure that there is steady progress
|
|
|
|
# on the networking side or we get long lockups that lead to timeouts.
|
|
|
|
const
|
|
|
|
# We cap waiting for an idle slot in case there's a lot of network traffic
|
|
|
|
# taking up all CPU - we don't want to _completely_ stop processing blocks
|
2021-04-26 20:39:44 +00:00
|
|
|
# in this case - doing so also allows us to benefit from more batching /
|
|
|
|
# larger network reads when under load.
|
2021-03-11 10:10:57 +00:00
|
|
|
idleTimeout = 10.milliseconds
|
|
|
|
|
|
|
|
discard await idleAsync().withTimeout(idleTimeout)
|
|
|
|
|
2021-12-06 09:49:01 +00:00
|
|
|
self[].processBlock(await self[].blockQueue.popFirst())
|