refactor: separate writeBlockInputToDb into parallel block and column writes

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

@@ -100,6 +100,18 @@ export type MissingColumnMeta = {
   versionedHashes: VersionedHashes;
 };
 
+/**
+ * Minimal interface required to write data columns to the DB.
+ * Used by `writeDataColumnsToDb` and designed to be reusable across forks (e.g. Fulu, Gloas).
+ */
+export interface IDataColumnsInput {
+  readonly slot: Slot;
+  readonly blockRootHex: string;
+  getCustodyColumns(): fulu.DataColumnSidecars;
+  hasComputedAllData(): boolean;
+  waitForComputedAllData(timeout: number, signal?: AbortSignal): Promise<fulu.DataColumnSidecars>;
+}
+
 /**
  * This is used to validate that BlockInput implementations follow some minimal subset of operations
  * and that adding a new implementation won't break consumers that rely on this subset.

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

@@ -95,7 +95,7 @@ export async function importBlock(
   // Without this, a supernode syncing from behind can accumulate many blocks worth of column
   // data in memory (up to 128 columns per block) causing OOM before persistence catches up.
   await this.unfinalizedBlockWrites.waitForSpace();
-  this.unfinalizedBlockWrites.push([blockInput]).catch((e) => {
+  this.unfinalizedBlockWrites.push(blockInput).catch((e) => {
     if (!isQueueErrorAborted(e)) {
       this.logger.error("Error pushing block to unfinalized write queue", {slot: blockSlot}, e as Error);
     }

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

@@ -1,7 +1,15 @@
-import {fulu} from "@lodestar/types";
-import {prettyPrintIndices, toRootHex} from "@lodestar/utils";
+import {ForkPostDeneb, isForkPostDeneb} from "@lodestar/params";
+import {SignedBeaconBlock} from "@lodestar/types";
+import {fromHex, toRootHex} from "@lodestar/utils";
+import {getBlobKzgCommitments} from "../../util/dataColumns.js";
 import {BeaconChain} from "../chain.js";
-import {IBlockInput, isBlockInputBlobs, isBlockInputColumns} from "./blockInput/index.js";
+import {
+  IBlockInput,
+  IDataColumnsInput,
+  isBlockInputBlobs,
+  isBlockInputColumns,
+  isBlockInputNoData,
+} from "./blockInput/index.js";
 import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js";
 
 /**
@@ -10,129 +18,139 @@ import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js";
  *
  * This operation may be performed before, during or after importing to the fork-choice. As long as errors
  * are handled properly for eventual consistency.
+ *
+ * Block+blobs (pre-fulu) and data columns (fulu+) are written in parallel.
  */
