feat: implement peerDAS on electra

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 rsync 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 rsync 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,
+          // custodyColumns 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/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);

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

@@ -1,6 +1,6 @@
 import {CachedBeaconStateAllForks, computeEpochAtSlot} from "@lodestar/state-transition";
 import {MaybeValidExecutionStatus, DataAvailabilityStatus} from "@lodestar/fork-choice";
-import {deneb, Slot, RootHex, SignedBeaconBlock} from "@lodestar/types";
+import {deneb, Slot, RootHex, SignedBeaconBlock, peerdas, ColumnIndex} from "@lodestar/types";
 import {ForkSeq, ForkName} from "@lodestar/params";
 import {ChainForkConfig} from "@lodestar/config";
 
@@ -29,23 +29,48 @@ export enum BlobsSource {
   byRoot = "req_resp_by_root",
 }
 
+export enum DataColumnsSource {
+  gossip = "gossip",
+  api = "api",
+  byRange = "req_resp_by_range",
+  byRoot = "req_resp_by_root",
+}
+
 export enum GossipedInputType {
   block = "block",
   blob = "blob",
+  dataColumn = "dataColumn",
 }
 
-type BlobsCacheMap = Map<number, {blobSidecar: deneb.BlobSidecar; blobBytes: Uint8Array | null}>;
+export type BlobsCacheMap = Map<number, {blobSidecar: deneb.BlobSidecar; blobBytes: Uint8Array | null}>;
+export type DataColumnsCacheMap = Map<
+  number,
+  {dataColumnSidecar: peerdas.DataColumnSidecar; dataColumnBytes: Uint8Array | null}
+>;
 
 type ForkBlobsInfo = {fork: ForkName.deneb};
 type BlobsData = {blobs: deneb.BlobSidecars; blobsBytes: (Uint8Array | null)[]; blobsSource: BlobsSource};
 export type BlockInputDataBlobs = ForkBlobsInfo & BlobsData;
-export type BlockInputData = BlockInputDataBlobs;
 
-export type BlockInputBlobs = {blobs: deneb.BlobSidecars; blobsBytes: (Uint8Array | null)[]; blobsSource: BlobsSource};
-type Availability<T> = {availabilityPromise: Promise<T>; resolveAvailability: (data: T) => void};
+type ForkDataColumnsInfo = {fork: ForkName.peerdas};
+type DataColumnsData = {
+  // marker of that columns are to be custodied
+  dataColumnsLen: number;
+  dataColumnsIndex: Uint8Array;
+  dataColumns: peerdas.DataColumnSidecars;
+  dataColumnsBytes: (Uint8Array | null)[];
+  dataColumnsSource: DataColumnsSource;
+};
+export type BlockInputDataDataColumns = ForkDataColumnsInfo & DataColumnsData;
+export type BlockInputData = BlockInputDataBlobs | BlockInputDataDataColumns;
 
+type Availability<T> = {availabilityPromise: Promise<T>; resolveAvailability: (data: T) => void};
 type CachedBlobs = {blobsCache: BlobsCacheMap} & Availability<BlockInputDataBlobs>;
-export type CachedData = ForkBlobsInfo & CachedBlobs;
+type CachedDataColumns = {dataColumnsCache: DataColumnsCacheMap} & Availability<BlockInputDataDataColumns>;
+export type CachedData = {cacheId: number} & (
+  | (ForkBlobsInfo & CachedBlobs)
+  | (ForkDataColumnsInfo & CachedDataColumns)
+);
 
 export type BlockInput = {block: SignedBeaconBlock; source: BlockSource; blockBytes: Uint8Array | null} & (
   | {type: BlockInputType.preData | BlockInputType.outOfRangeData}
@@ -161,6 +186,26 @@ export function getBlockInputBlobs(blobsCache: BlobsCacheMap): Omit<BlobsData, "
   return {blobs, blobsBytes};
 }
 
