Light forward sync mechanism

beacon_chain/beacon_node.nim

@@ -22,10 +22,12 @@ import
   ./el/el_manager,
   ./consensus_object_pools/[
     blockchain_dag, blob_quarantine, block_quarantine, consensus_manager,
-    attestation_pool, sync_committee_msg_pool, validator_change_pool],
+    attestation_pool, sync_committee_msg_pool, validator_change_pool,
+    blockchain_list],
   ./spec/datatypes/[base, altair],
   ./spec/eth2_apis/dynamic_fee_recipients,
-  ./sync/[sync_manager, request_manager],
+  ./spec/signatures_batch,
+  ./sync/[sync_manager, request_manager, sync_types],
   ./validators/[
     action_tracker, message_router, validator_monitor, validator_pool,
     keystore_management],
@@ -38,7 +40,7 @@ export
   eth2_network, el_manager, request_manager, sync_manager,
   eth2_processor, optimistic_processor, blockchain_dag, block_quarantine,
   base, message_router, validator_monitor, validator_pool,
-  consensus_manager, dynamic_fee_recipients
+  consensus_manager, dynamic_fee_recipients, sync_types
 
 type
   EventBus* = object
@@ -57,6 +59,7 @@ type
       RestVersioned[ForkedLightClientFinalityUpdate]]
     optUpdateQueue*: AsyncEventQueue[
       RestVersioned[ForkedLightClientOptimisticUpdate]]
+    optFinHeaderUpdateQueue*: AsyncEventQueue[ForkedLightClientHeader]
 
   BeaconNode* = ref object
     nickname*: string
@@ -71,6 +74,7 @@ type
       .Raising([CancelledError])
     lightClient*: LightClient
     dag*: ChainDAGRef
+    list*: ChainListRef
     quarantine*: ref Quarantine
     blobQuarantine*: ref BlobQuarantine
     attestationPool*: ref AttestationPool
@@ -87,8 +91,11 @@ type
     requestManager*: RequestManager
     syncManager*: SyncManager[Peer, PeerId]
     backfiller*: SyncManager[Peer, PeerId]
+    untrustedManager*: SyncManager[Peer, PeerId]
+    syncOverseer*: SyncOverseerRef
     genesisSnapshotContent*: string
     processor*: ref Eth2Processor
+    batchVerifier*: ref BatchVerifier
     blockProcessor*: ref BlockProcessor
     consensusManager*: ref ConsensusManager
     attachedValidatorBalanceTotal*: Gwei

beacon_chain/beacon_node_light_client.nim

@@ -53,10 +53,10 @@ proc initLightClient*(
       getBeaconTime, optimisticHandler)
 
     shouldInhibitSync = func(): bool =
-      if node.syncManager != nil:
-        not node.syncManager.inProgress  # No LC sync needed if DAG is in sync
-      else:
+      if isNil(node.syncOverseer):
         false