-export async function writeBlockInputToDb(this: BeaconChain, blocksInputs: IBlockInput[]): Promise<void> {
-  const fnPromises: Promise<void>[] = [];
-  // track slots for logging
-  const slots: number[] = [];
-
-  for (const blockInput of blocksInputs) {
-    const block = blockInput.getBlock();
-    const slot = block.message.slot;
-    slots.push(slot);
-    const blockRoot = this.config.getForkTypes(block.message.slot).BeaconBlock.hashTreeRoot(block.message);
-    const blockRootHex = toRootHex(blockRoot);
-    const blockBytes = this.serializedCache.get(block);
-    if (blockBytes) {
-      // skip serializing data if we already have it
-      this.metrics?.importBlock.persistBlockWithSerializedDataCount.inc();
-      fnPromises.push(this.db.block.putBinary(this.db.block.getId(block), blockBytes));
-    } else {
-      this.metrics?.importBlock.persistBlockNoSerializedDataCount.inc();
-      fnPromises.push(this.db.block.add(block));
-    }
+export async function writeBlockInputToDb(this: BeaconChain, blockInput: IBlockInput): Promise<void> {
+  const promises: Promise<void>[] = [writeBlockAndBlobsToDb.call(this, blockInput)];
 
-    this.logger.debug("Persist block to hot DB", {
-      slot: block.message.slot,
-      root: blockRootHex,
-      inputType: blockInput.type,
-    });
+  if (isBlockInputColumns(blockInput)) {
+    promises.push(writeDataColumnsToDb.call(this, blockInput));
+  }
 
-    if (!blockInput.hasAllData()) {
-      await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT);
-    }
+  await Promise.all(promises);
+  this.logger.debug("Persisted blockInput to db", {slot: blockInput.slot, root: blockInput.blockRootHex});
+}
 
-    // NOTE: Old data is pruned on archive
-    if (isBlockInputColumns(blockInput)) {
-      if (!blockInput.hasComputedAllData()) {
-        // Supernodes may only have a subset of the data columns by the time the block begins to be imported
-        // because full data availability can be assumed after NUMBER_OF_COLUMNS / 2 columns are available.
-        // Here, however, all data columns must be fully available/reconstructed before persisting to the DB.
-        await blockInput.waitForComputedAllData(BLOB_AVAILABILITY_TIMEOUT).catch(() => {
-          this.logger.debug("Failed to wait for computed all data", {slot, blockRoot: blockRootHex});
-        });
-      }
+async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput): Promise<void> {
+  const block = blockInput.getBlock();
+  const slot = block.message.slot;
+  const blockRoot = this.config.getForkTypes(slot).BeaconBlock.hashTreeRoot(block.message);
+  const blockRootHex = toRootHex(blockRoot);
+  const numBlobs = isForkPostDeneb(blockInput.forkName)
+    ? getBlobKzgCommitments(blockInput.forkName, block as SignedBeaconBlock<ForkPostDeneb>).length
+    : undefined;
+  const fnPromises: Promise<void>[] = [];
 
-      const {custodyColumns} = this.custodyConfig;
-      const blobsLen = (block.message as fulu.BeaconBlock).body.blobKzgCommitments.length;
-      let dataColumnsLen: number;
-      if (blobsLen === 0) {
-        dataColumnsLen = 0;
-      } else {
-        dataColumnsLen = custodyColumns.length;
-      }
+  const blockBytes = this.serializedCache.get(block);
+  if (blockBytes) {
+    // skip serializing data if we already have it
+    this.metrics?.importBlock.persistBlockWithSerializedDataCount.inc();
+    fnPromises.push(this.db.block.putBinary(this.db.block.getId(block), blockBytes));
+  } else {
+    this.metrics?.importBlock.persistBlockNoSerializedDataCount.inc();
+    fnPromises.push(this.db.block.add(block));
+  }
 
-      const dataColumnSidecars = blockInput.getCustodyColumns();
-      if (dataColumnSidecars.length !== dataColumnsLen) {
-        this.logger.debug(
-          `Invalid dataColumnSidecars=${dataColumnSidecars.length} for custody expected custodyColumnsLen=${dataColumnsLen}`
-        );
-      }
+  this.logger.debug("Persist block to hot DB", {slot, root: blockRootHex, inputType: blockInput.type, numBlobs});
 
-      const binaryPuts = [];
-      const nonbinaryPuts = [];
-      for (const dataColumnSidecar of dataColumnSidecars) {
-        // skip reserializing column if we already have it
-        const serialized = this.serializedCache.get(dataColumnSidecar);
-        if (serialized) {
-          binaryPuts.push({key: dataColumnSidecar.index, value: serialized});
-        } else {
-          nonbinaryPuts.push(dataColumnSidecar);
+  if (isBlockInputBlobs(blockInput)) {
+    fnPromises.push(
+      (async () => {
+        if (!blockInput.hasAllData()) {
+          await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT);
         }
-      }
-      fnPromises.push(this.db.dataColumnSidecar.putManyBinary(blockRoot, binaryPuts));
-      fnPromises.push(this.db.dataColumnSidecar.putMany(blockRoot, nonbinaryPuts));
-      this.logger.debug("Persisted dataColumnSidecars to hot DB", {
-        slot: block.message.slot,
-        root: blockRootHex,
-        dataColumnSidecars: dataColumnSidecars.length,
-        numBlobs: blobsLen,
-        custodyColumns: custodyColumns.length,
-      });
-    } else if (isBlockInputBlobs(blockInput)) {
-      const blobSidecars = blockInput.getBlobs();
-      fnPromises.push(this.db.blobSidecars.add({blockRoot, slot: block.message.slot, blobSidecars}));
-      this.logger.debug("Persisted blobSidecars to hot DB", {
-        blobsLen: blobSidecars.length,
-        slot: block.message.slot,
-        root: blockRootHex,
-      });
-    }
+        const blobSidecars = blockInput.getBlobs();
+        await this.db.blobSidecars.add({blockRoot, slot, blobSidecars});
+        this.logger.debug("Persisted blobSidecars to hot DB", {
+          slot,
+          root: blockRootHex,
+          numBlobs: blobSidecars.length,
+        });
+      })()
+    );
+  }
 
-    await Promise.all(fnPromises);
-    this.logger.debug("Persisted blocksInput to db", {
-      blocksInput: blocksInputs.length,
-      slots: prettyPrintIndices(slots),
+  await Promise.all(fnPromises);
+}
+
+/**
+ * Persists data columns to DB for a given block. Accepts a narrow sub-interface of IBlockInput
+ * so it can be reused across forks (e.g. Fulu, Gloas).
+ *
+ * NOTE: Old data is pruned on archive.
+ */
+export async function writeDataColumnsToDb(this: BeaconChain, blockInput: IDataColumnsInput): Promise<void> {
+  const {slot, blockRootHex} = blockInput;
+  const blockRoot = fromHex(blockRootHex);
+
+  if (!blockInput.hasComputedAllData()) {
+    // Supernodes may only have a subset of the data columns by the time the block begins to be imported
+    // because full data availability can be assumed after NUMBER_OF_COLUMNS / 2 columns are available.
+    // Here, however, all data columns must be fully available/reconstructed before persisting to the DB.
+    await blockInput.waitForComputedAllData(BLOB_AVAILABILITY_TIMEOUT).catch(() => {
+      this.logger.debug("Failed to wait for computed all data", {slot, blockRoot: blockRootHex});
     });
   }
+
+  const {custodyColumns} = this.custodyConfig;
+  const dataColumnSidecars = blockInput.getCustodyColumns();
+
+  const binaryPuts: {key: number; value: Uint8Array}[] = [];
+  const nonbinaryPuts = [];
+  for (const dataColumnSidecar of dataColumnSidecars) {
+    // skip reserializing column if we already have it
+    const serialized = this.serializedCache.get(dataColumnSidecar);
+    if (serialized) {
+      binaryPuts.push({key: dataColumnSidecar.index, value: serialized});
+    } else {
+      nonbinaryPuts.push(dataColumnSidecar);
+    }
+  }
+
+  await Promise.all([
+    this.db.dataColumnSidecar.putManyBinary(blockRoot, binaryPuts),
+    this.db.dataColumnSidecar.putMany(blockRoot, nonbinaryPuts),
+  ]);
+
+  this.logger.debug("Persisted dataColumnSidecars to hot DB", {
+    slot,
+    root: blockRootHex,
+    dataColumnSidecars: dataColumnSidecars.length,
+    custodyColumns: custodyColumns.length,
+    numBlobs: dataColumnSidecars[0]?.column.length,
+  });
 }
 
-export async function persistBlockInputs(this: BeaconChain, blockInputs: IBlockInput[]): Promise<void> {
+export async function persistBlockInput(this: BeaconChain, blockInput: IBlockInput): Promise<void> {
   await writeBlockInputToDb
-    .call(this, blockInputs)
+    .call(this, blockInput)
     .catch((e) => {
       this.logger.debug(
         "Error persisting block input in hot db",
         {
-          count: blockInputs.length,
-          slot: blockInputs[0].slot,
-          root: blockInputs[0].blockRootHex,
+          slot: blockInput.slot,
+          root: blockInput.blockRootHex,
         },
         e
       );
     })
     .finally(() => {
-      for (const blockInput of blockInputs) {
-        this.seenBlockInputCache.prune(blockInput.blockRootHex);
-      }
+      this.seenBlockInputCache.prune(blockInput.blockRootHex);
       // Without forcefully clearing this cache, we would rely on WeakMap to evict memory which is not reliable.
       // Clear here (after the DB write) so that writeBlockInputToDb can still use the cached serialized bytes.
-      this.serializedCache.clear();
-      if (blockInputs.length === 1) {
-        this.logger.debug("Pruned block input", {
-          slot: blockInputs[0].slot,
-          root: blockInputs[0].blockRootHex,
-        });
+      //
+      // For Gloas (BlockInputNoData), the execution payload and columns arrive separately after the beacon block.
+      // Do NOT clear the cache here — it must remain available for writeDataColumnsToDb when the payload arrives.
+      // The cache is cleared in the Gloas payload persistence path instead.
+      if (!isBlockInputNoData(blockInput)) {
+        // TODO: enhance this SerializedCache for Gloas because payload may not come
+        // see https://github.com/ChainSafe/lodestar/pull/8974#discussion_r2885598229
+        this.serializedCache.clear();
       }
+      this.logger.debug("Pruned block input", {
+        slot: blockInput.slot,
+        root: blockInput.blockRootHex,
+      });
     });
 }

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

@@ -77,7 +77,7 @@ import {CheckpointBalancesCache} from "./balancesCache.js";
 import {BeaconProposerCache} from "./beaconProposerCache.js";
 import {IBlockInput, isBlockInputBlobs, isBlockInputColumns} from "./blocks/blockInput/index.js";
 import {BlockProcessor, ImportBlockOpts} from "./blocks/index.js";
-import {persistBlockInputs} from "./blocks/writeBlockInputToDb.ts";
+import {persistBlockInput} from "./blocks/writeBlockInputToDb.ts";
 import {BlsMultiThreadWorkerPool, BlsSingleThreadVerifier, IBlsVerifier} from "./bls/index.js";
 import {ColumnReconstructionTracker} from "./ColumnReconstructionTracker.js";
 import {ChainEvent, ChainEventEmitter} from "./emitter.js";
@@ -164,7 +164,7 @@ export class BeaconChain implements IBeaconChain {
   readonly lightClientServer?: LightClientServer;
   readonly reprocessController: ReprocessController;
   readonly archiveStore: ArchiveStore;
-  readonly unfinalizedBlockWrites: JobItemQueue<[IBlockInput[]], void>;
+  readonly unfinalizedBlockWrites: JobItemQueue<[IBlockInput], void>;
 
   // Ops pool
   readonly attestationPool: AttestationPool;
@@ -429,7 +429,7 @@ export class BeaconChain implements IBeaconChain {
     );
 
     this.unfinalizedBlockWrites = new JobItemQueue(
-      persistBlockInputs.bind(this),
+      persistBlockInput.bind(this),
       {
         maxLength: DEFAULT_MAX_PENDING_UNFINALIZED_BLOCK_WRITES,
         signal,