feat: shared CustodyConfig on main thread and network thread

packages/beacon-node/src/chain/chain.ts

@@ -49,7 +49,7 @@ import {Metrics} from "../metrics/index.js";
 import {NodeId} from "../network/subnets/interface.js";
 import {BufferPool} from "../util/bufferPool.js";
 import {Clock, ClockEvent, IClock} from "../util/clock.js";
-import {getCustodyConfig} from "../util/dataColumns.js";
+import {CustodyConfig, computeCustodyConfig} from "../util/dataColumns.js";
 import {ensureDir, writeIfNotExist} from "../util/file.js";
 import {isOptimisticBlock} from "../util/forkChoice.js";
 import {SerializedCache} from "../util/serializedCache.js";
@@ -121,6 +121,9 @@ export class BeaconChain implements IBeaconChain {
   readonly executionBuilder?: IExecutionBuilder;
   // Expose config for convenience in modularized functions
   readonly config: BeaconConfig;
+  // TODO - das: mutate custodyConfig due to VALIDATOR_CUSTODY_REQURIEMENT
+  // need to sync with networkConfig inside Network too
+  readonly custodyConfig: CustodyConfig;
   readonly logger: Logger;
   readonly metrics: Metrics | null;
   readonly bufferPool: BufferPool | null;
@@ -253,8 +256,8 @@ export class BeaconChain implements IBeaconChain {
     this.seenAggregatedAttestations = new SeenAggregatedAttestations(metrics);
     this.seenContributionAndProof = new SeenContributionAndProof(metrics);
     this.seenAttestationDatas = new SeenAttestationDatas(metrics, this.opts?.attDataCacheSlotDistance);
-    const custodyConfig = getCustodyConfig(nodeId, config);
-    this.seenGossipBlockInput = new SeenGossipBlockInput(custodyConfig);
+    this.custodyConfig = computeCustodyConfig(nodeId, config);
+    this.seenGossipBlockInput = new SeenGossipBlockInput(this.custodyConfig);
 
     this.beaconProposerCache = new BeaconProposerCache(opts);
     this.checkpointBalancesCache = new CheckpointBalancesCache();

packages/beacon-node/src/chain/interface.ts

@@ -61,6 +61,7 @@ import {SeenAggregatedAttestations} from "./seenCache/seenAggregateAndProof.js";
 import {SeenAttestationDatas} from "./seenCache/seenAttestationData.js";
 import {SeenBlockAttesters} from "./seenCache/seenBlockAttesters.js";
 import {ShufflingCache} from "./shufflingCache.js";
+import {CustodyConfig} from "../util/dataColumns.js";
 
 export {BlockType, type AssembledBlockType};
 export {type ProposerPreparationData};
@@ -88,6 +89,7 @@ export interface IBeaconChain {
   readonly executionBuilder?: IExecutionBuilder;
   // Expose config for convenience in modularized functions
   readonly config: BeaconConfig;
+  readonly custodyConfig: CustodyConfig;
   readonly logger: Logger;
   readonly metrics: Metrics | null;
   readonly bufferPool: BufferPool | null;

packages/beacon-node/src/network/core/networkCore.ts

@@ -34,6 +34,7 @@ import {SyncnetsService} from "../subnets/syncnetsService.js";
 import {getConnectionsMap} from "../util.js";
 import {NetworkCoreMetrics, createNetworkCoreMetrics} from "./metrics.js";
 import {INetworkCore, MultiaddrStr, PeerIdStr} from "./types.js";
+import {NetworkConfig} from "../networkConfig.js";
 
 type Mods = {
   libp2p: Libp2p;
@@ -42,6 +43,7 @@ type Mods = {
   attnetsService: IAttnetsService;
   syncnetsService: SyncnetsService;
   peerManager: PeerManager;
+  networkConfig: NetworkConfig;
   peersData: PeersData;
   metadata: MetadataController;
   logger: LoggerNode;
@@ -88,6 +90,7 @@ export class NetworkCore implements INetworkCore {
   private readonly attnetsService: IAttnetsService;
   private readonly syncnetsService: SyncnetsService;
   private readonly peerManager: PeerManager;
+  private readonly networkConfig: NetworkConfig;
   private readonly peersData: PeersData;
   private readonly reqResp: ReqRespBeaconNode;
   private readonly gossip: Eth2Gossipsub;
@@ -111,6 +114,7 @@ export class NetworkCore implements INetworkCore {
     this.attnetsService = modules.attnetsService;
     this.syncnetsService = modules.syncnetsService;
     this.peerManager = modules.peerManager;
+    this.networkConfig = modules.networkConfig;
     this.peersData = modules.peersData;
     this.metadata = modules.metadata;
     this.logger = modules.logger;
@@ -193,13 +197,12 @@ export class NetworkCore implements INetworkCore {
     // should be called before AttnetsService constructor so that node subscribe to deterministic attnet topics
     await gossip.start();
 
-    const nodeId = computeNodeId(peerId);
-    const attnetsService = new AttnetsService(config, clock, gossip, metadata, logger, metrics, nodeId, opts);
+    const networkConfig = new NetworkConfig(peerId, config);
+    const attnetsService = new AttnetsService(config, clock, gossip, metadata, logger, metrics, networkConfig.getNodeId(), opts);
     const syncnetsService = new SyncnetsService(config, clock, gossip, metadata, logger, metrics, opts);
 
     const peerManager = await PeerManager.init(
       {
-        nodeId,
         libp2p,
         gossip: gossip,
         reqResp,
@@ -208,9 +211,9 @@ export class NetworkCore implements INetworkCore {
         logger,
         metrics,
         clock,
-        config,
         peerRpcScores,
         events,
+        networkConfig: networkConfig,
         peersData,
         statusCache,
       },
@@ -236,6 +239,7 @@ export class NetworkCore implements INetworkCore {
       attnetsService,
       syncnetsService,
       peerManager,
+      networkConfig: networkConfig,
       peersData,
       metadata,
       logger,
@@ -274,6 +278,10 @@ export class NetworkCore implements INetworkCore {
     this.closed = true;
   }
 
+  getNetworkConfig(): NetworkConfig {
+    return this.networkConfig;
+  }
+
   async scrapeMetrics(): Promise<string> {
     return [
       (await this.metrics?.register.metrics()) ?? "",

packages/beacon-node/src/network/network.ts

@@ -32,7 +32,7 @@ import {IBeaconChain} from "../chain/index.js";
 import {IBeaconDb} from "../db/interface.js";
 import {Metrics, RegistryMetricCreator} from "../metrics/index.js";
 import {IClock} from "../util/clock.js";
-import {CustodyConfig, getCustodyConfig} from "../util/dataColumns.js";
+import {CustodyConfig} from "../util/dataColumns.js";
 import {PeerIdStr, peerIdToString} from "../util/peerId.js";
 import {BlobSidecarsByRootRequest} from "../util/types.js";
 import {INetworkCore, NetworkCore, WorkerNetworkCore} from "./core/index.js";
@@ -120,7 +120,7 @@ export class Network implements INetwork {
     this.peerId = modules.peerId;
     this.nodeId = modules.nodeId;
     this.config = modules.config;
-    this.custodyConfig = getCustodyConfig(modules.nodeId, modules.config);
+    this.custodyConfig = modules.chain.custodyConfig;
     this.logger = modules.logger;
     this.chain = modules.chain;
     this.clock = modules.chain.clock;

packages/beacon-node/src/network/networkConfig.ts

@@ -0,0 +1,42 @@
+import {BeaconConfig} from "@lodestar/config";
+import {NodeId, computeNodeId} from "./subnets";
+import {CustodyConfig, computeCustodyConfig} from "../util/dataColumns";
+import {PeerId} from "@libp2p/interface";
+
+/**
+ * Store shared data for different modules in the network stack.
+ * TODO: consider moving similar shared data, for example PeersData, under NetworkConfig.
+ */
+export class NetworkConfig {
+  private readonly nodeId: NodeId;
+  private readonly config: BeaconConfig;
+  private custodyConfig: CustodyConfig;
+
+  constructor(peerId: PeerId, config: BeaconConfig) {
+    this.nodeId = computeNodeId(peerId);
+    this.config = config;
+    this.custodyConfig = computeCustodyConfig(this.nodeId, config);
+  }
+
+  getConfig(): BeaconConfig {
+    return this.config;
+  }
+
+  getNodeId(): NodeId {
+    return this.nodeId;
+  }
+
+  /**
+   * Consumer should never mutate returned CustodyConfig
+   */
+  getCustodyConfig(): CustodyConfig {
+    return this.custodyConfig;
+  }
+
+  /**
+   * Recompute CustodyConfig based on connected validators.
+   */
+  recomputeCustodyConfig(): void {
+    // TODO - das
+  }
+}

packages/beacon-node/src/network/peers/discover.ts

@@ -6,7 +6,6 @@ import {LoggerNode} from "@lodestar/logger/node";
 import {ATTESTATION_SUBNET_COUNT, ForkSeq, SYNC_COMMITTEE_SUBNET_COUNT} from "@lodestar/params";
 import {CustodyIndex, SubnetID} from "@lodestar/types";
 import {pruneSetToMax, sleep} from "@lodestar/utils";
-import {ColumnIndex} from "@lodestar/types";
 import {bytesToInt} from "@lodestar/utils";
 import {Multiaddr} from "@multiformats/multiaddr";
 import {getCustodyGroups, getDataColumns} from "../../util/dataColumns.js";
@@ -21,6 +20,7 @@ import {IPeerRpcScoreStore, ScoreState} from "./score/index.js";
 import {deserializeEnrSubnets, zeroAttnets, zeroSyncnets} from "./utils/enrSubnetsDeserialize.js";
 import {type GroupQueries } from "./utils/prioritizePeers.js";
 import {IClock} from "../../util/clock.js";
+import {NetworkConfig} from "../networkConfig.js";
 
 /** Max number of cached ENRs after discovering a good peer */
 const MAX_CACHED_ENRS = 100;
@@ -39,13 +39,12 @@ export type PeerDiscoveryOpts = {
 };
 
 export type PeerDiscoveryModules = {
-  nodeId: NodeId;
+  networkConfig: NetworkConfig;
   libp2p: Libp2p;
   clock: IClock;
   peerRpcScores: IPeerRpcScoreStore;
   metrics: NetworkCoreMetrics | null;
   logger: LoggerNode;
-  config: BeaconConfig;
 };
 
 type PeerIdStr = string;
@@ -133,22 +132,19 @@ export class PeerDiscovery {
   private onlyConnectToMinimalCustodyOverlapNodes: boolean | undefined = false;
 
   constructor(modules: PeerDiscoveryModules, opts: PeerDiscoveryOpts, discv5: Discv5Worker) {
-    const {libp2p, clock, peerRpcScores, metrics, logger, config, nodeId} = modules;
+    const {libp2p, clock, peerRpcScores, metrics, logger, networkConfig} = modules;
     this.libp2p = libp2p;
     this.clock = clock;
     this.peerRpcScores = peerRpcScores;
     this.metrics = metrics;
     this.logger = logger;
-    this.config = config;
+    this.config = networkConfig.getConfig();
     this.discv5 = discv5;
     // TODO-das: remove
-    this.nodeId = nodeId;
+    this.nodeId = networkConfig.getNodeId();
     // we will only connect to peers that can provide us custody
     // TODO: @matthewkeil check if this needs to be updated for custody groups
-    this.sampleSubnets = getDataColumns(
-      nodeId,
-      Math.max(config.CUSTODY_REQUIREMENT, config.NODE_CUSTODY_REQUIREMENT, config.SAMPLES_PER_SLOT)
-    );
+    this.sampleSubnets = networkConfig.getCustodyConfig().sampledSubnets;
     this.groupRequests = new Map();
 
     this.discv5StartMs = 0;
@@ -208,7 +204,7 @@ export class PeerDiscovery {
       peerId: modules.libp2p.peerId,
       metrics: modules.metrics ?? undefined,
       logger: modules.logger,
-      config: modules.config,
+      config: modules.networkConfig.getConfig(),
     });
 
     return new PeerDiscovery(modules, opts, discv5);

packages/beacon-node/src/network/peers/peerManager.ts

@@ -30,6 +30,7 @@ import {
   renderIrrelevantPeerType,
   PrioritizePeersOpts,
 } from "./utils/index.js";
+import {NetworkConfig} from "../networkConfig.js";
 
 /** heartbeat performs regular updates such as updating reputations and performing discovery requests */
 const HEARTBEAT_INTERVAL_MS = 30 * 1000;
@@ -96,7 +97,6 @@ export interface IReqRespBeaconNodePeerManager {
 }
 
 export type PeerManagerModules = {
-  nodeId: NodeId;
   libp2p: Libp2p;
   logger: LoggerNode;
   metrics: NetworkCoreMetrics | null;
@@ -105,9 +105,9 @@ export type PeerManagerModules = {
   attnetsService: SubnetsService;
   syncnetsService: SubnetsService;
   clock: IClock;
-  config: BeaconConfig;
   peerRpcScores: IPeerRpcScoreStore;
   events: INetworkEventBus;
+  networkConfig: NetworkConfig;
   peersData: PeersData;
   statusCache: StatusCache;
 };
@@ -156,6 +156,8 @@ export class PeerManager {
   private intervals: NodeJS.Timeout[] = [];
 
   constructor(modules: PeerManagerModules, opts: PeerManagerOpts, discovery: PeerDiscovery | null) {
+    const {networkConfig} = modules;
+    const custodyConfig = networkConfig.getCustodyConfig();
     this.libp2p = modules.libp2p;
     this.logger = modules.logger;
     this.metrics = modules.metrics;
@@ -165,25 +167,19 @@ export class PeerManager {
     this.syncnetsService = modules.syncnetsService;
     this.statusCache = modules.statusCache;
     this.clock = modules.clock;
-    this.config = modules.config;
+    this.config = networkConfig.getConfig();
     this.peerRpcScores = modules.peerRpcScores;
     this.networkEventBus = modules.events;
     this.connectedPeers = modules.peersData.connectedPeers;
     this.opts = opts;
     this.discovery = discovery;
-    this.nodeId = modules.nodeId;
+    this.nodeId = networkConfig.getNodeId();
     // we will only connect to peers that can provide us custody
     // TODO: @matthewkeil check if this needs to be updated for custody groups
     // TODO(das): may not need this, use `this.samplingGroups` instead
-    this.sampleSubnets = getDataColumns(
-      this.nodeId,
-      Math.max(this.config.CUSTODY_REQUIREMENT, this.config.NODE_CUSTODY_REQUIREMENT, this.config.SAMPLES_PER_SLOT)
-    );
+    this.sampleSubnets = custodyConfig.sampledSubnets;
     // TODO(das): get from custodyConfig or a centralized place every time, instead of computing once here
-    this.samplingGroups = getCustodyGroups(
-      this.nodeId,
-      Math.max(this.config.CUSTODY_REQUIREMENT, this.config.NODE_CUSTODY_REQUIREMENT, this.config.SAMPLES_PER_SLOT)
-    );
+    this.samplingGroups = custodyConfig.sampleGroups;
 
     const {metrics} = modules;
     if (metrics) {

packages/beacon-node/src/util/dataColumns.ts

@@ -10,17 +10,32 @@ export type CustodyConfig = {
   custodyColumnsIndex: Uint8Array;
   custodyColumnsLen: number;
   custodyColumns: ColumnIndex[];
+  sampleGroups: CustodyIndex[];
   sampledColumns: ColumnIndex[];
+  sampledSubnets: number[];
 };
 
-export function getCustodyConfig(nodeId: NodeId, config: ChainForkConfig): CustodyConfig {
+/**
+ * Compute CustodyConfig, should be computed once after startup and when connected validators change.
+ */
+export function computeCustodyConfig(nodeId: NodeId, config: ChainForkConfig): CustodyConfig {
   const custodyColumns = getDataColumns(nodeId, Math.max(config.CUSTODY_REQUIREMENT, config.NODE_CUSTODY_REQUIREMENT));
-  const sampledColumns = getDataColumns(
-    nodeId,
-    Math.max(config.CUSTODY_REQUIREMENT, config.NODE_CUSTODY_REQUIREMENT, config.SAMPLES_PER_SLOT)
-  );
+  // the same to getDataColumns but here we compute step by step to also get custodyGroups
+  // const sampledColumns = getDataColumns(
+  //   nodeId,
+  //   Math.max(config.CUSTODY_REQUIREMENT, config.NODE_CUSTODY_REQUIREMENT, config.SAMPLES_PER_SLOT)
+  // );
+  const custodyGroupCount = Math.max(config.CUSTODY_REQUIREMENT, config.NODE_CUSTODY_REQUIREMENT, config.SAMPLES_PER_SLOT);
+  const sampleGroups = getCustodyGroups(nodeId, custodyGroupCount)
+  const sampledColumns = sampleGroups.flatMap(computeColumnsForCustodyGroup)
+    .sort((a, b) => a - b);
   const custodyMeta = getCustodyColumnsMeta(custodyColumns);
-  return {...custodyMeta, custodyColumns, sampledColumns};
+  const sampledSubnets = sampledColumns.map(computeSubnetForDataColumn);
+  return {...custodyMeta, custodyColumns, sampleGroups, sampledColumns, sampledSubnets};
+}
+
+function computeSubnetForDataColumn(columnIndex: ColumnIndex): number {
+  return columnIndex % DATA_COLUMN_SIDECAR_SUBNET_COUNT;
 }
 
 function getCustodyColumnsMeta(custodyColumns: ColumnIndex[]): {
@@ -45,14 +60,14 @@ function getCustodyColumnsMeta(custodyColumns: ColumnIndex[]): {
  * SPEC FUNCTION
  * https://github.com/ethereum/consensus-specs/blob/dev/specs/fulu/das-core.md#compute_columns_for_custody_group
  */
-export function computeColumnsForCustodyGroup(custodyGroup: CustodyIndex): ColumnIndex[] {
-  if (custodyGroup > NUMBER_OF_CUSTODY_GROUPS) {
-    custodyGroup = NUMBER_OF_CUSTODY_GROUPS;
+export function computeColumnsForCustodyGroup(custodyIndex: CustodyIndex): ColumnIndex[] {
+  if (custodyIndex > NUMBER_OF_CUSTODY_GROUPS) {
+    custodyIndex = NUMBER_OF_CUSTODY_GROUPS;
   }
   const columnsPerCustodyGroup = Number(NUMBER_OF_COLUMNS / NUMBER_OF_CUSTODY_GROUPS);
   const columnIndexes = [];
   for (let i = 0; i < columnsPerCustodyGroup; i++) {
-    columnIndexes.push(NUMBER_OF_CUSTODY_GROUPS * i + custodyGroup);
+    columnIndexes.push(NUMBER_OF_CUSTODY_GROUPS * i + custodyIndex);
   }
   columnIndexes.sort((a, b) => a - b);
   return columnIndexes;

packages/beacon-node/test/unit/util/dataColumn.test.ts

@@ -8,7 +8,7 @@ import {afterEach, beforeAll, describe, expect, it} from "vitest";
 
 import {validateDataColumnsSidecars} from "../../../src/chain/validation/dataColumnSidecar.js";
 import {computeDataColumnSidecars} from "../../../src/util/blobs.js";
-import {getCustodyConfig, getDataColumns} from "../../../src/util/dataColumns.js";
+import {computeCustodyConfig, getDataColumns} from "../../../src/util/dataColumns.js";
 import {ckzg, initCKZG, loadEthereumTrustedSetup} from "../../../src/util/kzg.js";
 import {getMockedBeaconChain} from "../../mocks/mockedBeaconChain.js";
 import {generateRandomBlob, transactionForKzgCommitment} from "../../utils/kzg.js";
@@ -24,7 +24,7 @@ describe("getCustodyConfig", () => {
       FULU_FORK_EPOCH: Infinity,
     });
     const nodeId = fromHexString("cdbee32dc3c50e9711d22be5565c7e44ff6108af663b2dc5abd2df573d2fa83f");
-    const custodyConfig = getCustodyConfig(nodeId, config);
+    const custodyConfig = computeCustodyConfig(nodeId, config);
     const {custodyColumnsLen, custodyColumns, sampledColumns} = custodyConfig;
 
     expect(custodyColumnsLen).toEqual(4);