feat: peerDAS - implement validator custody

packages/beacon-node/src/api/impl/validator/index.ts

@@ -1420,6 +1420,11 @@ export function getValidatorApi(
       // If the discovery mechanism isn't disabled, attempt to set up a peer discovery for the
       // required subnets.
 
+      // Register validators in LocalValidatorRegistry
+      for (const subscription of subscriptions) {
+        chain.localValidatorRegistry.registerLocalValidator(subscription.validatorIndex);
+      }
+
       if (metrics) {
         for (const subscription of subscriptions) {
           metrics.registerLocalValidator(subscription.validatorIndex);

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

@@ -50,15 +50,15 @@ export async function writeBlockInputToDb(this: BeaconChain, blocksInput: BlockI
         });
       } else {
         const {custodyConfig} = this.seenGossipBlockInput;
-        const {custodyColumnsLen, custodyColumnsIndex, custodyColumns} = custodyConfig;
+        const {custodyColumnsIndex, custodyColumns} = custodyConfig.getCustodyColumnsWithIndex();
         const blobsLen = (block.message as fulu.BeaconBlock).body.blobKzgCommitments.length;
         let dataColumnsLen: number;
         let dataColumnsIndex: Uint8Array;
         if (blobsLen === 0) {
           dataColumnsLen = 0;
           dataColumnsIndex = new Uint8Array(NUMBER_OF_COLUMNS);
         } else {
-          dataColumnsLen = custodyColumnsLen;
+          dataColumnsLen = custodyColumns.length;
           dataColumnsIndex = custodyColumnsIndex;
         }
 
@@ -127,11 +127,8 @@ export async function removeEagerlyPersistedBlockInputs(this: BeaconChain, block
           blobsToRemove.push({blockRoot, slot, blobSidecars});
         } else {
           const {custodyConfig} = this.seenGossipBlockInput;
-          const {
-            custodyColumnsLen: dataColumnsLen,
-            custodyColumnsIndex: dataColumnsIndex,
-            custodyColumns,
-          } = custodyConfig;
+          const {custodyColumnsIndex: dataColumnsIndex, custodyColumns} = custodyConfig.getCustodyColumnsWithIndex();
+          const dataColumnsLen = custodyColumns.length;
           const dataColumnSidecars = (blockData as BlockInputDataColumns).dataColumns.filter((dataColumnSidecar) =>
             custodyColumns.includes(dataColumnSidecar.index)
           );

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} from "../util/dataColumns.js";
 import {ensureDir, writeIfNotExist} from "../util/file.js";
 import {isOptimisticBlock} from "../util/forkChoice.js";
 import {SerializedCache} from "../util/serializedCache.js";
@@ -105,6 +105,7 @@ import {FileCPStateDatastore} from "./stateCache/datastore/file.js";
 import {FIFOBlockStateCache} from "./stateCache/fifoBlockStateCache.js";
 import {InMemoryCheckpointStateCache} from "./stateCache/inMemoryCheckpointsCache.js";
 import {PersistentCheckpointStateCache} from "./stateCache/persistentCheckpointsCache.js";
+import {LocalValidatorRegistry} from "../node/localValidatorRegistry.js";
 
 /**
  * Arbitrary constants, blobs and payloads should be consumed immediately in the same slot
@@ -143,6 +144,9 @@ export class BeaconChain implements IBeaconChain {
   readonly syncContributionAndProofPool = new SyncContributionAndProofPool();
   readonly opPool = new OpPool();
 
+  // Data column custody config
+  readonly custodyConfig: CustodyConfig;
+
   // Gossip seen cache
   readonly seenAttesters = new SeenAttesters();
   readonly seenAggregators = new SeenAggregators();
@@ -180,6 +184,7 @@ export class BeaconChain implements IBeaconChain {
   protected readonly db: IBeaconDb;
   private abortController = new AbortController();
   private processShutdownCallback: ProcessShutdownCallback;
+  readonly localValidatorRegistry: LocalValidatorRegistry;
 
   constructor(
     opts: IChainOptions,
@@ -197,6 +202,7 @@ export class BeaconChain implements IBeaconChain {
       eth1,
       executionEngine,
       executionBuilder,
+      localValidatorRegistry,
     }: {
       nodeId: NodeId;
       config: BeaconConfig;
@@ -212,6 +218,7 @@ export class BeaconChain implements IBeaconChain {
       eth1: IEth1ForBlockProduction;
       executionEngine: IExecutionEngine;
       executionBuilder?: IExecutionBuilder;
+      localValidatorRegistry: LocalValidatorRegistry;
     }
   ) {
     this.opts = opts;
@@ -250,11 +257,12 @@ export class BeaconChain implements IBeaconChain {
       this.opts?.preaggregateSlotDistance
     );
 
+    this.custodyConfig = new CustodyConfig(nodeId, config);
+
     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.seenGossipBlockInput = new SeenGossipBlockInput(this.custodyConfig);
 
     this.beaconProposerCache = new BeaconProposerCache(opts);
     this.checkpointBalancesCache = new CheckpointBalancesCache();
@@ -387,6 +395,8 @@ export class BeaconChain implements IBeaconChain {
     clock.addListener(ClockEvent.epoch, this.onClockEpoch.bind(this));
     emitter.addListener(ChainEvent.forkChoiceFinalized, this.onForkChoiceFinalized.bind(this));
     emitter.addListener(ChainEvent.forkChoiceJustified, this.onForkChoiceJustified.bind(this));
+
+    this.localValidatorRegistry = localValidatorRegistry;
   }
 
   async init(): Promise<void> {
@@ -1134,11 +1144,13 @@ export class BeaconChain implements IBeaconChain {
     }
 
     const metrics = this.metrics;
-    if (metrics && (slot + 1) % SLOTS_PER_EPOCH === 0) {
+    if ((slot + 1) % SLOTS_PER_EPOCH === 0) {
       // On the last slot of the epoch
       sleep((1000 * this.config.SECONDS_PER_SLOT) / 2)
-        .then(() => metrics.onceEveryEndOfEpoch(this.getHeadState()))
-        .catch((e) => {
+        .then(() => {
+          metrics?.onceEveryEndOfEpoch(this.getHeadState())
+          this.localValidatorRegistry.pruneInactiveValidators()
+        }).catch((e) => {
           if (!isErrorAborted(e)) this.logger.error("Error on validator monitor onceEveryEndOfEpoch", {slot}, e);
         });
     }
@@ -1188,6 +1200,9 @@ export class BeaconChain implements IBeaconChain {
 
     if (headState) {
       this.opPool.pruneAll(headBlock, headState);
+      // Update custody requirement based on finalized state
+      const validatorIndices = this.localValidatorRegistry.getLocalValidatorIndices();
+      this.custodyConfig.updateCustodyRequirement(headState, validatorIndices);
     }
 
     if (headState === null) {

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

@@ -61,6 +61,8 @@ 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";
+import {LocalValidatorRegistry} from "../node/localValidatorRegistry.js";
 
 export {BlockType, type AssembledBlockType};
 export {type ProposerPreparationData};
@@ -91,6 +93,7 @@ export interface IBeaconChain {
   readonly logger: Logger;
   readonly metrics: Metrics | null;
   readonly bufferPool: BufferPool | null;
+  readonly localValidatorRegistry: LocalValidatorRegistry;
 
   /** The initial slot that the chain is started with */
   readonly anchorStateLatestBlockSlot: Slot;
@@ -113,6 +116,9 @@ export interface IBeaconChain {
   readonly syncContributionAndProofPool: SyncContributionAndProofPool;
   readonly opPool: OpPool;
 
+  // Data column custody config
+  readonly custodyConfig: CustodyConfig;
+
   // Gossip seen cache
   readonly seenAttesters: SeenAttesters;
   readonly seenAggregators: SeenAggregators;

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

@@ -247,15 +247,13 @@ export class SeenGossipBlockInput {
           };
         }
 
+        const sampledColumns = this.custodyConfig.getSampledColumns();
         const sampledIndexesPresent =
-          dataColumnsCache.size >= this.custodyConfig.sampledColumns.length &&
-          this.custodyConfig.sampledColumns.reduce(
-            (acc, columnIndex) => acc && dataColumnsCache.has(columnIndex),
-            true
-          );
+          dataColumnsCache.size >= sampledColumns.length &&
+          sampledColumns.reduce((acc, columnIndex) => acc && dataColumnsCache.has(columnIndex), true);
 
         if (sampledIndexesPresent) {
-          const allDataColumns = getBlockInputDataColumns(dataColumnsCache, this.custodyConfig.sampledColumns);
+          const allDataColumns = getBlockInputDataColumns(dataColumnsCache, sampledColumns);
           metrics?.syncUnknownBlock.resolveAvailabilitySource.inc({source: BlockInputAvailabilitySource.GOSSIP});
           const {dataColumns} = allDataColumns;
           const blockData: BlockInputDataColumns = {
@@ -274,7 +272,7 @@ export class SeenGossipBlockInput {
             blockInputMeta: {
               pending: null,
               haveColumns: dataColumns.length,
-              expectedColumns: this.custodyConfig.sampledColumns.length,
+              expectedColumns: sampledColumns.length,
             },
           };
         }
@@ -287,7 +285,7 @@ export class SeenGossipBlockInput {
           blockInputMeta: {
             pending: GossipedInputType.dataColumn,
             haveColumns: dataColumnsCache.size,
-            expectedColumns: this.custodyConfig.sampledColumns.length,
+            expectedColumns: sampledColumns.length,
           },
         };
       }

packages/beacon-node/src/metrics/validatorMonitor.ts

@@ -85,6 +85,7 @@ export type ValidatorMonitor = {
   ): void;
   onceEveryEndOfEpoch(state: CachedBeaconStateAllForks): void;
   scrapeMetrics(slotClock: Slot): void;
+  getLocalValidatorIndices(): ValidatorIndex[];
 };
 
 export type ValidatorMonitorOpts = {
@@ -789,6 +790,13 @@ export function createValidatorMonitor(
       metrics.validatorMonitor.prevEpochSyncCommitteeHits.set(prevEpochSyncCommitteeHits);
       metrics.validatorMonitor.prevEpochSyncCommitteeMisses.set(prevEpochSyncCommitteeMisses);
     },
+
+    /**
+     * Get the indices of the validators that are being monitored.
+     */
+    getLocalValidatorIndices() {
+      return Array.from(validators.keys());
+    },
   };
 }
 

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 {CustodyConfig} from "../../util/dataColumns.js";
 
 type Mods = {
   libp2p: Libp2p;
@@ -64,6 +65,8 @@ export type BaseNetworkInit = {
   getReqRespHandler: GetReqRespHandlerFn;
   activeValidatorCount: number;
   initialStatus: phase0.Status;
+  initialSamplingGroupCount: number;
+  initialAdvertisedGroupCount: number;
 };
 
 /**
@@ -135,6 +138,8 @@ export class NetworkCore implements INetworkCore {
     getReqRespHandler,
     activeValidatorCount,
     initialStatus,
+    initialSamplingGroupCount,
+    initialAdvertisedGroupCount,
   }: BaseNetworkInit): Promise<NetworkCore> {
     const libp2p = await createNodeJsLibp2p(peerId, opts, {
       peerStoreDir,
@@ -154,7 +159,7 @@ export class NetworkCore implements INetworkCore {
     const onMetadataSetValue = function onMetadataSetValue(key: string, value: Uint8Array): void {
       discv5?.setEnrValue(key, value).catch((e) => logger.error("error on setEnrValue", {key}, e));
     };
-    const metadata = new MetadataController({}, {config, onSetValue: onMetadataSetValue});
+    const metadata = new MetadataController({}, {config, onSetValue: onMetadataSetValue, initialAdvertisedGroupCount});
 
     const reqResp = new ReqRespBeaconNode(
       {
@@ -213,6 +218,7 @@ export class NetworkCore implements INetworkCore {
         events,
         peersData,
         statusCache,
+        initialSamplingGroupCount,
       },
       opts
     );
@@ -342,6 +348,14 @@ export class NetworkCore implements INetworkCore {
     return recipients.length;
   }
 
+  async setSamplingGroupCount(count: number): Promise<void> {
+    this.peerManager.setSamplingGroupCount(count);
+  }
+
+  async setAdvertisedGroupCount(count: number): Promise<void> {
+    this.metadata.cgc = count;
+  }
+
   // REST API queries
 
   async getNetworkIdentity(): Promise<routes.node.NetworkIdentity> {

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

@@ -102,6 +102,8 @@ const core = await NetworkCore.init({
     reqRespBridgeRespCaller.getAsyncIterable({method, req, peerId: peerIdToString(peerId)}),
   activeValidatorCount: workerData.activeValidatorCount,
   initialStatus: workerData.initialStatus,
+  initialSamplingGroupCount: workerData.initialSamplingGroupCount,
+  initialAdvertisedGroupCount: workerData.initialAdvertisedGroupCount,
 });
 
 wireEventsOnWorkerThread<NetworkEventData>(
@@ -138,6 +140,9 @@ const libp2pWorkerApi: NetworkWorkerApi = {
   // sendReqRespRequest - handled via events with AsyncIterableBridgeHandler
   publishGossip: (topic, data, opts) => core.publishGossip(topic, data, opts),
 
+  setSamplingGroupCount: (count) => core.setSamplingGroupCount(count),
+  setAdvertisedGroupCount: (count) => core.setAdvertisedGroupCount(count),
+
   // Debug
 
   getNetworkIdentity: () => core.getNetworkIdentity(),

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

@@ -38,6 +38,8 @@ export type WorkerNetworkCoreOpts = NetworkOptions & {
   activeValidatorCount: number;
   genesisTime: number;
   initialStatus: phase0.Status;
+  initialSamplingGroupCount: number;
+  initialAdvertisedGroupCount: number;
 };
 
 export type WorkerNetworkCoreInitModules = {
@@ -104,7 +106,16 @@ export class WorkerNetworkCore implements INetworkCore {
 
   static async init(modules: WorkerNetworkCoreInitModules): Promise<WorkerNetworkCore> {
     const {opts, config, peerId} = modules;
-    const {genesisTime, peerStoreDir, activeValidatorCount, localMultiaddrs, metricsEnabled, initialStatus} = opts;
+    const {
+      genesisTime,
+      peerStoreDir,
+      activeValidatorCount,
+      localMultiaddrs,
+      metricsEnabled,
+      initialStatus,
+      initialSamplingGroupCount,
+      initialAdvertisedGroupCount,
+    } = opts;
 
     const workerData: NetworkWorkerData = {
       opts,
@@ -116,6 +127,8 @@ export class WorkerNetworkCore implements INetworkCore {
       peerStoreDir,
       genesisTime,
       initialStatus,
+      initialSamplingGroupCount,
+      initialAdvertisedGroupCount,
       activeValidatorCount,
       loggerOpts: modules.logger.toOpts(),
     };
@@ -217,6 +230,15 @@ export class WorkerNetworkCore implements INetworkCore {
     return this.getApi().publishGossip(topic, data, opts);
   }
 
+  // Custody
+
+  setSamplingGroupCount(count: number): Promise<void> {
+    return this.getApi().setSamplingGroupCount(count);
+  }
+  setAdvertisedGroupCount(count: number): Promise<void> {
+    return this.getApi().setAdvertisedGroupCount(count);
+  }
+
   // Debug
 
   connectToPeer(peer: PeerIdStr, multiaddr: MultiaddrStr[]): Promise<void> {

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

@@ -54,6 +54,8 @@ export interface INetworkCore extends INetworkCorePublic {
   /** Chain must push status updates to the network core */
   updateStatus(status: phase0.Status): Promise<void>;
 
+  setSamplingGroupCount(count: number): Promise<void>;
+  setAdvertisedGroupCount(count: number): Promise<void>;
   /** Opens stream to handle ReqResp outgoing request */
   sendReqRespRequest(data: OutgoingRequestArgs): AsyncIterable<ResponseIncoming>;
   /** Publish gossip message to peers */
@@ -78,6 +80,8 @@ export type NetworkWorkerData = {
   genesisTime: number;
   activeValidatorCount: number;
   initialStatus: phase0.Status;
+  initialSamplingGroupCount: number;
+  initialAdvertisedGroupCount: number;
   peerIdProto: Uint8Array;
   localMultiaddrs: string[];
   metricsEnabled: boolean;
@@ -101,6 +105,9 @@ export type NetworkWorkerApi = INetworkCorePublic & {
   getConnectedPeerCount(): Promise<number>;
   updateStatus(status: phase0.Status): Promise<void>;
 
+  setSamplingGroupCount(count: number): Promise<void>;
+  setAdvertisedGroupCount(count: number): Promise<void>;
+
   // sendReqRespRequest - implemented via events
   publishGossip(topic: string, data: Uint8Array, opts?: PublishOpts): Promise<number>;