feat: fulu hard-fork implementation

Dockerfile

@@ -4,7 +4,7 @@
 FROM --platform=${BUILDPLATFORM:-amd64} node:22-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-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

@@ -133,7 +133,7 @@
     "@lodestar/utils": "^1.25.0",
     "@lodestar/validator": "^1.25.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

@@ -1,40 +1,51 @@
 import {routes} from "@lodestar/api";
 import {ApiError, ApplicationMethods} from "@lodestar/api/server";
-import {ForkExecution, SLOTS_PER_HISTORICAL_ROOT, isForkExecution, isForkPostElectra} from "@lodestar/params";
+import {
+  ForkExecution,
+  SLOTS_PER_HISTORICAL_ROOT,
+  isForkExecution,
+  isForkPostElectra,
+  ForkName,
+  isForkPostFulu,
+} from "@lodestar/params";
 import {
   computeEpochAtSlot,
   computeTimeAtSlot,
   reconstructFullBlockOrContents,
   signedBeaconBlockToBlinded,
 } from "@lodestar/state-transition";
+import {sleep, fromHex, toHex, toRootHex} from "@lodestar/utils";
 import {
+  deneb,
+  fulu,
+  isSignedBlockContents,
   ProducedBlockSource,
   SignedBeaconBlock,
   SignedBeaconBlockOrContents,
   SignedBlindedBeaconBlock,
   WithOptionalBytes,
-  deneb,
-  isSignedBlockContents,
 } from "@lodestar/types";
-import {fromHex, sleep, toRootHex} from "@lodestar/utils";
 import {
   BlobsSource,
   BlockInput,
-  BlockInputDataBlobs,
   BlockSource,
   ImportBlockOpts,
   getBlockInput,
+  BlockInputBlobs,
+  BlockInputDataColumns,
+  DataColumnsSource,
+  BlockInputAvailableData,
 } from "../../../../chain/blocks/types.js";
-import {verifyBlocksInEpoch} from "../../../../chain/blocks/verifyBlock.js";
-import {BeaconChain} from "../../../../chain/chain.js";
+import {promiseAllMaybeAsync} from "../../../../util/promises.js";
+import {isOptimisticBlock} from "../../../../util/forkChoice.js";
+import {computeBlobSidecars, computeDataColumnSidecars} from "../../../../util/blobs.js";
 import {BlockError, BlockErrorCode, BlockGossipError} from "../../../../chain/errors/index.js";
-import {validateGossipBlock} from "../../../../chain/validation/block.js";
 import {OpSource} from "../../../../metrics/validatorMonitor.js";
 import {NetworkEvent} from "../../../../network/index.js";
-import {computeBlobSidecars} from "../../../../util/blobs.js";
-import {isOptimisticBlock} from "../../../../util/forkChoice.js";
-import {promiseAllMaybeAsync} from "../../../../util/promises.js";
 import {ApiModules} from "../../types.js";
+import {validateGossipBlock} from "../../../../chain/validation/block.js";
+import {verifyBlocksInEpoch} from "../../../../chain/blocks/verifyBlock.js";
+import {BeaconChain} from "../../../../chain/chain.js";
 import {getBlockResponse, toBeaconHeaderResponse} from "./utils.js";
 
 type PublishBlockOpts = ImportBlockOpts;
