BN: New syncing algorithm

AllTests-mainnet.md

@@ -159,6 +159,20 @@ AllTests-mainnet
 + atSlot sanity                                                                              OK
 + parent sanity                                                                              OK
 ```
+## BlocksRangeBuffer test suite
+```diff
++ Add and query blocks test [backward]                                                       OK
++ Add and query blocks test [forward]                                                        OK
++ Block insertion test [backward]                                                            OK
++ Block insertion test [forward]                                                             OK
++ Buffer advance test [backward]                                                             OK
++ Buffer advance test [forward]                                                              OK
++ Buffer invalidate test [backward]                                                          OK
++ Buffer invalidate test [forward]                                                           OK
++ Range peek real test cases [forward]                                                       OK
++ Range peek test [backward]                                                                 OK
++ Range peek test [forward]                                                                  OK
+```
 ## ColumnMap test suite
 ```diff
 + and() operation test                                                                       OK
@@ -990,29 +1004,19 @@ AllTests-mainnet
 ## SyncManager test suite
 ```diff
 + [SyncManager] groupBlobs() test                                                            OK
-+ [SyncQueue# & Backward] Combination of missing parent and good blocks [3 peers] test       OK
-+ [SyncQueue# & Backward] Empty responses should not advance queue until other peers will no OK
-+ [SyncQueue# & Backward] Empty responses should not be accounted [3 peers] test             OK
-+ [SyncQueue# & Backward] Failure request push test                                          OK
-+ [SyncQueue# & Backward] Invalid block [3 peers] test                                       OK
-+ [SyncQueue# & Backward] Smoke [3 peers] test                                               OK
-+ [SyncQueue# & Backward] Smoke [single peer] test                                           OK
-+ [SyncQueue# & Backward] Unviable block [3 peers] test                                      OK
-+ [SyncQueue# & Backward] epochFilter() test                                                 OK
-+ [SyncQueue# & Forward] Combination of missing parent and good blocks [3 peers] test        OK
-+ [SyncQueue# & Forward] Empty responses should not advance queue until other peers will not OK
-+ [SyncQueue# & Forward] Empty responses should not be accounted [3 peers] test              OK
-+ [SyncQueue# & Forward] Failure request push test                                           OK
-+ [SyncQueue# & Forward] Invalid block [3 peers] test                                        OK
-+ [SyncQueue# & Forward] Smoke [3 peers] test                                                OK
-+ [SyncQueue# & Forward] Smoke [single peer] test                                            OK
-+ [SyncQueue# & Forward] Unviable block [3 peers] test                                       OK
-+ [SyncQueue# & Forward] epochFilter() test                                                  OK
++ [SyncQueue#Backward] Combination of missing parent and good blocks [3 peers] test          OK
++ [SyncQueue#Backward] Empty responses should not advance queue until other peers will not c OK
++ [SyncQueue#Backward] Empty responses should not be accounted [3 peers] test                OK
++ [SyncQueue#Backward] Failure request push test                                             OK
++ [SyncQueue#Backward] Invalid block [3 peers] test                                          OK
 + [SyncQueue#Backward] Missing parent and exponential rewind [3 peers] test                  OK
++ [SyncQueue#Backward] Smoke [3 peers] test                                                  OK
++ [SyncQueue#Backward] Smoke [single peer] test                                              OK
++ [SyncQueue#Backward] Unviable block [3 peers] test                                         OK
++ [SyncQueue#Backward] epochFilter() test                                                    OK
 + [SyncQueue#Backward] getRewindPoint() test                                                 OK
 + [SyncQueue#Forward] Missing parent and exponential rewind [3 peers] test                   OK
 + [SyncQueue#Forward] getRewindPoint() test                                                  OK
-+ [SyncQueue] checkBlobsResponse() test                                                      OK
 + [SyncQueue] checkResponse() test                                                           OK
 + [SyncQueue] hasEndGap() test                                                               OK
 ```

beacon_chain/beacon_node.nim