+export function getBlockInputDataColumns(
+  dataColumnsCache: DataColumnsCacheMap,
+  columnIndexes: ColumnIndex[]
+): Omit<DataColumnsData, "dataColumnsLen" | "dataColumnsIndex" | "dataColumnsSource"> {
+  const dataColumns = [];
+  const dataColumnsBytes = [];
+
+  for (const index of columnIndexes) {
+    const dataColumnCache = dataColumnsCache.get(index);
+    if (dataColumnCache === undefined) {
+      // check if the index is correct as per the custody columns
+      throw Error(`Missing dataColumnCache at index=${index}`);
+    }
+    const {dataColumnSidecar, dataColumnBytes} = dataColumnCache;
+    dataColumns.push(dataColumnSidecar);
+    dataColumnsBytes.push(dataColumnBytes);
+  }
+  return {dataColumns, dataColumnsBytes};
+}
+
 export enum AttestationImportOpt {
   Skip,
   Force,

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

@@ -3,10 +3,19 @@ import {DataAvailabilityStatus} from "@lodestar/fork-choice";
 import {ChainForkConfig} from "@lodestar/config";
 import {deneb, UintNum64} from "@lodestar/types";
 import {Logger} from "@lodestar/utils";
+import {ForkName} from "@lodestar/params";
 import {BlockError, BlockErrorCode} from "../errors/index.js";
 import {validateBlobSidecars} from "../validation/blobSidecar.js";
+import {validateDataColumnsSidecars} from "../validation/dataColumnSidecar.js";
 import {Metrics} from "../../metrics/metrics.js";
-import {BlockInput, BlockInputType, ImportBlockOpts, BlobSidecarValidation, getBlockInput} from "./types.js";
+import {
+  BlockInput,
+  BlockInputType,
+  ImportBlockOpts,
+  BlobSidecarValidation,
+  getBlockInput,
+  BlockInputData,
+} from "./types.js";
 
 // we can now wait for full 12 seconds because unavailable block sync will try pulling
 // the blobs from the network anyway after 500ms of seeing the block
@@ -88,27 +97,37 @@ async function maybeValidateBlobs(
       // run full validation
       const {block} = blockInput;
       const blockSlot = block.message.slot;
-
-      const blobsData =
-        blockInput.type === BlockInputType.availableData
-          ? blockInput.blockData
-          : await raceWithCutoff(chain, blockInput, blockInput.cachedData.availabilityPromise);
-      const {blobs} = blobsData;
-
       const {blobKzgCommitments} = (block as deneb.SignedBeaconBlock).message.body;
       const beaconBlockRoot = chain.config.getForkTypes(blockSlot).BeaconBlock.hashTreeRoot(block.message);
-
-      // if the blob siddecars have been individually verified then we can skip kzg proof check
-      // but other checks to match blobs with block data still need to be performed
-      const skipProofsCheck = opts.validBlobSidecars === BlobSidecarValidation.Individual;
-      validateBlobSidecars(blockSlot, beaconBlockRoot, blobKzgCommitments, blobs, {skipProofsCheck});
+      const blockData =
+        blockInput.type === BlockInputType.availableData
+          ? blockInput.blockData
+          : await raceWithCutoff(
+              chain,
+              blockInput,
+              blockInput.cachedData.availabilityPromise as Promise<BlockInputData>
+            );
+
+      if (blockData.fork === ForkName.deneb) {
+        const {blobs} = blockData;
+
+        // if the blob siddecars have been individually verified then we can skip kzg proof check
+        // but other checks to match blobs with block data still need to be performed
+        const skipProofsCheck = opts.validBlobSidecars === BlobSidecarValidation.Individual;
+        validateBlobSidecars(blockSlot, beaconBlockRoot, blobKzgCommitments, blobs, {skipProofsCheck});
+      } else if (blockData.fork === ForkName.peerdas) {
+        const {dataColumns} = blockData;
+        const skipProofsCheck = opts.validBlobSidecars === BlobSidecarValidation.Individual;
+        // might require numColumns, custodyColumns from blockData as input to below
+        validateDataColumnsSidecars(blockSlot, beaconBlockRoot, blobKzgCommitments, dataColumns, {skipProofsCheck});
+      }
 
       const availableBlockInput = getBlockInput.availableData(
         chain.config,
         blockInput.block,
         blockInput.source,
         blockInput.blockBytes,
-        blobsData
+        blockData
       );
       return {dataAvailabilityStatus: DataAvailabilityStatus.Available, availableBlockInput: availableBlockInput};
     }