@@ -66,17 +77,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: fulu.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: BlockInputAvailableData;
+      if (isForkPostFulu(fork)) {
+        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 BlockInputDataColumns;
+        blobSidecars = [];
+      } else if (fork === ForkName.deneb || fork === ForkName.electra) {
+        blobSidecars = computeBlobSidecars(config, signedBlock, signedBlockOrContents);
+        blockData = {
+          fork,
+          blobs: blobSidecars,
+          blobsSource: BlobsSource.api,
+          blobsBytes: blobSidecars.map(() => null),
+        } as BlockInputBlobs;
+        dataColumnSidecars = [];
+      } else {
+        throw Error(`Invalid data fork=${fork} for publish`);
+      }
+
       blockForImport = getBlockInput.availableData(
         config,
         signedBlock,
@@ -88,6 +122,7 @@ export function getBeaconBlockApi({
     } else {
       signedBlock = signedBlockOrContents;
       blobSidecars = [];
+      dataColumnSidecars = [];
       blockForImport = getBlockInput.preData(config, signedBlock, BlockSource.api, context?.sszBytes ?? null);
     }
 
@@ -215,11 +250,12 @@ export function getBeaconBlockApi({
       // specification is very clear that this is the desired behaviour.
       //
       // i) Publish blobs and block before importing so that network can see them asap
-      // ii) publish block first because
-      //     a) as soon as node sees block they can start processing it while blobs arrive
-      //     b) getting block first allows nodes to use getBlobs from local ELs and save
-      //        import latency and hopefully bandwidth
+      // ii) publish blobs first because
+      //     a) by the times nodes see block, they might decide to pull blobs
+      //     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
       () => network.publishBeaconBlock(signedBlock) as Promise<unknown>,
+      ...dataColumnSidecars.map((dataColumnSidecar) => () => network.publishDataColumnSidecar(dataColumnSidecar)),
       ...blobSidecars.map((blobSidecar) => () => network.publishBlobSidecar(blobSidecar)),
       () =>
         // there is no rush to persist block since we published it to gossip anyway

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

@@ -4,7 +4,7 @@ import {IForkChoice} from "@lodestar/fork-choice";
 import {ForkSeq, SLOTS_PER_EPOCH} from "@lodestar/params";
 import {computeEpochAtSlot, computeStartSlotAtEpoch} from "@lodestar/state-transition";
 import {Epoch, RootHex, Slot} from "@lodestar/types";
-import {Logger, fromHex, toRootHex} from "@lodestar/utils";
+import {Logger, fromHex, toRootHex, toHex} from "@lodestar/utils";
 import {IBeaconDb} from "../../db/index.js";
 import {BlockArchiveBatchPutBinaryItem} from "../../db/repositories/index.js";
 import {LightClientServer} from "../lightClient/index.js";
@@ -44,6 +44,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 finalizedPostFulu = finalizedCheckpoint.epoch >= config.FULU_FORK_EPOCH;
 
   const finalizedCanonicalBlockRoots: BlockRootSlot[] = finalizedCanonicalBlocks.map((block) => ({
     slot: block.slot,
@@ -59,8 +60,13 @@ export async function archiveBlocks(
     });
 
     if (finalizedPostDeneb) {
-      const migrate = await migrateBlobSidecarsFromHotToColdDb(config, db, finalizedCanonicalBlockRoots, currentEpoch);
-      logger.verbose(migrate ? "Migrated blobSidecars from hot DB to cold DB" : "Skip blobSidecars migration");
+      const migratedEntries = await migrateBlobSidecarsFromHotToColdDb(config, db, finalizedCanonicalBlockRoots);
+      logger.verbose("Migrated blobSidecars from hot DB to cold DB", {migratedEntries});
+    }
+
+    if (finalizedPostFulu) {
+      const migratedEntries = await migrateDataColumnSidecarsFromHotToColdDb(config, db, finalizedCanonicalBlockRoots);
+      logger.verbose("Migrated dataColumnSidecars from hot DB to cold DB", {migratedEntries});
     }
   }
 
@@ -76,7 +82,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 (finalizedPostFulu) {
+      await db.dataColumnSidecars.batchDelete(nonCanonicalBlockRoots);
+      logger.verbose("Deleted non canonical dataColumnSidecars from hot DB");
     }
   }
 
@@ -103,6 +114,11 @@ export async function archiveBlocks(
     }
   }
 
+  if (finalizedPostFulu) {
+    // 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];
@@ -164,28 +180,22 @@ async function migrateBlocksFromHotToColdDb(db: IBeaconDb, blocks: BlockRootSlot
 async function migrateBlobSidecarsFromHotToColdDb(
   config: ChainForkConfig,
   db: IBeaconDb,
-  blocks: BlockRootSlot[],
-  currentEpoch: Epoch
-): Promise<boolean> {
-  let result = false;
-
+  blocks: BlockRootSlot[]
+): 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 false;
+    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) => {
-          const blockSlot = block.slot;
-          const blockEpoch = computeEpochAtSlot(blockSlot);
-          return (
-            config.getForkSeq(blockSlot) >= ForkSeq.deneb &&
-            blockEpoch >= currentEpoch - config.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS
-          );
+          const blkSeq = config.getForkSeq(block.slot);
+          return blkSeq >= ForkSeq.deneb && blkSeq < ForkSeq.fulu;
         })
         .map(async (block) => {
           const bytes = await db.blobSidecars.getBinary(block.root);
@@ -196,20 +206,52 @@ async function migrateBlobSidecarsFromHotToColdDb(
         })
     );
 
-    const migrate = canonicalBlobSidecarsEntries.length > 0;
+    // put to blockArchive db and delete block db
+    await Promise.all([
+      db.blobSidecarsArchive.batchPutBinary(canonicalBlobSidecarsEntries),
+      db.blobSidecars.batchDelete(canonicalBlocks.map((block) => block.root)),
+    ]);
+    migratedWrappedBlobSidecars += canonicalBlobSidecarsEntries.length;
+  }
 
-    if (migrate) {
-      // put to blockArchive db and delete block db
-      await Promise.all([
-        db.blobSidecarsArchive.batchPutBinary(canonicalBlobSidecarsEntries),
-        db.blobSidecars.batchDelete(canonicalBlocks.map((block) => block.root)),
-      ]);
-    }
+  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;
 
-    result = result || migrate;
+    // 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.fulu)
+        .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 result;
+  return migratedWrappedDataColumns;
 }
 
 /**