ColumnSyncer

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],
+  ./sync/[sync_manager, request_manager, sync_types, column_syncer],
   ./validators/[
     action_tracker, message_router, validator_monitor, validator_pool,
     keystore_management],
@@ -96,6 +96,7 @@ type
     vcProcess*: Process
     requestManager*: RequestManager
     syncManager*: SyncManager[Peer, PeerId]
+    columnManager*: ColumnManager[Peer, PeerId]
     backfiller*: SyncManager[Peer, PeerId]
     untrustedManager*: SyncManager[Peer, PeerId]
     syncOverseer*: SyncOverseerRef

beacon_chain/conf.nim

@@ -133,6 +133,10 @@ type
     Light = "light",
     Lenient = "lenient"
 
+  ColumnSyncerStrategy* {.pure.} = enum
+    Greedy =  "greedy"
+    Impartial = "impartial"
+
   BeaconNodeConf* = object
     configFile* {.
       desc: "Loads the configuration from a TOML file"
@@ -559,6 +563,14 @@ type
         desc: "Maximum number of sync committee periods to retain light client data"
         name: "light-client-data-max-periods" .}: Option[uint64]
 
+      columnSyncerStrategy* {.
+        hidden
+        desc: "Choose how to sync columns, " &
+              " 1) Greedy would make the BN actively filter peers based on higher column custody. " &
+              " 2) Impartial would rely on every peer for columns till requisite column count is reached. "
+        defaultValue: ColumnSyncerStrategy.Impartial
+        name: "debug-column-syncer".}: ColumnSyncerStrategy
+
       longRangeSync* {.
         hidden
         desc: "Enable long-range syncing (genesis sync)",

beacon_chain/networking/peer_scores.nim

@@ -1,5 +1,5 @@
 # beacon_chain
-# Copyright (c) 2018-2024 Status Research & Development GmbH
+# Copyright (c) 2018-2025 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).
@@ -14,6 +14,11 @@ const
     ## Score after which peer will be kicked
   PeerScoreHighLimit* = 1000
     ## Max value of peer's score
+  PeerScoreSupernode* = 700
+    ## Score assigned to a supernode in order to prevent disconnection
+  PeerScoreIntersectingColumns* = 200
+    ## Score assigned to a node with intersecting columns
+    ## to the local BN in order to prevent disconnection
   PeerScorePoorRequest* = -50
     ## This peer is not responding on time or behaving improperly otherwise
   PeerScoreInvalidRequest* = -500

beacon_chain/nimbus_beacon_node.nim

@@ -22,7 +22,9 @@ import
   ./spec/[
     deposit_snapshots, engine_authentication, weak_subjectivity,
     peerdas_helpers],
-  ./sync/[sync_protocol, light_client_protocol, sync_overseer],
+  ./sync/[
+    column_syncer, sync_protocol, light_client_protocol,
+    sync_overseer, column_syncer_assist],
   ./validators/[keystore_management, beacon_validators],
   "."/[
     beacon_node, beacon_node_light_client, deposits,
@@ -445,7 +447,16 @@ proc initFullNode(
       config.dumpEnabled, config.dumpDirInvalid, config.dumpDirIncoming,
       batchVerifier, consensusManager, node.validatorMonitor,
       blobQuarantine, dataColumnQuarantine, getBeaconTime)
-
+    peerdasBlockVerifier = proc(signedBlock: ForkedSignedBeaconBlock,
+                                columns: Opt[DataColumnSidecars], maybeFinalized: bool):
+        Future[Result[void, VerifierError]] {.async: (raises: [CancelledError], raw: true).} =
+      # The design with a callback for block verification is unusual compared
+      # to the rest of the application, but fits with the general approach
+      # taken in the sync/request managers - this is an architectural compromise
+      # that should probably be reimagined more holistically in the future.
+      blockProcessor[].addBlock(
+        MsgSource.gossip, signedBlock, Opt.none(BlobSidecars), columns,
+        maybeFinalized = maybeFinalized)
     blockVerifier = proc(signedBlock: ForkedSignedBeaconBlock,
                          blobs: Opt[BlobSidecars], maybeFinalized: bool):
         Future[Result[void, VerifierError]] {.async: (raises: [CancelledError], raw: true).} =
@@ -536,6 +547,29 @@ proc initFullNode(
         {SyncManagerFlag.NoGenesisSync}
       else:
         {}
+    columnManagerModes =
+      if node.config.longRangeSync != LongRangeSyncMode.Lenient:
+        {ColumnSyncerMode.NoGenesisSync}
+      else:
+        {}
+    columnSyncerFlags =
+      if node.config.columnSyncerStrategy == ColumnSyncerStrategy.Impartial:
+        {ColumnSyncerFlag.Impartial}
+      else:
+        {ColumnSyncerFlag.Greedy}
+
+    columnManager = newColumnManager[Peer, PeerId](
+      node.network.peerPool, dag.cfg, supernode, custody_columns_set,
+      custody_columns_list, dag.cfg.FULU_FORK_EPOCH,
+      dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
+      dag.cfg.MAX_BLOBS_PER_BLOCK_ELECTRA,
+      ColumnSyncerDirection.Forward, getLocalHeadSlot,
+      getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
+      getFrontfillSlot, isWithinWeakSubjectivityPeriod,
+      dag.tail.slot, peerdasBlockVerifier,
+      shutdownEvent = node.shutdownEvent,
+      flags = columnSyncerFlags,
+      modes = columnManagerModes)
     syncManager = newSyncManager[Peer, PeerId](
       node.network.peerPool,
       dag.cfg.DENEB_FORK_EPOCH,
@@ -645,6 +679,7 @@ proc initFullNode(
   node.blockProcessor = blockProcessor
   node.consensusManager = consensusManager
   node.requestManager = requestManager
+  node.columnManager = columnManager
   node.syncManager = syncManager
   node.backfiller = backfiller
   node.untrustedManager = untrustedManager
@@ -1895,6 +1930,28 @@ func formatNextConsensusFork(
     (if withVanityArt: nextConsensusFork.getVanityMascot & " " else: "") &
     $nextConsensusFork & ":" & $nextForkEpoch)
 
+func columnSyncStatus(node: BeaconNode, wallSlot: Slot): string =
+  let optimisticHead = not node.dag.head.executionValid
+  if node.columnManager.inProgress:
+    let
+      optimisticSuffix =
+        if optimisticHead:
+          "/opt"
+        else:
+          ""
+      lightClientSuffix =
+        if node.consensusManager[].shouldSyncOptimistically(wallSlot):
+          " - lc: " & $shortLog(node.consensusManager[].optimisticHead)
+        else:
+          ""
+    node.columnManager.syncStatus & optimisticSuffix & lightClientSuffix
+  elif node.backfiller.inProgress:
+    "backfill: " & node.backfiller.syncStatus
+  elif optimistic_head:
+    "synced/opt"
+  else:
+    "synced"
+
 func syncStatus(node: BeaconNode, wallSlot: Slot): string =
   node.syncOverseer.statusMsg.valueOr:
     let optimisticHead = not node.dag.head.executionValid
@@ -1946,6 +2003,7 @@ proc onSlotStart(node: BeaconNode, wallTime: BeaconTime,
       slot = shortLog(wallSlot)
       epoch = shortLog(wallSlot.epoch)
       sync = node.syncStatus(wallSlot)
+      columnSync = node.columnSyncStatus(wallSlot)
       peers = len(node.network.peerPool)
       head = shortLog(node.dag.head)
       finalized = shortLog(getStateField(
@@ -2256,6 +2314,7 @@ proc run(node: BeaconNode) {.raises: [CatchableError].} =
       node.network.cfg.FULU_FORK_EPOCH:
     node.requestManager.switchToColumnLoop()
   node.syncOverseer.start()
+  node.columnManager.start()
 
   waitFor node.updateGossipStatus(wallSlot)
 
@@ -2438,6 +2497,9 @@ when not defined(windows):
 
       of "sync_status":
         node.syncStatus(node.currentSlot)
+
+      of "column_sync_status":
+        node.columnSyncStatus(node.currentSlot)
       else:
         # We ignore typos for now and just render the expression
         # as it was written. TODO: come up with a good way to show