From 27aedf328b0833127a0085340bcc5d53b0c83b2a Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Tue, 3 Mar 2026 11:03:49 +0700 Subject: [PATCH 1/8] fix: write single IBlockInput to db --- .../src/chain/blocks/importBlock.ts | 2 +- .../src/chain/blocks/writeBlockInputToDb.ts | 181 ++++++++---------- packages/beacon-node/src/chain/chain.ts | 2 +- 3 files changed, 86 insertions(+), 99 deletions(-) diff --git a/packages/beacon-node/src/chain/blocks/importBlock.ts b/packages/beacon-node/src/chain/blocks/importBlock.ts index 7225441df389..1262721d40ac 100644 --- a/packages/beacon-node/src/chain/blocks/importBlock.ts +++ b/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); } diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index bcc30c1e40c5..31aa16f9e39a 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -1,5 +1,5 @@ import {fulu} from "@lodestar/types"; -import {prettyPrintIndices, toRootHex} from "@lodestar/utils"; +import {toRootHex} from "@lodestar/utils"; import {BeaconChain} from "../chain.js"; import {IBlockInput, isBlockInputBlobs, isBlockInputColumns} from "./blockInput/index.js"; import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js"; @@ -11,128 +11,115 @@ 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. */ -export async function writeBlockInputToDb(this: BeaconChain, blocksInputs: IBlockInput[]): Promise { +export async function writeBlockInputToDb(this: BeaconChain, blockInput: IBlockInput): Promise { const fnPromises: Promise[] = []; - // 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)); - } + const block = blockInput.getBlock(); + const slot = block.message.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)); + } - this.logger.debug("Persist block to hot DB", { - slot: block.message.slot, - root: blockRootHex, - inputType: blockInput.type, - }); + this.logger.debug("Persist block to hot DB", { + slot: block.message.slot, + root: blockRootHex, + inputType: blockInput.type, + }); - if (!blockInput.hasAllData()) { - await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT); - } + if (!blockInput.hasAllData()) { + await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT); + } - // 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}); - }); - } + // 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}); + }); + } - 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 {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 dataColumnSidecars = blockInput.getCustodyColumns(); - if (dataColumnSidecars.length !== dataColumnsLen) { - this.logger.debug( - `Invalid dataColumnSidecars=${dataColumnSidecars.length} for custody expected custodyColumnsLen=${dataColumnsLen}` - ); - } + const dataColumnSidecars = blockInput.getCustodyColumns(); + if (dataColumnSidecars.length !== dataColumnsLen) { + this.logger.debug( + `Invalid dataColumnSidecars=${dataColumnSidecars.length} for custody expected custodyColumnsLen=${dataColumnsLen}` + ); + } - 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); - } + 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); } - 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, - }); } - - await Promise.all(fnPromises); - this.logger.debug("Persisted blocksInput to db", { - blocksInput: blocksInputs.length, - slots: prettyPrintIndices(slots), + 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, }); } + + await Promise.all(fnPromises); + this.logger.debug("Persisted blockInput to db", {slot}); } -export async function persistBlockInputs(this: BeaconChain, blockInputs: IBlockInput[]): Promise { +export async function persistBlockInputs(this: BeaconChain, blockInput: IBlockInput): Promise { 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, - }); - } + this.logger.debug("Pruned block input", { + slot: blockInput.slot, + root: blockInput.blockRootHex, + }); }); } diff --git a/packages/beacon-node/src/chain/chain.ts b/packages/beacon-node/src/chain/chain.ts index a0a29bcc0c0e..6befe7ec3c2f 100644 --- a/packages/beacon-node/src/chain/chain.ts +++ b/packages/beacon-node/src/chain/chain.ts @@ -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; From abb9a4ab5da4ab5811398b102279a3d890481bf9 Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Tue, 3 Mar 2026 11:29:18 +0700 Subject: [PATCH 2/8] refactor: separate writeBlockInputToDb into parallel block and column writes Add IDataColumnsInput interface and extract writeDataColumnsToDb() as a standalone exported function accepting a narrow sub-interface of IBlockInput. Block+blobs (pre-fulu) and data columns (fulu+) are now written in parallel, enabling reuse of writeDataColumnsToDb() for the Gloas fork later. Co-Authored-By: Claude Sonnet 4.6 --- .../src/chain/blocks/blockInput/types.ts | 12 ++ .../src/chain/blocks/writeBlockInputToDb.ts | 133 +++++++++--------- 2 files changed, 78 insertions(+), 67 deletions(-) diff --git a/packages/beacon-node/src/chain/blocks/blockInput/types.ts b/packages/beacon-node/src/chain/blocks/blockInput/types.ts index 587ad83c70e8..d6acd478fc1a 100644 --- a/packages/beacon-node/src/chain/blocks/blockInput/types.ts +++ b/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; +} + /** * 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. diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 31aa16f9e39a..358ddca745d5 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -1,7 +1,6 @@ -import {fulu} from "@lodestar/types"; -import {toRootHex} from "@lodestar/utils"; +import {fromHex, toRootHex} from "@lodestar/utils"; import {BeaconChain} from "../chain.js"; -import {IBlockInput, isBlockInputBlobs, isBlockInputColumns} from "./blockInput/index.js"; +import {IBlockInput, IDataColumnsInput, isBlockInputBlobs, isBlockInputColumns} from "./blockInput/index.js"; import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js"; /** @@ -10,93 +9,93 @@ 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, blockInput: IBlockInput): Promise { - const fnPromises: Promise[] = []; + const promises: Promise[] = [writeBlockAndBlobsToDb.call(this, blockInput)]; + + if (isBlockInputColumns(blockInput)) { + promises.push(writeDataColumnsToDb.call(this, blockInput)); + } + await Promise.all(promises); + this.logger.debug("Persisted blockInput to db", {slot: blockInput.slot}); +} + +async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput): Promise { const block = blockInput.getBlock(); const slot = block.message.slot; - const blockRoot = this.config.getForkTypes(block.message.slot).BeaconBlock.hashTreeRoot(block.message); + const blockRoot = this.config.getForkTypes(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)); + await this.db.block.putBinary(this.db.block.getId(block), blockBytes); } else { this.metrics?.importBlock.persistBlockNoSerializedDataCount.inc(); - fnPromises.push(this.db.block.add(block)); + await this.db.block.add(block); } - this.logger.debug("Persist block to hot DB", { - slot: block.message.slot, - root: blockRootHex, - inputType: blockInput.type, - }); + this.logger.debug("Persist block to hot DB", {slot, root: blockRootHex, inputType: blockInput.type}); - if (!blockInput.hasAllData()) { - await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT); + if (isBlockInputBlobs(blockInput)) { + if (!blockInput.hasAllData()) { + await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT); + } + const blobSidecars = blockInput.getBlobs(); + await this.db.blobSidecars.add({blockRoot, slot, blobSidecars}); + this.logger.debug("Persisted blobSidecars to hot DB", {blobsLen: blobSidecars.length, slot, root: 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}); - }); - } +/** + * 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 { + const {slot, blockRootHex} = blockInput; + const blockRoot = fromHex(blockRootHex); - 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; - } + 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 dataColumnSidecars = blockInput.getCustodyColumns(); - if (dataColumnSidecars.length !== dataColumnsLen) { - this.logger.debug( - `Invalid dataColumnSidecars=${dataColumnSidecars.length} for custody expected custodyColumnsLen=${dataColumnsLen}` - ); - } + const {custodyColumns} = this.custodyConfig; + const dataColumnSidecars = blockInput.getCustodyColumns(); - 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); - } + 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); } - 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, - }); } - await Promise.all(fnPromises); - this.logger.debug("Persisted blockInput to db", {slot}); + 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, + }); } export async function persistBlockInputs(this: BeaconChain, blockInput: IBlockInput): Promise { From b1722b2cadf2fbea62f0ac23e84e361871772e6c Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Tue, 3 Mar 2026 13:12:30 +0700 Subject: [PATCH 3/8] refactor: rename persistBlockInputs to persistBlockInput Co-Authored-By: Claude Sonnet 4.6 --- packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts | 2 +- packages/beacon-node/src/chain/chain.ts | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 358ddca745d5..5e92ab3e088a 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -98,7 +98,7 @@ export async function writeDataColumnsToDb(this: BeaconChain, blockInput: IDataC }); } -export async function persistBlockInputs(this: BeaconChain, blockInput: IBlockInput): Promise { +export async function persistBlockInput(this: BeaconChain, blockInput: IBlockInput): Promise { await writeBlockInputToDb .call(this, blockInput) .catch((e) => { diff --git a/packages/beacon-node/src/chain/chain.ts b/packages/beacon-node/src/chain/chain.ts index 6befe7ec3c2f..6bb44fcd3585 100644 --- a/packages/beacon-node/src/chain/chain.ts +++ b/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"; @@ -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, From 9cde8a5e0fc160f1489d618f5adde44f89c3daca Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Tue, 3 Mar 2026 13:23:12 +0700 Subject: [PATCH 4/8] fix: only clear serializedCache for pre-gloas --- .../src/chain/blocks/writeBlockInputToDb.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 5e92ab3e088a..48b307f8fe74 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -1,6 +1,6 @@ import {fromHex, toRootHex} from "@lodestar/utils"; import {BeaconChain} from "../chain.js"; -import {IBlockInput, IDataColumnsInput, isBlockInputBlobs, isBlockInputColumns} from "./blockInput/index.js"; +import {IBlockInput, IDataColumnsInput, isBlockInputBlobs, isBlockInputColumns, isBlockInputNoData} from "./blockInput/index.js"; import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js"; /** @@ -115,7 +115,13 @@ export async function persistBlockInput(this: BeaconChain, blockInput: IBlockInp 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(); + // + // 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)) { + this.serializedCache.clear(); + } this.logger.debug("Pruned block input", { slot: blockInput.slot, root: blockInput.blockRootHex, From 8cef8fd2c9f0ebc58bb6db52c553c4070e5d51dc Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Tue, 3 Mar 2026 13:40:53 +0700 Subject: [PATCH 5/8] fix: guard serializedCache.clear() for pre-gloas only in persistBlockInput MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit For Gloas (BlockInputNoData), the execution payload and columns arrive separately after the beacon block. The serialized cache must not be cleared when the block is persisted — it remains available for writeDataColumnsToDb when the payload arrives later. Co-Authored-By: Claude Sonnet 4.6 --- .../beacon-node/src/chain/blocks/writeBlockInputToDb.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 48b307f8fe74..0173b71b74f9 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -1,6 +1,12 @@ import {fromHex, toRootHex} from "@lodestar/utils"; import {BeaconChain} from "../chain.js"; -import {IBlockInput, IDataColumnsInput, isBlockInputBlobs, isBlockInputColumns, isBlockInputNoData} from "./blockInput/index.js"; +import { + IBlockInput, + IDataColumnsInput, + isBlockInputBlobs, + isBlockInputColumns, + isBlockInputNoData, +} from "./blockInput/index.js"; import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js"; /** From 5dc06f8642189f70ec5723e70cd8d4b45b990d05 Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Mon, 9 Mar 2026 10:17:37 +0700 Subject: [PATCH 6/8] fix: add numBlobs to log --- packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 0173b71b74f9..430716478ee7 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -101,6 +101,7 @@ export async function writeDataColumnsToDb(this: BeaconChain, blockInput: IDataC root: blockRootHex, dataColumnSidecars: dataColumnSidecars.length, custodyColumns: custodyColumns.length, + numBlobs: dataColumnSidecars[0]?.column.length, }); } @@ -126,6 +127,8 @@ export async function persistBlockInput(this: BeaconChain, blockInput: IBlockInp // 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", { From 81ba7dd633674a816676e09979a5525135b6ae61 Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Mon, 9 Mar 2026 10:27:36 +0700 Subject: [PATCH 7/8] fix: write block + blobs in parallel --- .../src/chain/blocks/writeBlockInputToDb.ts | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 430716478ee7..67fe8d472191 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -34,27 +34,34 @@ async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput const slot = block.message.slot; const blockRoot = this.config.getForkTypes(slot).BeaconBlock.hashTreeRoot(block.message); const blockRootHex = toRootHex(blockRoot); + const fnPromises: Promise[] = []; const blockBytes = this.serializedCache.get(block); if (blockBytes) { // skip serializing data if we already have it this.metrics?.importBlock.persistBlockWithSerializedDataCount.inc(); - await this.db.block.putBinary(this.db.block.getId(block), blockBytes); + fnPromises.push(this.db.block.putBinary(this.db.block.getId(block), blockBytes)); } else { this.metrics?.importBlock.persistBlockNoSerializedDataCount.inc(); - await this.db.block.add(block); + fnPromises.push(this.db.block.add(block)); } this.logger.debug("Persist block to hot DB", {slot, root: blockRootHex, inputType: blockInput.type}); if (isBlockInputBlobs(blockInput)) { - if (!blockInput.hasAllData()) { - await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT); - } - const blobSidecars = blockInput.getBlobs(); - await this.db.blobSidecars.add({blockRoot, slot, blobSidecars}); - this.logger.debug("Persisted blobSidecars to hot DB", {blobsLen: blobSidecars.length, slot, root: blockRootHex}); + fnPromises.push( + (async () => { + if (!blockInput.hasAllData()) { + await blockInput.waitForAllData(BLOB_AVAILABILITY_TIMEOUT); + } + const blobSidecars = blockInput.getBlobs(); + await this.db.blobSidecars.add({blockRoot, slot, blobSidecars}); + this.logger.debug("Persisted blobSidecars to hot DB", {blobsLen: blobSidecars.length, slot, root: blockRootHex}); + })() + ); } + + await Promise.all(fnPromises); } /** From 40b89d14d26b16b8e8a92d2a6a654f8b8eef0f22 Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Mon, 9 Mar 2026 10:35:13 +0700 Subject: [PATCH 8/8] chore: log numBlobs for all functions --- .../src/chain/blocks/writeBlockInputToDb.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index 67fe8d472191..4a8ec0f7dbe8 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -1,4 +1,7 @@ +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, @@ -26,7 +29,7 @@ export async function writeBlockInputToDb(this: BeaconChain, blockInput: IBlockI } await Promise.all(promises); - this.logger.debug("Persisted blockInput to db", {slot: blockInput.slot}); + this.logger.debug("Persisted blockInput to db", {slot: blockInput.slot, root: blockInput.blockRootHex}); } async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput): Promise { @@ -34,6 +37,9 @@ async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput 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).length + : undefined; const fnPromises: Promise[] = []; const blockBytes = this.serializedCache.get(block); @@ -46,7 +52,7 @@ async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput fnPromises.push(this.db.block.add(block)); } - this.logger.debug("Persist block to hot DB", {slot, root: blockRootHex, inputType: blockInput.type}); + this.logger.debug("Persist block to hot DB", {slot, root: blockRootHex, inputType: blockInput.type, numBlobs}); if (isBlockInputBlobs(blockInput)) { fnPromises.push( @@ -56,7 +62,11 @@ async function writeBlockAndBlobsToDb(this: BeaconChain, blockInput: IBlockInput } const blobSidecars = blockInput.getBlobs(); await this.db.blobSidecars.add({blockRoot, slot, blobSidecars}); - this.logger.debug("Persisted blobSidecars to hot DB", {blobsLen: blobSidecars.length, slot, root: blockRootHex}); + this.logger.debug("Persisted blobSidecars to hot DB", { + slot, + root: blockRootHex, + numBlobs: blobSidecars.length, + }); })() ); }