@@ -28,7 +28,7 @@ import
   ./spec/datatypes/[base, altair],
   ./spec/eth2_apis/dynamic_fee_recipients,
   ./spec/signatures_batch,
-  ./sync/[sync_manager, request_manager, sync_types, validator_custody],
+  ./sync/[sync_overseer2, sync_manager, request_manager, sync_types, validator_custody],
   ./validators/[
     action_tracker, message_router, validator_monitor, validator_pool,
     keystore_management],
@@ -41,13 +41,14 @@ 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, sync_types
+  consensus_manager, dynamic_fee_recipients, sync_types, sync_overseer2
 
 type
   EventBus* = object
     headQueue*: AsyncEventQueue[HeadChangeInfoObject]
     blocksQueue*: AsyncEventQueue[EventBeaconBlockObject]
     blockGossipQueue*: AsyncEventQueue[EventBeaconBlockGossipObject]
+    blockGossipPeerQueue*: AsyncEventQueue[EventBeaconBlockGossipPeerObject]
     phase0AttestQueue*: AsyncEventQueue[phase0.Attestation]
     singleAttestQueue*: AsyncEventQueue[SingleAttestation]
     exitQueue*: AsyncEventQueue[SignedVoluntaryExit]
@@ -100,7 +101,7 @@ type
     syncManager*: SyncManager[Peer, PeerId]
     backfiller*: SyncManager[Peer, PeerId]
     untrustedManager*: SyncManager[Peer, PeerId]
-    syncOverseer*: SyncOverseerRef
+    syncOverseer*: SyncOverseerRef2
     processor*: ref Eth2Processor
     batchVerifier*: ref BatchVerifier
     blockProcessor*: ref BlockProcessor

beacon_chain/beacon_node_light_client.nim