+      else:
+        not node.syncOverseer.syncInProgress # No LC sync needed if DAG is in sync
     lightClient = createLightClient(
       node.network, rng, config, cfg, forkDigests, getBeaconTime,
       genesis_validators_root, LightClientFinalizationMode.Strict,
@@ -107,7 +107,16 @@ proc initLightClient*(
             # The execution block hash is only available from Capella onward
             info "Ignoring new LC optimistic header until Capella"
 
+    proc onFinalizedHeader(
+        lightClient: LightClient,
+        finalizedHeader: ForkedLightClientHeader) =
+      if not node.consensusManager[].shouldSyncOptimistically(node.currentSlot):
+        return
+
+      node.eventBus.optFinHeaderUpdateQueue.emit(finalizedHeader)
+
     lightClient.onOptimisticHeader = onOptimisticHeader
+    lightClient.onFinalizedHeader = onFinalizedHeader
     lightClient.trustedBlockRoot = config.trustedBlockRoot
 
   elif config.trustedBlockRoot.isSome:

beacon_chain/consensus_object_pools/block_clearance.nim

@@ -8,16 +8,15 @@
 {.push raises: [].}
 
 import
+  std/sequtils,
   chronicles,
   results,
   stew/assign2,
   ../spec/[
     beaconstate, forks, signatures, signatures_batch,
     state_transition, state_transition_epoch],
-  "."/[block_dag, blockchain_dag, blockchain_dag_light_client]
-
-from ../spec/datatypes/capella import asSigVerified, asTrusted, shortLog
-from ../spec/datatypes/deneb import asSigVerified, asTrusted, shortLog
+  "."/[block_pools_types, block_dag, blockchain_dag,
+       blockchain_dag_light_client]
 
 export results, signatures_batch, block_dag, blockchain_dag
 
@@ -114,15 +113,18 @@ proc addResolvedHeadBlock(
   blockRef
 
 proc checkStateTransition(
-       dag: ChainDAGRef, signedBlock: ForkySigVerifiedSignedBeaconBlock,
-       cache: var StateCache): Result[void, VerifierError] =
+    dag: ChainDAGRef,
+    signedBlock: ForkySigVerifiedSignedBeaconBlock,
+    cache: var StateCache,
+    updateFlags: UpdateFlags,
+): Result[void, VerifierError] =
   ## Ensure block can be applied on a state
   func restore(v: var ForkedHashedBeaconState) =
     assign(dag.clearanceState, dag.headState)
 
   let res = state_transition_block(
       dag.cfg, dag.clearanceState, signedBlock,
-      cache, dag.updateFlags, restore)
+      cache, updateFlags, restore)
 
   if res.isErr():
     info "Invalid block",
@@ -150,7 +152,8 @@ proc advanceClearanceState*(dag: ChainDAGRef) =
     var
       cache = StateCache()
       info = ForkedEpochInfo()
-    dag.advanceSlots(dag.clearanceState, next, true, cache, info)
+    dag.advanceSlots(dag.clearanceState, next, true, cache, info,
+                     dag.updateFlags)
     debug "Prepared clearance state for next block",
       next, updateStateDur = Moment.now() - startTick
 
@@ -267,7 +270,7 @@ proc addHeadBlockWithParent*(
   # onto which we can apply the new block
   let clearanceBlock = BlockSlotId.init(parent.bid, signedBlock.message.slot)
   if not updateState(
-      dag, dag.clearanceState, clearanceBlock, true, cache):
+      dag, dag.clearanceState, clearanceBlock, true, cache, dag.updateFlags):
     # We should never end up here - the parent must be a block no older than and
     # rooted in the finalized checkpoint, hence we should always be able to
     # load its corresponding state
@@ -297,7 +300,8 @@ proc addHeadBlockWithParent*(
 
   let sigVerifyTick = Moment.now()
 
-  ? checkStateTransition(dag, signedBlock.asSigVerified(), cache)
+  ? checkStateTransition(dag, signedBlock.asSigVerified(), cache,
+                         dag.updateFlags)
 
   let stateVerifyTick = Moment.now()
   # Careful, clearanceState.data has been updated but not blck - we need to
@@ -449,3 +453,110 @@ proc addBackfillBlock*(
     putBlockDur = putBlockTick - sigVerifyTick
 
   ok()
+
+template BlockAdded(kind: static ConsensusFork): untyped =
+  when kind == ConsensusFork.Electra:
+    OnElectraBlockAdded
+  elif kind == ConsensusFork.Deneb:
+    OnDenebBlockAdded
+  elif kind == ConsensusFork.Capella:
+    OnCapellaBlockAdded
+  elif kind == ConsensusFork.Bellatrix:
+    OnBellatrixBlockAdded
+  elif kind == ConsensusFork.Altair:
+    OnAltairBlockAdded
+  elif kind == ConsensusFork.Phase0:
+    OnPhase0BlockAdded
+  else:
+    static: raiseAssert "Unreachable"
+
+proc verifyBlockProposer*(
+    dag: ChainDAGRef,
+    verifier: var BatchVerifier,
+    blocks: openArray[ForkedSignedBeaconBlock]
+): Result[void, string] =
+  var sigs: seq[SignatureSet]
+
+  ? sigs.collectProposerSignatureSet(
+    blocks, dag.db.immutableValidators, dag.clearanceState)
+
+  if not verifier.batchVerify(sigs):
+    err("Block batch signature verification failed")
+  else:
+    ok()
+
+proc addBackfillBlockData*(
+    dag: ChainDAGRef,
+    bdata: BlockData,
+    onStateUpdated: OnStateUpdated,
+    onBlockAdded: OnForkedBlockAdded
+): Result[void, VerifierError] =
+  var cache = StateCache()
+
+  withBlck(bdata.blck):
+    let
+      parent = checkHeadBlock(dag, forkyBlck).valueOr:
+        if error == VerifierError.Duplicate:
+          return ok()
+        return err(error)
+      startTick = Moment.now()
+      parentBlock = dag.getForkedBlock(parent.bid.root).get()
+      trustedStateRoot =
+        withBlck(parentBlock):
+          forkyBlck.message.state_root
+      clearanceBlock = BlockSlotId.init(parent.bid, forkyBlck.message.slot)
+      updateFlags1 = dag.updateFlags + {skipLastStateRootCalculation}
+
+    if not updateState(dag, dag.clearanceState, clearanceBlock, true, cache,
+                       updateFlags1):
+      error "Unable to load clearance state for parent block, " &
+            "database corrupt?", clearanceBlock = shortLog(clearanceBlock)
+      return err(VerifierError.MissingParent)
+
+    dag.clearanceState.setStateRoot(trustedStateRoot)
+
+    let proposerVerifyTick = Moment.now()
+
+    if not(isNil(onStateUpdated)):
+      ? onStateUpdated(forkyBlck.message.slot)
+
+    let
+      stateDataTick = Moment.now()
+      updateFlags2 =
+        dag.updateFlags + {skipBlsValidation, skipStateRootValidation}
+
+    ? checkStateTransition(dag, forkyBlck.asSigVerified(), cache, updateFlags2)
+
+    let stateVerifyTick = Moment.now()
+
+    if bdata.blob.isSome():
+      for blob in bdata.blob.get():
+        dag.db.putBlobSidecar(blob[])
+
+    type Trusted = typeof forkyBlck.asTrusted()
+
+    proc onBlockAddedHandler(
+        blckRef: BlockRef,
+        trustedBlock: Trusted,
+        epochRef: EpochRef,
+        unrealized: FinalityCheckpoints
+    ) {.gcsafe, raises: [].} =
+      onBlockAdded(
+        blckRef,
+        ForkedTrustedSignedBeaconBlock.init(trustedBlock),
+        epochRef,
+        unrealized)
+
+    let blockHandler: BlockAdded(consensusFork) = onBlockAddedHandler
+
+    discard addResolvedHeadBlock(
+      dag, dag.clearanceState,
+      forkyBlck.asTrusted(),
+      true,
+      parent, cache,
+      blockHandler,
+      proposerVerifyTick - startTick,
+      stateDataTick - proposerVerifyTick,
+      stateVerifyTick - stateDataTick)
+
+  ok()

beacon_chain/consensus_object_pools/block_pools_types.nim

@@ -285,7 +285,11 @@ type
     # balances, as used in fork choice
     effective_balances_bytes*: seq[byte]
 
-  OnBlockAdded[T: ForkyTrustedSignedBeaconBlock] = proc(
+  BlockData* = object
+    blck*: ForkedSignedBeaconBlock
+    blob*: Opt[BlobSidecars]
+
+  OnBlockAdded*[T: ForkyTrustedSignedBeaconBlock] = proc(
     blckRef: BlockRef, blck: T, epochRef: EpochRef,
     unrealized: FinalityCheckpoints) {.gcsafe, raises: [].}
   OnPhase0BlockAdded* = OnBlockAdded[phase0.TrustedSignedBeaconBlock]
@@ -299,6 +303,13 @@ type
     OnPhase0BlockAdded | OnAltairBlockAdded | OnBellatrixBlockAdded |
     OnCapellaBlockAdded | OnDenebBlockAdded | OnElectraBlockAdded
 
+  OnForkedBlockAdded* = proc(
+    blckRef: BlockRef, blck: ForkedTrustedSignedBeaconBlock, epochRef: EpochRef,
+    unrealized: FinalityCheckpoints) {.gcsafe, raises: [].}
+
+  OnStateUpdated* = proc(
+    slot: Slot): Result[void, VerifierError] {.gcsafe, raises: [].}
+
   HeadChangeInfoObject* = object
     slot*: Slot
     block_root* {.serializedFieldName: "block".}: Eth2Digest