feat: peerdas - ensure there are at least n peers per sampling column subnet

Dockerfile

@@ -4,7 +4,7 @@
 FROM --platform=${BUILDPLATFORM:-amd64} node:22.4-slim as build_src
 ARG COMMIT
 WORKDIR /usr/app
-RUN apt-get update && apt-get install -y g++ make python3 python3-setuptools && apt-get clean && rm -rf /var/lib/apt/lists/*
+RUN apt-get update && apt-get install -y git g++ make python3 python3-setuptools && apt-get clean && rm -rf /var/lib/apt/lists/*
 
 COPY . .
 
@@ -23,7 +23,7 @@ RUN cd packages/cli && GIT_COMMIT=${COMMIT} yarn write-git-data
 # Note: This step is redundant for the host arch
 FROM node:22.4-slim as build_deps
 WORKDIR /usr/app
-RUN apt-get update && apt-get install -y g++ make python3 python3-setuptools && apt-get clean && rm -rf /var/lib/apt/lists/*
+RUN apt-get update && apt-get install -y git g++ make python3 python3-setuptools && apt-get clean && rm -rf /var/lib/apt/lists/*
 
 COPY --from=build_src /usr/app .
 

packages/beacon-node/package.json

@@ -132,7 +132,7 @@
     "@lodestar/utils": "^1.21.0",
     "@lodestar/validator": "^1.21.0",
     "@multiformats/multiaddr": "^12.1.3",
-    "c-kzg": "^2.1.2",
+    "c-kzg": "^4.0.1",
     "datastore-core": "^9.1.1",
     "datastore-level": "^10.1.1",
     "deepmerge": "^4.3.1",

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

@@ -6,9 +6,10 @@ import {
   reconstructFullBlockOrContents,
   signedBeaconBlockToBlinded,
 } from "@lodestar/state-transition";
-import {ForkExecution, SLOTS_PER_HISTORICAL_ROOT, isForkExecution} from "@lodestar/params";
+import {ForkExecution, SLOTS_PER_HISTORICAL_ROOT, isForkExecution, ForkName} from "@lodestar/params";
 import {sleep, fromHex, toHex} from "@lodestar/utils";
 import {
+  peerdas,
   deneb,
   isSignedBlockContents,
   ProducedBlockSource,
@@ -23,10 +24,13 @@ import {
   BlockInput,
   BlobsSource,
   BlockInputDataBlobs,
+  BlockInputDataDataColumns,
+  DataColumnsSource,
+  BlockInputData,
 } from "../../../../chain/blocks/types.js";
 import {promiseAllMaybeAsync} from "../../../../util/promises.js";
 import {isOptimisticBlock} from "../../../../util/forkChoice.js";
-import {computeBlobSidecars} from "../../../../util/blobs.js";
+import {computeBlobSidecars, computeDataColumnSidecars} from "../../../../util/blobs.js";
 import {BlockError, BlockErrorCode, BlockGossipError} from "../../../../chain/errors/index.js";
 import {OpSource} from "../../../../metrics/validatorMonitor.js";
 import {NetworkEvent} from "../../../../network/index.js";
@@ -65,17 +69,40 @@ export function getBeaconBlockApi({
     opts: PublishBlockOpts = {}
   ) => {
     const seenTimestampSec = Date.now() / 1000;
-    let blockForImport: BlockInput, signedBlock: SignedBeaconBlock, blobSidecars: deneb.BlobSidecars;
+    let blockForImport: BlockInput,
+      signedBlock: SignedBeaconBlock,
+      blobSidecars: deneb.BlobSidecars,
+      dataColumnSidecars: peerdas.DataColumnSidecars;
 
     if (isSignedBlockContents(signedBlockOrContents)) {
       ({signedBlock} = signedBlockOrContents);
-      blobSidecars = computeBlobSidecars(config, signedBlock, signedBlockOrContents);
-      const blockData = {
-        fork: config.getForkName(signedBlock.message.slot),
-        blobs: blobSidecars,
-        blobsSource: BlobsSource.api,
-        blobsBytes: blobSidecars.map(() => null),
-      } as BlockInputDataBlobs;
+      const fork = config.getForkName(signedBlock.message.slot);
+      let blockData: BlockInputData;
+      if (fork === ForkName.peerdas) {
+        dataColumnSidecars = computeDataColumnSidecars(config, signedBlock, signedBlockOrContents);
+        blockData = {
+          fork,
+          dataColumnsLen: dataColumnSidecars.length,
+          // dataColumnsIndex is a 1 based index of ith column present in dataColumns[custodyColumns[i-1]]
+          dataColumnsIndex: new Uint8Array(Array.from({length: dataColumnSidecars.length}, (_, j) => 1 + j)),
+          dataColumns: dataColumnSidecars,
+          dataColumnsBytes: dataColumnSidecars.map(() => null),
+          dataColumnsSource: DataColumnsSource.api,
+        } as BlockInputDataDataColumns;
+        blobSidecars = [];
+      } else if (fork === ForkName.deneb) {
+        blobSidecars = computeBlobSidecars(config, signedBlock, signedBlockOrContents);
+        blockData = {
+          fork,
+          blobs: blobSidecars,
+          blobsSource: BlobsSource.api,
+          blobsBytes: blobSidecars.map(() => null),
+        } as BlockInputDataBlobs;
+        dataColumnSidecars = [];
+      } else {
+        throw Error(`Invalid data fork=${fork} for publish`);
+      }
+
       blockForImport = getBlockInput.availableData(
         config,
         signedBlock,
@@ -87,6 +114,7 @@ export function getBeaconBlockApi({
     } else {
       signedBlock = signedBlockOrContents;
       blobSidecars = [];
+      dataColumnSidecars = [];
       blockForImport = getBlockInput.preData(config, signedBlock, BlockSource.api, context?.sszBytes ?? null);
     }
 
@@ -221,6 +249,7 @@ export function getBeaconBlockApi({
       //     b) they might require more hops to reach recipients in peerDAS kind of setup where
       //        blobs might need to hop between nodes because of partial subnet subscription
       ...blobSidecars.map((blobSidecar) => () => network.publishBlobSidecar(blobSidecar)),
+      ...dataColumnSidecars.map((dataColumnSidecar) => () => network.publishDataColumnSidecar(dataColumnSidecar)),
       () => network.publishBeaconBlock(signedBlock) as Promise<unknown>,
       () =>
         // there is no rush to persist block since we published it to gossip anyway

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

@@ -45,6 +45,7 @@ export async function archiveBlocks(
 
   // NOTE: The finalized block will be exactly the first block of `epoch` or previous
   const finalizedPostDeneb = finalizedCheckpoint.epoch >= config.DENEB_FORK_EPOCH;
+  const finalizedPostPeerDAS = finalizedCheckpoint.epoch >= config.PEERDAS_FORK_EPOCH;
 
   const finalizedCanonicalBlockRoots: BlockRootSlot[] = finalizedCanonicalBlocks.map((block) => ({
     slot: block.slot,
@@ -60,8 +61,13 @@ export async function archiveBlocks(
     });
 
     if (finalizedPostDeneb) {
-      await migrateBlobSidecarsFromHotToColdDb(config, db, finalizedCanonicalBlockRoots);
-      logger.verbose("Migrated blobSidecars from hot DB to cold DB");
+      const migratedEntries = await migrateBlobSidecarsFromHotToColdDb(config, db, finalizedCanonicalBlockRoots);
+      logger.verbose("Migrated blobSidecars from hot DB to cold DB", {migratedEntries});
+    }
+
+    if (finalizedPostPeerDAS) {
+      const migratedEntries = await migrateDataColumnSidecarsFromHotToColdDb(config, db, finalizedCanonicalBlockRoots);
+      logger.verbose("Migrated dataColumnSidecars from hot DB to cold DB", {migratedEntries});
     }
   }
 
@@ -77,7 +83,12 @@ export async function archiveBlocks(
 
     if (finalizedPostDeneb) {
       await db.blobSidecars.batchDelete(nonCanonicalBlockRoots);
-      logger.verbose("Deleted non canonical blobsSider from hot DB");
+      logger.verbose("Deleted non canonical blobSidecars from hot DB");
+    }
+
+    if (finalizedPostPeerDAS) {
+      await db.dataColumnSidecars.batchDelete(nonCanonicalBlockRoots);
+      logger.verbose("Deleted non canonical dataColumnSidecars from hot DB");
     }
   }
 
@@ -104,6 +115,11 @@ export async function archiveBlocks(
     }
   }
 
+  if (finalizedPostPeerDAS) {
+    // TODO
+    // Keep only `[current_epoch - max(MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS, archiveBlobEpochs)]
+  }
+
   // Prunning potential checkpoint data
   const finalizedCanonicalNonCheckpointBlocks = getNonCheckpointBlocks(finalizedCanonicalBlockRoots);
   const nonCheckpointBlockRoots: Uint8Array[] = [...nonCanonicalBlockRoots];
@@ -162,18 +178,22 @@ async function migrateBlobSidecarsFromHotToColdDb(
   config: ChainForkConfig,
   db: IBeaconDb,
   blocks: BlockRootSlot[]
-): Promise<void> {
+): Promise<number> {
+  let migratedWrappedBlobSidecars = 0;
   for (let i = 0; i < blocks.length; i += BLOB_SIDECAR_BATCH_SIZE) {
     const toIdx = Math.min(i + BLOB_SIDECAR_BATCH_SIZE, blocks.length);
     const canonicalBlocks = blocks.slice(i, toIdx);
 
     // processCanonicalBlocks
-    if (canonicalBlocks.length === 0) return;
+    if (canonicalBlocks.length === 0) break;
 
     // load Buffer instead of ssz deserialized to improve performance
     const canonicalBlobSidecarsEntries: KeyValue<Slot, Uint8Array>[] = await Promise.all(
       canonicalBlocks
-        .filter((block) => config.getForkSeq(block.slot) >= ForkSeq.deneb)
+        .filter((block) => {
+          const blkSeq = config.getForkSeq(block.slot);
+          return blkSeq >= ForkSeq.deneb && blkSeq < ForkSeq.peerdas;
+        })
         .map(async (block) => {
           const bytes = await db.blobSidecars.getBinary(block.root);
           if (!bytes) {
@@ -188,7 +208,47 @@ async function migrateBlobSidecarsFromHotToColdDb(
       db.blobSidecarsArchive.batchPutBinary(canonicalBlobSidecarsEntries),
       db.blobSidecars.batchDelete(canonicalBlocks.map((block) => block.root)),
     ]);
+    migratedWrappedBlobSidecars += canonicalBlobSidecarsEntries.length;
   }
+
+  return migratedWrappedBlobSidecars;
+}
+
+async function migrateDataColumnSidecarsFromHotToColdDb(
+  config: ChainForkConfig,
+  db: IBeaconDb,
+  blocks: BlockRootSlot[]
+): Promise<number> {
+  let migratedWrappedDataColumns = 0;
+  for (let i = 0; i < blocks.length; i += BLOB_SIDECAR_BATCH_SIZE) {
+    const toIdx = Math.min(i + BLOB_SIDECAR_BATCH_SIZE, blocks.length);
+    const canonicalBlocks = blocks.slice(i, toIdx);
+
+    // processCanonicalBlocks
+    if (canonicalBlocks.length === 0) break;
+
+    // load Buffer instead of ssz deserialized to improve performance
+    const canonicalDataColumnSidecarsEntries: KeyValue<Slot, Uint8Array>[] = await Promise.all(
+      canonicalBlocks
+        .filter((block) => config.getForkSeq(block.slot) >= ForkSeq.peerdas)
+        .map(async (block) => {
+          const bytes = await db.dataColumnSidecars.getBinary(block.root);
+          if (!bytes) {
+            throw Error(`No dataColumnSidecars found for slot ${block.slot} root ${toHex(block.root)}`);
+          }
+          return {key: block.slot, value: bytes};
+        })
+    );
+
+    // put to blockArchive db and delete block db
+    await Promise.all([
+      db.dataColumnSidecarsArchive.batchPutBinary(canonicalDataColumnSidecarsEntries),
+      db.dataColumnSidecars.batchDelete(canonicalBlocks.map((block) => block.root)),
+    ]);
+    migratedWrappedDataColumns += canonicalDataColumnSidecarsEntries.length;
+  }
+
+  return migratedWrappedDataColumns;
 }
 
 /**

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

@@ -1,6 +1,13 @@
 import {toHexString} from "@chainsafe/ssz";
 import {capella, ssz, altair, BeaconBlock} from "@lodestar/types";
-import {ForkLightClient, ForkSeq, INTERVALS_PER_SLOT, MAX_SEED_LOOKAHEAD, SLOTS_PER_EPOCH} from "@lodestar/params";
+import {
+  ForkName,
+  ForkLightClient,
+  ForkSeq,
+  INTERVALS_PER_SLOT,
+  MAX_SEED_LOOKAHEAD,
+  SLOTS_PER_EPOCH,
+} from "@lodestar/params";
 import {
   CachedBeaconStateAltair,
   computeEpochAtSlot,
@@ -101,6 +108,39 @@ export async function importBlock(
   this.metrics?.importBlock.bySource.inc({source});
   this.logger.verbose("Added block to forkchoice and state cache", {slot: blockSlot, root: blockRootHex});
 
+  // We want to import block asap so call all event handler in the next event loop
+  callInNextEventLoop(async () => {
+    this.emitter.emit(routes.events.EventType.block, {
+      block: blockRootHex,
+      slot: blockSlot,
+      executionOptimistic: blockSummary != null && isOptimisticBlock(blockSummary),
+    });
+
+    // dataPromise will not end up here, but preDeneb could. In future we might also allow syncing
+    // out of data range blocks and import then in forkchoice although one would not be able to
+    // attest and propose with such head similar to optimistic sync
+    if (blockInput.type === BlockInputType.availableData) {
+      const {blockData} = blockInput;
+      if (blockData.fork === ForkName.deneb) {
+        const {blobsSource, blobs} = blockData;
+
+        this.metrics?.importBlock.blobsBySource.inc({blobsSource});
+        for (const blobSidecar of blobs) {
+          const {index, kzgCommitment} = blobSidecar;
+          this.emitter.emit(routes.events.EventType.blobSidecar, {
+            blockRoot: blockRootHex,
+            slot: blockSlot,
+            index,
+            kzgCommitment: toHexString(kzgCommitment),
+            versionedHash: toHexString(kzgCommitmentToVersionedHash(kzgCommitment)),
+          });
+        }
+      } else if (blockData.fork === ForkName.peerdas) {
+        // TODO peerDAS build and emit the event for the datacolumns
+      }
+    }
+  });
+
   // 3. Import attestations to fork choice
   //
   // - For each attestation
@@ -424,16 +464,20 @@ export async function importBlock(
         blockInput.type === BlockInputType.availableData &&
         this.emitter.listenerCount(routes.events.EventType.blobSidecar)
       ) {
-        const {blobs} = blockInput.blockData;
-        for (const blobSidecar of blobs) {
-          const {index, kzgCommitment} = blobSidecar;
-          this.emitter.emit(routes.events.EventType.blobSidecar, {
-            blockRoot: blockRootHex,
-            slot: blockSlot,
-            index,
-            kzgCommitment: toHexString(kzgCommitment),
-            versionedHash: toHexString(kzgCommitmentToVersionedHash(kzgCommitment)),
-          });
+        if (blockInput.blockData.fork === ForkName.deneb) {
+          const {blobs} = blockInput.blockData;
+          for (const blobSidecar of blobs) {
+            const {index, kzgCommitment} = blobSidecar;
+            this.emitter.emit(routes.events.EventType.blobSidecar, {
+              blockRoot: blockRootHex,
+              slot: blockSlot,
+              index,
+              kzgCommitment: toHexString(kzgCommitment),
+              versionedHash: toHexString(kzgCommitmentToVersionedHash(kzgCommitment)),
+            });
+          }
+        } else {
+          // TODO add event for datacolumns
         }
       }
     });
@@ -454,8 +498,12 @@ export async function importBlock(
   // out of data range blocks and import then in forkchoice although one would not be able to
   // attest and propose with such head similar to optimistic sync
   if (blockInput.type === BlockInputType.availableData) {
-    const {blobsSource} = blockInput.blockData;
-    this.metrics?.importBlock.blobsBySource.inc({blobsSource});
+    if (blockInput.blockData.fork === ForkName.deneb) {
+      const {blobsSource} = blockInput.blockData;
+      this.metrics?.importBlock.blobsBySource.inc({blobsSource});
+    } else {
+      // TODO add data columns metrics
+    }
   }
 
   const advancedSlot = this.clock.slotWithFutureTolerance(REPROCESS_MIN_TIME_TO_NEXT_SLOT_SEC);