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/importBlock.ts b/packages/beacon-node/src/chain/blocks/importBlock.ts index aaef371a480c..7e61ce6671d9 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..4a8ec0f7dbe8 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/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 { - 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)); - } +export async function writeBlockInputToDb(this: BeaconChain, blockInput: IBlockInput): Promise { + const promises: Promise[] = [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 { + 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).length + : undefined; + const fnPromises: Promise[] = []; - 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 { + 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 { +export async function persistBlockInput(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, - }); + // + // 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, + }); }); } diff --git a/packages/beacon-node/src/chain/chain.ts b/packages/beacon-node/src/chain/chain.ts index 5a063713008f..e485e5cd7741 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"; @@ -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,