@@ -54,7 +54,7 @@ proc initLightClient*(
     optimisticProcessor = initOptimisticProcessor(
       cfg.timeParams, getBeaconTime, optimisticHandler)
 
-    shouldInhibitSync = func(): bool =
+    shouldInhibitSync = proc(): bool =
       if isNil(node.syncOverseer):
         false
       else:

beacon_chain/consensus_object_pools/block_pools_types.nim

@@ -11,8 +11,7 @@ import
   # Standard library
   std/[tables, hashes],
   # Status libraries
-  chronicles,
-  results,
+  chronicles, libp2p/peerid, results,
   # Internals
   ../spec/[signatures_batch, forks, helpers],
   ".."/[beacon_chain_db, era_db],
@@ -46,6 +45,9 @@ type
     Duplicate
       ## We've seen this value already, can't add again
 
+    MissingSidecars
+      ## We do not have sidecars at the moment
+
   OnBlockCallback* =
     proc(data: ForkedTrustedSignedBeaconBlock) {.gcsafe, raises: [].}
   OnBlockGossipCallback* =
@@ -336,6 +338,30 @@ type
     slot*: Slot
     block_root* {.serializedFieldName: "block".}: Eth2Digest
 
+  EventBeaconBlockGossipPeerObject* = object
+    blck*: ForkedSignedBeaconBlock
+    src*: PeerId
+
+template OnBlockAddedCallback*(kind: static ConsensusFork): auto =
+  when kind == ConsensusFork.Gloas:
+    typedesc[OnGloasBlockAdded]
+  elif kind == ConsensusFork.Fulu:
+    typedesc[OnFuluBlockAdded]
+  elif kind == ConsensusFork.Electra:
+    typedesc[OnElectraBlockAdded]
+  elif kind == ConsensusFork.Deneb:
+    typedesc[OnDenebBlockAdded]
+  elif kind == ConsensusFork.Capella:
+    typedesc[OnCapellaBlockAdded]
+  elif kind == ConsensusFork.Bellatrix:
+    typedesc[OnBellatrixBlockAdded]
+  elif kind == ConsensusFork.Altair:
+    typedesc[OnAltairBlockAdded]
+  elif kind == ConsensusFork.Phase0:
+    typedesc[OnPhase0BlockAdded]
+  else:
+    static: raiseAssert "Unreachable"
+
 template timeParams*(dag: ChainDAGRef): TimeParams =
   dag.cfg.timeParams
 
@@ -470,3 +496,13 @@ func init*(t: typedesc[EventBeaconBlockGossipObject],
       slot: forkyBlck.message.slot,
       block_root: forkyBlck.root
     )
+
+func init*(
+    t: typedesc[EventBeaconBlockGossipPeerObject],
+    v: ForkySignedBeaconBlock,
+    s: PeerId
+): EventBeaconBlockGossipPeerObject =
+  EventBeaconBlockGossipPeerObject(
+    blck: ForkedSignedBeaconBlock.init(v),
+    src: s
+  )

beacon_chain/nimbus_beacon_node.nim

@@ -24,7 +24,7 @@ import
   ./spec/datatypes/[altair, bellatrix, phase0],
   ./spec/[
     engine_authentication, weak_subjectivity, peerdas_helpers],
-  ./sync/[sync_protocol, light_client_protocol, sync_overseer, validator_custody],
+  ./sync/[sync_protocol, light_client_protocol, sync_overseer2, validator_custody],
   ./validators/[keystore_management, beacon_validators],
   ./[
     beacon_node, beacon_node_light_client, buildinfo, deposits,
@@ -535,56 +535,57 @@ proc initFullNode(
         {SyncManagerFlag.NoGenesisSync}
       else:
         {}
-    syncManager = newSyncManager[Peer, PeerId](
-      node.network.peerPool,
-      dag.cfg.DENEB_FORK_EPOCH,
-      dag.cfg.FULU_FORK_EPOCH,
-      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
-      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
-      SyncQueueKind.Forward, getLocalHeadSlot,
-      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
-      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
-      dag.tail.slot, blockVerifier, forkAtEpoch,
-      shutdownEvent = node.shutdownEvent,
-      flags = syncManagerFlags)
-    backfiller = newSyncManager[Peer, PeerId](
-      node.network.peerPool,
-      dag.cfg.DENEB_FORK_EPOCH,
-      dag.cfg.FULU_FORK_EPOCH,
-      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
-      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
-      SyncQueueKind.Backward, getLocalHeadSlot,
-      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
-      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
-      dag.backfill.slot, blockVerifier, forkAtEpoch, maxHeadAge = 0,
-      shutdownEvent = node.shutdownEvent,
-      flags = syncManagerFlags)
-    clistPivotSlot =
-      if clist.tail.isSome():
-        clist.tail.get().blck.slot()
-      else:
-        getLocalWallSlot()
+#    syncManager = newSyncManager[Peer, PeerId](
+#      node.network.peerPool,
+#      dag.cfg.DENEB_FORK_EPOCH,
+#      dag.cfg.FULU_FORK_EPOCH,
+#      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
+#      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
+#      SyncQueueKind.Forward, getLocalHeadSlot,
+#      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
+#      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
+#      dag.tail.slot, blockVerifier, forkAtEpoch,
+#      shutdownEvent = node.shutdownEvent,
+#      flags = syncManagerFlags)
+#    backfiller = newSyncManager[Peer, PeerId](
+#      node.network.peerPool,
+#      dag.cfg.DENEB_FORK_EPOCH,
+#      dag.cfg.FULU_FORK_EPOCH,
+#      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
+#      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
+#      SyncQueueKind.Backward, getLocalHeadSlot,
+#      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
+#      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
+#      dag.backfill.slot, blockVerifier, forkAtEpoch, maxHeadAge = 0,
+#      shutdownEvent = node.shutdownEvent,
+#      flags = syncManagerFlags)
+#    clistPivotSlot =
+#      if clist.tail.isSome():
+#        clist.tail.get().blck.slot()
+#      else:
+#        getLocalWallSlot()
     eaSlot = dag.head.slot
-    untrustedManager = newSyncManager[Peer, PeerId](
-      node.network.peerPool,
-      dag.cfg.DENEB_FORK_EPOCH,
-      dag.cfg.FULU_FORK_EPOCH,
-      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
-      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
-      SyncQueueKind.Backward, getLocalHeadSlot,
-      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getUntrustedBackfillSlot,
-      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
-      clistPivotSlot, untrustedBlockVerifier, forkAtEpoch, maxHeadAge = 0,
-      shutdownEvent = node.shutdownEvent,
-      flags = syncManagerFlags)
+    erSlot = dag.head.slot
+#    untrustedManager = newSyncManager[Peer, PeerId](
+#      node.network.peerPool,
+#      dag.cfg.DENEB_FORK_EPOCH,
+#      dag.cfg.FULU_FORK_EPOCH,
+#      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
+#      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
+#      SyncQueueKind.Backward, getLocalHeadSlot,
+#      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getUntrustedBackfillSlot,
+#      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
+#      clistPivotSlot, untrustedBlockVerifier, forkAtEpoch, maxHeadAge = 0,
+#      shutdownEvent = node.shutdownEvent,
+#      flags = syncManagerFlags)
     router = (ref MessageRouter)(
       processor: processor,
       network: node.network)
-    requestManager = RequestManager.init(
-      node.network, supernode, custodyColumns,
-      dag.cfg.DENEB_FORK_EPOCH, getBeaconTime, (proc(): bool = syncManager.inProgress),
-      quarantine, blobQuarantine, dataColumnQuarantine, rmanBlockVerifier,
-      rmanBlockLoader, rmanBlobLoader, rmanDataColumnLoader)
+#    requestManager = RequestManager.init(
+#      node.network, supernode, custodyColumns,
+#      dag.cfg.DENEB_FORK_EPOCH, getBeaconTime, (proc(): bool = syncManager.inProgress),
+#      quarantine, blobQuarantine, dataColumnQuarantine, rmanBlockVerifier,
+#      rmanBlockLoader, rmanBlobLoader, rmanDataColumnLoader)
     validatorCustody = ValidatorCustodyRef.init(node.network, dag, custodyColumns,
       dataColumnQuarantine)
 
@@ -647,22 +648,18 @@ proc initFullNode(
   node.batchVerifier = batchVerifier
   node.blockProcessor = blockProcessor
   node.consensusManager = consensusManager
-  node.requestManager = requestManager
   node.validatorCustody = validatorCustody
-  node.syncManager = syncManager
-  node.backfiller = backfiller
-  node.untrustedManager = untrustedManager
-  node.syncOverseer = SyncOverseerRef.new(node.consensusManager,
-                                          node.validatorMonitor,
-                                          config,
-                                          getBeaconTime,
-                                          node.list,
-                                          node.beaconClock,
-                                          node.eventBus.optFinHeaderUpdateQueue,
-                                          node.network.peerPool,
-                                          node.batchVerifier,
-                                          syncManager, backfiller,
-                                          untrustedManager)
+  # node.requestManager = requestManager
+  # node.syncManager = syncManager
+  # node.backfiller = backfiller
+  # node.untrustedManager = untrustedManager
+  node.syncOverseer =
+    SyncOverseerRef2.new(node.network, node.consensusManager, config,
+                         getBeaconTime, node.beaconClock, blockProcessor,
+                         quarantine, blobQuarantine, dataColumnQuarantine,
+                         node.eventBus.blockGossipPeerQueue,
+                         node.eventBus.blocksQueue,
+                         node.eventBus.finalQueue)
   node.router = router
 
   await node.addValidators()
@@ -766,6 +763,7 @@ proc init*(
       headQueue: newAsyncEventQueue[HeadChangeInfoObject](),
       blocksQueue: newAsyncEventQueue[EventBeaconBlockObject](),
       blockGossipQueue: newAsyncEventQueue[EventBeaconBlockGossipObject](),
+      blockGossipPeerQueue: newAsyncEventQueue[EventBeaconBlockGossipPeerObject](),
       phase0AttestQueue: newAsyncEventQueue[phase0.Attestation](),
       singleAttestQueue: newAsyncEventQueue[SingleAttestation](),
       exitQueue: newAsyncEventQueue[SignedVoluntaryExit](),
@@ -1813,8 +1811,6 @@ proc onSlotEnd(node: BeaconNode, slot: Slot) {.async.} =
           .pruneAfterFinalization(
             node.dag.finalizedHead.slot.epoch()
           )
-    node.processor.blobQuarantine[].pruneAfterFinalization(
-      node.dag.finalizedHead.slot.epoch(), node.dag.needsBackfill())
     node.processor.quarantine[].pruneAfterFinalization(
       node.dag.finalizedHead.slot.epoch(), node.dag.needsBackfill())
 
@@ -2149,7 +2145,7 @@ proc onSlotStart(node: BeaconNode, wallTime: BeaconTime,
   node.consensusManager[].updateHead(wallSlot)
 
   await node.handleValidatorDuties(lastSlot, wallSlot)
-  node.requestManager.switchToColumnLoop()
+  # node.requestManager.switchToColumnLoop()
   await onSlotEnd(node, wallSlot)
 
   # https://github.com/ethereum/builder-specs/blob/v0.4.0/specs/bellatrix/validator.md#registration-dissemination
@@ -2289,20 +2285,27 @@ proc installMessageValidators(node: BeaconNode) =
         # beacon_block
         # https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.8/specs/phase0/p2p-interface.md#beacon_block
         # https://github.com/ethereum/consensus-specs/blob/v1.6.0-beta.0/specs/gloas/p2p-interface.md#beacon_block
-        node.network.addValidator(
-          getBeaconBlocksTopic(digest), proc (
-            signedBlock: consensusFork.SignedBeaconBlock,
-            src: PeerId,
-          ): ValidationResult =
-            if node.shouldSyncOptimistically(node.currentSlot):
-              toValidationResult(
-                node.optimisticProcessor.processSignedBeaconBlock(
-                  signedBlock))
-            else:
-              toValidationResult(
-                node.processor[].processSignedBeaconBlock(
-                  MsgSource.gossip, signedBlock)))
-
+        when consensusFork >= ConsensusFork.Gloas:
+          debugGloasComment " "
+        else:
+          node.network.addValidator(
+            getBeaconBlocksTopic(digest), proc (
+              signedBlock: consensusFork.SignedBeaconBlock,
+              src: PeerId,
+            ): ValidationResult =
+              if node.shouldSyncOptimistically(node.currentSlot):
+                toValidationResult(
+                  node.optimisticProcessor.processSignedBeaconBlock(
+                    signedBlock))
+              else:
+                let res =
+                  toValidationResult(
+                    node.processor[].processSignedBeaconBlock(
+                      MsgSource.gossip, signedBlock))
+                if res == ValidationResult.Accept:
+                  node.eventBus.blockGossipPeerQueue.emit(
+                    EventBeaconBlockGossipPeerObject.init(signedBlock, src))
+                res)
         # beacon_attestation_{subnet_id}
         # https://github.com/ethereum/consensus-specs/blob/v1.4.0-beta.5/specs/phase0/p2p-interface.md#beacon_attestation_subnet_id
         # https://github.com/ethereum/consensus-specs/blob/v1.6.0-beta.0/specs/gloas/p2p-interface.md#beacon_attestation_subnet_id
@@ -2571,7 +2574,7 @@ proc run*(node: BeaconNode, stopper: StopFuture) {.raises: [CatchableError].} =
     wallSlot = wallTime.slotOrZero(node.dag.timeParams)
 
   node.startLightClient()
-  node.requestManager.start()
+  # node.requestManager.start()
   node.syncOverseer.start()
 
   waitFor node.updateGossipStatus(wallSlot)