diff --git a/dashboards/lodestar_beacon_chain.json b/dashboards/lodestar_beacon_chain.json index 15da60130f9d..affcaf153d05 100644 --- a/dashboards/lodestar_beacon_chain.json +++ b/dashboards/lodestar_beacon_chain.json @@ -1326,6 +1326,343 @@ ], "title": "Grow", "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 60 + }, + "id": 39, + "panels": [], + "title": "Unfinalized Block Writes Queue", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [] + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 61 + }, + "id": 40, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": false + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "lodestar_unfinalized_block_writes_queue_length", + "instant": false, + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Queue Length", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "fieldConfig": { + "defaults": { + "custom": { + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "scaleDistribution": { + "type": "linear" + } + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 61 + }, + "id": 41, + "options": { + "calculate": false, + "cellGap": 1, + "color": { + "exponent": 0.5, + "fill": "dark-orange", + "mode": "scheme", + "reverse": false, + "scale": "exponential", + "scheme": "Magma", + "steps": 64 + }, + "exemplars": { + "color": "rgba(255,0,255,0.7)" + }, + "filterValues": { + "le": 1e-9 + }, + "legend": { + "show": true + }, + "rowsFrame": { + "layout": "auto" + }, + "tooltip": { + "mode": "single", + "showColorScale": false, + "yHistogram": false + }, + "yAxis": { + "axisPlacement": "left", + "reverse": false + } + }, + "pluginVersion": "10.4.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(lodestar_unfinalized_block_writes_queue_job_time_seconds_bucket[$rate_interval])", + "format": "heatmap", + "instant": false, + "legendFormat": "{{le}}", + "range": true, + "refId": "A" + } + ], + "title": "Job Time", + "type": "heatmap" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [] + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 69 + }, + "id": 43, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": false + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "lodestar_unfinalized_block_writes_queue_concurrency", + "instant": false, + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Queue Concurrency", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "fieldConfig": { + "defaults": { + "custom": { + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "scaleDistribution": { + "type": "linear" + } + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 69 + }, + "id": 42, + "options": { + "calculate": false, + "cellGap": 1, + "color": { + "exponent": 0.5, + "fill": "dark-orange", + "mode": "scheme", + "reverse": false, + "scale": "exponential", + "scheme": "Magma", + "steps": 64 + }, + "exemplars": { + "color": "rgba(255,0,255,0.7)" + }, + "filterValues": { + "le": 1e-9 + }, + "legend": { + "show": true + }, + "rowsFrame": { + "layout": "auto" + }, + "tooltip": { + "mode": "single", + "showColorScale": false, + "yHistogram": false + }, + "yAxis": { + "axisPlacement": "left", + "reverse": false + } + }, + "pluginVersion": "10.4.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(lodestar_unfinalized_block_writes_queue_job_wait_time_seconds_bucket[$rate_interval])", + "format": "heatmap", + "instant": false, + "legendFormat": "{{le}}", + "range": true, + "refId": "A" + } + ], + "title": "Job Wait Time", + "type": "heatmap" } ], "refresh": "10s", diff --git a/packages/beacon-node/src/api/impl/beacon/blocks/index.ts b/packages/beacon-node/src/api/impl/beacon/blocks/index.ts index 9748c88c50c7..a4c3bf89cbf1 100644 --- a/packages/beacon-node/src/api/impl/beacon/blocks/index.ts +++ b/packages/beacon-node/src/api/impl/beacon/blocks/index.ts @@ -30,7 +30,7 @@ import { isDenebBlockContents, sszTypesFor, } from "@lodestar/types"; -import {fromAsync, fromHex, sleep, toHex, toRootHex} from "@lodestar/utils"; +import {fromHex, sleep, toHex, toRootHex} from "@lodestar/utils"; import {BlockInputSource, isBlockInputBlobs, isBlockInputColumns} from "../../../../chain/blocks/blockInput/index.js"; import {ImportBlockOpts} from "../../../../chain/blocks/types.js"; import {verifyBlocksInEpoch} from "../../../../chain/blocks/verifyBlock.js"; @@ -285,7 +285,7 @@ export function getBeaconBlockApi({ () => // there is no rush to persist block since we published it to gossip anyway chain - .processBlock(blockForImport, {...opts, eagerPersistBlock: false}) + .processBlock(blockForImport, opts) .catch((e) => { if (e instanceof BlockError && e.type.code === BlockErrorCode.PARENT_UNKNOWN) { chain.emitter.emit(ChainEvent.unknownParent, { @@ -626,6 +626,7 @@ export function getBeaconBlockApi({ const {block, executionOptimistic, finalized} = await getBlockResponse(chain, blockId); const fork = config.getForkName(block.message.slot); const blockRoot = sszTypesFor(fork).BeaconBlock.hashTreeRoot(block.message); + const blockRootHex = toRootHex(blockRoot); let data: deneb.BlobSidecars; @@ -642,10 +643,7 @@ export function getBeaconBlockApi({ const blobCount = blobKzgCommitments.length; if (blobCount > 0) { - let dataColumnSidecars = await fromAsync(db.dataColumnSidecar.valuesStream(blockRoot)); - if (dataColumnSidecars.length === 0) { - dataColumnSidecars = await fromAsync(db.dataColumnSidecarArchive.valuesStream(block.message.slot)); - } + const dataColumnSidecars = await chain.getDataColumnSidecars(block.message.slot, blockRootHex); if (dataColumnSidecars.length === 0) { throw new ApiError( @@ -682,10 +680,7 @@ export function getBeaconBlockApi({ data = []; } } else if (isForkPostDeneb(fork)) { - let {blobSidecars} = (await db.blobSidecars.get(blockRoot)) ?? {}; - if (!blobSidecars) { - ({blobSidecars} = (await db.blobSidecarsArchive.get(block.message.slot)) ?? {}); - } + const blobSidecars = await chain.getBlobSidecars(block.message.slot, blockRootHex); if (!blobSidecars) { throw new ApiError( @@ -715,6 +710,7 @@ export function getBeaconBlockApi({ const {block, executionOptimistic, finalized} = await getBlockResponse(chain, blockId); const fork = config.getForkName(block.message.slot); const blockRoot = sszTypesFor(fork).BeaconBlock.hashTreeRoot(block.message); + const blockRootHex = toRootHex(blockRoot); let blobs: deneb.Blobs; @@ -731,10 +727,7 @@ export function getBeaconBlockApi({ const blobCount = blobKzgCommitments.length; if (blobCount > 0) { - let dataColumnSidecars = await fromAsync(db.dataColumnSidecar.valuesStream(blockRoot)); - if (dataColumnSidecars.length === 0) { - dataColumnSidecars = await fromAsync(db.dataColumnSidecarArchive.valuesStream(block.message.slot)); - } + const dataColumnSidecars = await chain.getDataColumnSidecars(block.message.slot, blockRootHex); if (dataColumnSidecars.length === 0) { throw new ApiError( @@ -766,10 +759,7 @@ export function getBeaconBlockApi({ blobs = []; } } else if (isForkPostDeneb(fork)) { - let {blobSidecars} = (await db.blobSidecars.get(blockRoot)) ?? {}; - if (!blobSidecars) { - ({blobSidecars} = (await db.blobSidecarsArchive.get(block.message.slot)) ?? {}); - } + const blobSidecars = await chain.getBlobSidecars(block.message.slot, blockRootHex); if (!blobSidecars) { throw new ApiError( diff --git a/packages/beacon-node/src/api/impl/debug/index.ts b/packages/beacon-node/src/api/impl/debug/index.ts index f1f0562fd067..ae2b21611221 100644 --- a/packages/beacon-node/src/api/impl/debug/index.ts +++ b/packages/beacon-node/src/api/impl/debug/index.ts @@ -3,7 +3,7 @@ import {ApplicationMethods} from "@lodestar/api/server"; import {ExecutionStatus} from "@lodestar/fork-choice"; import {ZERO_HASH_HEX, isForkPostDeneb, isForkPostFulu} from "@lodestar/params"; import {BeaconState, deneb, fulu, sszTypesFor} from "@lodestar/types"; -import {fromAsync, toRootHex} from "@lodestar/utils"; +import {toRootHex} from "@lodestar/utils"; import {isOptimisticBlock} from "../../../util/forkChoice.js"; import {getStateSlotFromBytes} from "../../../util/multifork.js"; import {getBlockResponse} from "../beacon/blocks/utils.js"; @@ -14,7 +14,6 @@ import {assertUniqueItems} from "../utils.js"; export function getDebugApi({ chain, config, - db, }: Pick): ApplicationMethods { return { async getDebugChainHeadsV2() { @@ -104,10 +103,7 @@ export function getDebugApi({ : 0; if (isForkPostFulu(fork) && blobCount > 0) { - dataColumnSidecars = await fromAsync(db.dataColumnSidecar.valuesStream(blockRoot)); - if (dataColumnSidecars.length === 0) { - dataColumnSidecars = await fromAsync(db.dataColumnSidecarArchive.valuesStream(block.message.slot)); - } + dataColumnSidecars = await chain.getDataColumnSidecars(block.message.slot, toRootHex(blockRoot)); if (dataColumnSidecars.length === 0) { throw Error( diff --git a/packages/beacon-node/src/chain/archiveStore/utils/archiveBlocks.ts b/packages/beacon-node/src/chain/archiveStore/utils/archiveBlocks.ts index 893327e9b376..75cd1df77d1a 100644 --- a/packages/beacon-node/src/chain/archiveStore/utils/archiveBlocks.ts +++ b/packages/beacon-node/src/chain/archiveStore/utils/archiveBlocks.ts @@ -238,6 +238,7 @@ async function migrateBlocksFromHotToColdDb(db: IBeaconDb, blocks: BlockRootSlot // load Buffer instead of SignedBeaconBlock to improve performance const canonicalBlockEntries: BlockArchiveBatchPutBinaryItem[] = await Promise.all( canonicalBlocks.map(async (block) => { + // Here we assume the blocks are already in the hot db const blockBuffer = await db.block.getBinary(block.root); if (!blockBuffer) { throw Error(`Block not found for slot ${block.slot} root ${toRootHex(block.root)}`); @@ -294,6 +295,8 @@ async function migrateBlobSidecarsFromHotToColdDb( ); }) .map(async (block) => { + // Here we assume the blob sidecars are already in the hot db + // instead of checking first the block input cache const bytes = await db.blobSidecars.getBinary(block.root); if (!bytes) { throw Error(`No blobSidecars found for slot ${block.slot} root ${toRootHex(block.root)}`); @@ -343,6 +346,7 @@ async function migrateDataColumnSidecarsFromHotToColdDb( continue; } + // Here we assume the data column sidecars are already in the hot db const dataColumnSidecarBytes = await fromAsync(db.dataColumnSidecar.valuesStreamBinary(block.root)); // there could be 0 dataColumnSidecarBytes if block has no blob logger.verbose("migrateDataColumnSidecarsFromHotToColdDb", { diff --git a/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts b/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts index d464b1362ec5..a35c155663e3 100644 --- a/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts +++ b/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts @@ -412,6 +412,10 @@ export class BlockInputBlobs extends AbstractBlockInput { - this.logger.warn( - "Error pruning eagerly imported block inputs, DB may grow in size if this error happens frequently", - {slot: blocks.map((block) => block.getBlock().message.slot).join(",")}, - e - ); - }); - } - throw err; } } diff --git a/packages/beacon-node/src/chain/blocks/types.ts b/packages/beacon-node/src/chain/blocks/types.ts index f8f091a5b07f..50ed0076515c 100644 --- a/packages/beacon-node/src/chain/blocks/types.ts +++ b/packages/beacon-node/src/chain/blocks/types.ts @@ -78,8 +78,6 @@ export type ImportBlockOpts = { validBlobSidecars?: BlobSidecarValidation; /** Seen timestamp seconds */ seenTimestampSec?: number; - /** Set to true if persist block right at verification time */ - eagerPersistBlock?: boolean; }; /** diff --git a/packages/beacon-node/src/chain/blocks/verifyBlock.ts b/packages/beacon-node/src/chain/blocks/verifyBlock.ts index 961867260931..0f0169a7ab84 100644 --- a/packages/beacon-node/src/chain/blocks/verifyBlock.ts +++ b/packages/beacon-node/src/chain/blocks/verifyBlock.ts @@ -21,7 +21,6 @@ import {verifyBlocksDataAvailability} from "./verifyBlocksDataAvailability.js"; import {SegmentExecStatus, verifyBlocksExecutionPayload} from "./verifyBlocksExecutionPayloads.js"; import {verifyBlocksSignatures} from "./verifyBlocksSignatures.js"; import {verifyBlocksStateTransitionOnly} from "./verifyBlocksStateTransitionOnly.js"; -import {writeBlockInputToDb} from "./writeBlockInputToDb.js"; /** * Verifies 1 or more blocks are fully valid; from a linear sequence of blocks. @@ -156,13 +155,6 @@ export async function verifyBlocksInEpoch( opts ) : Promise.resolve({verifySignaturesTime: Date.now()}), - - // ideally we want to only persist blocks after verifying them however the reality is there are - // rarely invalid blocks we'll batch all I/O operation here to reduce the overhead if there's - // an error, we'll remove blocks not in forkchoice - opts.verifyOnly !== true && opts.eagerPersistBlock - ? writeBlockInputToDb.call(this, blockInputs) - : Promise.resolve(), ]); if (opts.verifyOnly !== true) { diff --git a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts index c2d96afe41f8..b008bcd94eff 100644 --- a/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts +++ b/packages/beacon-node/src/chain/blocks/writeBlockInputToDb.ts @@ -98,35 +98,29 @@ export async function writeBlockInputToDb(this: BeaconChain, blocksInputs: IBloc } } -/** - * Prunes eagerly persisted block inputs only if not known to the fork-choice - */ -export async function removeEagerlyPersistedBlockInputs(this: BeaconChain, blockInputs: IBlockInput[]): Promise { - const blockToRemove = []; - const blobsToRemove = []; - const dataColumnsToRemove = []; - - for (const blockInput of blockInputs) { - const block = blockInput.getBlock(); - const slot = block.message.slot; - const blockRoot = this.config.getForkTypes(slot).BeaconBlock.hashTreeRoot(block.message); - const blockRootHex = toRootHex(blockRoot); - if (!this.forkChoice.hasBlockHex(blockRootHex)) { - blockToRemove.push(block); - - if (isBlockInputColumns(blockInput) && blockInput.getCustodyColumns().length > 0) { - dataColumnsToRemove.push(blockRoot); - } else if (isBlockInputBlobs(blockInput)) { - const blobSidecars = blockInput.getBlobs(); - blobsToRemove.push({blockRoot, slot, blobSidecars}); +export async function persistBlockInputs(this: BeaconChain, blockInputs: IBlockInput[]): Promise { + await writeBlockInputToDb + .call(this, blockInputs) + .catch((e) => { + this.logger.debug( + "Error persisting block input in hot db", + { + count: blockInputs.length, + slot: blockInputs[0].slot, + root: blockInputs[0].blockRootHex, + }, + e + ); + }) + .finally(() => { + for (const blockInput of blockInputs) { + this.seenBlockInputCache.prune(blockInput.blockRootHex); } - } - } - - await Promise.all([ - // TODO: Batch DB operations not with Promise.all but with level db ops - this.db.block.batchRemove(blockToRemove), - this.db.blobSidecars.batchRemove(blobsToRemove), - this.db.dataColumnSidecar.deleteMany(dataColumnsToRemove), - ]); + if (blockInputs.length === 1) { + this.logger.debug("Pruned block input", { + slot: blockInputs[0].slot, + root: blockInputs[0].blockRootHex, + }); + } + }); } diff --git a/packages/beacon-node/src/chain/chain.ts b/packages/beacon-node/src/chain/chain.ts index 8fb52cf57439..d78300be552c 100644 --- a/packages/beacon-node/src/chain/chain.ts +++ b/packages/beacon-node/src/chain/chain.ts @@ -37,14 +37,19 @@ import { UintNum64, ValidatorIndex, Wei, + deneb, + fulu, isBlindedBeaconBlock, phase0, rewards, + ssz, + sszTypesFor, } from "@lodestar/types"; import {Logger, fromHex, gweiToWei, isErrorAborted, pruneSetToMax, sleep, toRootHex} from "@lodestar/utils"; import {ProcessShutdownCallback} from "@lodestar/validator"; import {GENESIS_EPOCH, ZERO_HASH} from "../constants/index.js"; import {IBeaconDb} from "../db/index.js"; +import {BLOB_SIDECARS_IN_WRAPPER_INDEX} from "../db/repositories/blobSidecars.ts"; import {BuilderStatus} from "../execution/builder/http.js"; import {IExecutionBuilder, IExecutionEngine} from "../execution/index.js"; import {Metrics} from "../metrics/index.js"; @@ -55,12 +60,15 @@ import {CustodyConfig, getValidatorsCustodyRequirement} from "../util/dataColumn import {callInNextEventLoop} from "../util/eventLoop.js"; import {ensureDir, writeIfNotExist} from "../util/file.js"; import {isOptimisticBlock} from "../util/forkChoice.js"; +import {JobItemQueue} from "../util/queue/itemQueue.ts"; import {SerializedCache} from "../util/serializedCache.js"; +import {getSlotFromSignedBeaconBlockSerialized} from "../util/sszBytes.ts"; import {ArchiveStore} from "./archiveStore/archiveStore.js"; import {CheckpointBalancesCache} from "./balancesCache.js"; import {BeaconProposerCache} from "./beaconProposerCache.js"; -import {IBlockInput} from "./blocks/blockInput/index.js"; +import {IBlockInput, isBlockInputBlobs, isBlockInputColumns} from "./blocks/blockInput/index.js"; import {BlockProcessor, ImportBlockOpts} from "./blocks/index.js"; +import {persistBlockInputs} from "./blocks/writeBlockInputToDb.ts"; import {BlsMultiThreadWorkerPool, BlsSingleThreadVerifier, IBlsVerifier} from "./bls/index.js"; import {ColumnReconstructionTracker} from "./ColumnReconstructionTracker.js"; import {ChainEvent, ChainEventEmitter} from "./emitter.js"; @@ -113,6 +121,11 @@ import {ValidatorMonitor} from "./validatorMonitor.js"; */ const DEFAULT_MAX_CACHED_PRODUCED_RESULTS = 4; +/** + * The maximum number of pending unfinalized block writes to the database before backpressure is applied. + */ +const DEFAULT_MAX_PENDING_UNFINALIZED_BLOCK_WRITES = 32; + export class BeaconChain implements IBeaconChain { readonly genesisTime: UintNum64; readonly genesisValidatorsRoot: Root; @@ -136,6 +149,7 @@ export class BeaconChain implements IBeaconChain { readonly lightClientServer?: LightClientServer; readonly reprocessController: ReprocessController; readonly archiveStore: ArchiveStore; + readonly unfinalizedBlockWrites: JobItemQueue<[IBlockInput[]], void>; // Ops pool readonly attestationPool: AttestationPool; @@ -405,6 +419,15 @@ export class BeaconChain implements IBeaconChain { signal ); + this.unfinalizedBlockWrites = new JobItemQueue( + persistBlockInputs.bind(this), + { + maxLength: DEFAULT_MAX_PENDING_UNFINALIZED_BLOCK_WRITES, + signal, + }, + metrics?.unfinalizedBlockWritesQueue + ); + // always run PrepareNextSlotScheduler except for fork_choice spec tests if (!opts?.disablePrepareNextSlot) { new PrepareNextSlotScheduler(this, this.config, metrics, this.logger, signal); @@ -430,6 +453,12 @@ export class BeaconChain implements IBeaconChain { async close(): Promise { await this.archiveStore.close(); await this.bls.close(); + + // Since we don't persist unfinalized fork-choice, + // we can abort any ongoing unfinalized block writes. + // TODO: persist fork choice to disk and allow unfinalized block writes to complete. + this.unfinalizedBlockWrites.dropAllJobs(); + this.abortController.abort(); } @@ -652,6 +681,13 @@ export class BeaconChain implements IBeaconChain { // Unfinalized slot, attempt to find in fork-choice const block = this.forkChoice.getCanonicalBlockAtSlot(slot); if (block) { + // Block found in fork-choice. + // It may be in the block input cache, awaiting full DA reconstruction, check there first + // Otherwise (most likely), check the hot db + const blockInput = this.seenBlockInputCache.get(block.blockRoot); + if (blockInput?.hasBlock()) { + return {block: blockInput.getBlock(), executionOptimistic: isOptimisticBlock(block), finalized: false}; + } const data = await this.db.block.get(fromHex(block.blockRoot)); if (data) { return {block: data, executionOptimistic: isOptimisticBlock(block), finalized: false}; @@ -671,6 +707,13 @@ export class BeaconChain implements IBeaconChain { ): Promise<{block: SignedBeaconBlock; executionOptimistic: boolean; finalized: boolean} | null> { const block = this.forkChoice.getBlockHex(root); if (block) { + // Block found in fork-choice. + // It may be in the block input cache, awaiting full DA reconstruction, check there first + // Otherwise (most likely), check the hot db + const blockInput = this.seenBlockInputCache.get(block.blockRoot); + if (blockInput?.hasBlock()) { + return {block: blockInput.getBlock(), executionOptimistic: isOptimisticBlock(block), finalized: false}; + } const data = await this.db.block.get(fromHex(root)); if (data) { return {block: data, executionOptimistic: isOptimisticBlock(block), finalized: false}; @@ -683,6 +726,133 @@ export class BeaconChain implements IBeaconChain { return data && {block: data, executionOptimistic: false, finalized: true}; } + async getSerializedBlockByRoot( + root: string + ): Promise<{block: Uint8Array; executionOptimistic: boolean; finalized: boolean; slot: Slot} | null> { + const block = this.forkChoice.getBlockHex(root); + if (block) { + // Block found in fork-choice. + // It may be in the block input cache, awaiting full DA reconstruction, check there first + // Otherwise (most likely), check the hot db + const blockInput = this.seenBlockInputCache.get(block.blockRoot); + if (blockInput?.hasBlock()) { + const signedBlock = blockInput.getBlock(); + const serialized = this.serializedCache.get(signedBlock); + if (serialized) { + return { + block: serialized, + executionOptimistic: isOptimisticBlock(block), + finalized: false, + slot: blockInput.slot, + }; + } + return { + block: sszTypesFor(blockInput.forkName).SignedBeaconBlock.serialize(signedBlock), + executionOptimistic: isOptimisticBlock(block), + finalized: false, + slot: blockInput.slot, + }; + } + const data = await this.db.block.getBinary(fromHex(root)); + if (data) { + const slot = getSlotFromSignedBeaconBlockSerialized(data); + if (slot === null) throw new Error(`Invalid block data stored in DB for root: ${root}`); + return {block: data, executionOptimistic: isOptimisticBlock(block), finalized: false, slot}; + } + // If block is not found in hot db, try cold db since there could be an archive cycle happening + // TODO: Add a lock to the archiver to have deterministic behavior on where are blocks + } + + const data = await this.db.blockArchive.getBinaryEntryByRoot(fromHex(root)); + return data && {block: data.value, executionOptimistic: false, finalized: true, slot: data.key}; + } + + async getBlobSidecars(blockSlot: Slot, blockRootHex: string): Promise { + const blockInput = this.seenBlockInputCache.get(blockRootHex); + if (blockInput) { + if (!isBlockInputBlobs(blockInput)) { + throw new Error(`Expected block input to have blobs: slot=${blockSlot} root=${blockRootHex}`); + } + if (!blockInput.hasAllData()) { + return null; + } + return blockInput.getBlobs(); + } + const unfinalizedBlobSidecars = (await this.db.blobSidecars.get(fromHex(blockRootHex)))?.blobSidecars ?? null; + if (unfinalizedBlobSidecars) { + return unfinalizedBlobSidecars; + } + return (await this.db.blobSidecarsArchive.get(blockSlot))?.blobSidecars ?? null; + } + + async getSerializedBlobSidecars(blockSlot: Slot, blockRootHex: string): Promise { + const blockInput = this.seenBlockInputCache.get(blockRootHex); + if (blockInput) { + if (!isBlockInputBlobs(blockInput)) { + throw new Error(`Expected block input to have blobs: slot=${blockSlot} root=${blockRootHex}`); + } + if (!blockInput.hasAllData()) { + return null; + } + return ssz.deneb.BlobSidecars.serialize(blockInput.getBlobs()); + } + const unfinalizedBlobSidecarsWrapper = await this.db.blobSidecars.getBinary(fromHex(blockRootHex)); + if (unfinalizedBlobSidecarsWrapper) { + return unfinalizedBlobSidecarsWrapper.slice(BLOB_SIDECARS_IN_WRAPPER_INDEX); + } + const finalizedBlobSidecarsWrapper = await this.db.blobSidecarsArchive.getBinary(blockSlot); + if (finalizedBlobSidecarsWrapper) { + return finalizedBlobSidecarsWrapper.slice(BLOB_SIDECARS_IN_WRAPPER_INDEX); + } + return null; + } + + async getDataColumnSidecars(blockSlot: Slot, blockRootHex: string): Promise { + const blockInput = this.seenBlockInputCache.get(blockRootHex); + if (blockInput) { + if (!isBlockInputColumns(blockInput)) { + throw new Error(`Expected block input to have columns: slot=${blockSlot} root=${blockRootHex}`); + } + return blockInput.getAllColumns(); + } + const sidecarsUnfinalized = await this.db.dataColumnSidecar.values(fromHex(blockRootHex)); + if (sidecarsUnfinalized.length > 0) { + return sidecarsUnfinalized; + } + const sidecarsFinalized = await this.db.dataColumnSidecarArchive.values(blockSlot); + return sidecarsFinalized; + } + + async getSerializedDataColumnSidecars( + blockSlot: Slot, + blockRootHex: string, + indices: number[] + ): Promise<(Uint8Array | undefined)[]> { + const blockInput = this.seenBlockInputCache.get(blockRootHex); + if (blockInput) { + if (!isBlockInputColumns(blockInput)) { + throw new Error(`Expected block input to have columns: slot=${blockSlot} root=${blockRootHex}`); + } + return indices.map((index) => { + const sidecar = blockInput.getColumn(index); + if (!sidecar) { + return undefined; + } + const serialized = this.serializedCache.get(sidecar); + if (serialized) { + return serialized; + } + return ssz.fulu.DataColumnSidecar.serialize(sidecar); + }); + } + const sidecarsUnfinalized = await this.db.dataColumnSidecar.getManyBinary(fromHex(blockRootHex), indices); + if (sidecarsUnfinalized.some((sidecar) => sidecar != null)) { + return sidecarsUnfinalized; + } + const sidecarsFinalized = await this.db.dataColumnSidecarArchive.getManyBinary(blockSlot, indices); + return sidecarsFinalized; + } + async produceCommonBlockBody(blockAttributes: BlockAttributes): Promise { const {slot, parentBlockRoot} = blockAttributes; const state = await this.regen.getBlockSlotState( diff --git a/packages/beacon-node/src/chain/interface.ts b/packages/beacon-node/src/chain/interface.ts index fe39d1e02d42..995e2c1b3563 100644 --- a/packages/beacon-node/src/chain/interface.ts +++ b/packages/beacon-node/src/chain/interface.ts @@ -22,6 +22,8 @@ import { Wei, altair, capella, + deneb, + fulu, phase0, rewards, } from "@lodestar/types"; @@ -193,12 +195,26 @@ export interface IBeaconChain { getCanonicalBlockAtSlot( slot: Slot ): Promise<{block: SignedBeaconBlock; executionOptimistic: boolean; finalized: boolean} | null>; + /** + * Get local block by root, does not fetch from the network + */ + getSerializedBlockByRoot( + root: RootHex + ): Promise<{block: Uint8Array; executionOptimistic: boolean; finalized: boolean; slot: Slot} | null>; /** * Get local block by root, does not fetch from the network */ getBlockByRoot( root: RootHex ): Promise<{block: SignedBeaconBlock; executionOptimistic: boolean; finalized: boolean} | null>; + getBlobSidecars(blockSlot: Slot, blockRootHex: string): Promise; + getSerializedBlobSidecars(blockSlot: Slot, blockRootHex: string): Promise; + getDataColumnSidecars(blockSlot: Slot, blockRootHex: string): Promise; + getSerializedDataColumnSidecars( + blockSlot: Slot, + blockRootHex: string, + indices: number[] + ): Promise<(Uint8Array | undefined)[]>; produceCommonBlockBody(blockAttributes: BlockAttributes): Promise; produceBlock(blockAttributes: BlockAttributes & {commonBlockBodyPromise: Promise}): Promise<{ diff --git a/packages/beacon-node/src/metrics/metrics/lodestar.ts b/packages/beacon-node/src/metrics/metrics/lodestar.ts index 2def2c231f42..753a4a524b47 100644 --- a/packages/beacon-node/src/metrics/metrics/lodestar.ts +++ b/packages/beacon-node/src/metrics/metrics/lodestar.ts @@ -212,6 +212,31 @@ export function createLodestarMetrics( }), }, + unfinalizedBlockWritesQueue: { + length: register.gauge({ + name: "lodestar_unfinalized_block_writes_queue_length", + help: "Count of total unfinalized block writes queue length", + }), + droppedJobs: register.gauge({ + name: "lodestar_unfinalized_block_writes_queue_dropped_jobs_total", + help: "Count of total unfinalized block writes queue dropped jobs", + }), + jobTime: register.histogram({ + name: "lodestar_unfinalized_block_writes_queue_job_time_seconds", + help: "Time to process unfinalized block writes queue job in seconds", + buckets: [0.01, 0.1, 1, 4, 12], + }), + jobWaitTime: register.histogram({ + name: "lodestar_unfinalized_block_writes_queue_job_wait_time_seconds", + help: "Time from job added to the unfinalized block writes queue to starting in seconds", + buckets: [0.01, 0.1, 1, 4, 12], + }), + concurrency: register.gauge({ + name: "lodestar_unfinalized_block_writes_queue_concurrency", + help: "Current concurrency of unfinalized block writes queue", + }), + }, + engineHttpProcessorQueue: { length: register.gauge({ name: "lodestar_engine_http_processor_queue_length", diff --git a/packages/beacon-node/src/network/processor/gossipHandlers.ts b/packages/beacon-node/src/network/processor/gossipHandlers.ts index 988834973fa2..29712cee4aa6 100644 --- a/packages/beacon-node/src/network/processor/gossipHandlers.ts +++ b/packages/beacon-node/src/network/processor/gossipHandlers.ts @@ -436,14 +436,11 @@ function getSequentialHandlers(modules: ValidatorFnsModules, options: GossipHand blsVerifyOnMainThread: true, // to track block process steps seenTimestampSec, - // gossip block is validated, we want to process it asap - eagerPersistBlock: true, }) .then(() => { // Returns the delay between the start of `block.slot` and `current time` const delaySec = chain.clock.secFromSlot(slot); metrics?.gossipBlock.elapsedTimeTillProcessed.observe(delaySec); - chain.seenBlockInputCache.prune(blockInput.blockRootHex); }) .catch((e) => { // Adjust verbosity based on error type diff --git a/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRange.ts b/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRange.ts index 4cfc43ea38a5..b4eda0db765e 100644 --- a/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRange.ts +++ b/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRange.ts @@ -4,7 +4,6 @@ import {GENESIS_SLOT, isForkPostDeneb, isForkPostFulu} from "@lodestar/params"; import {RespStatus, ResponseError, ResponseOutgoing} from "@lodestar/reqresp"; import {computeEpochAtSlot} from "@lodestar/state-transition"; import {deneb, phase0} from "@lodestar/types"; -import {fromHex} from "@lodestar/utils"; import {IBeaconChain} from "../../../chain/index.js"; import {IBeaconDb} from "../../../db/index.js"; import {prettyPrintPeerId} from "../../util.ts"; @@ -22,7 +21,6 @@ export async function* onBeaconBlocksByRange( const endSlot = startSlot + count; const finalized = db.blockArchive; - const unfinalized = db.block; // in the case of initializing from a non-finalized state, we don't have the finalized block so this api does not work // chain.forkChoice.getFinalizeBlock().slot const finalizedSlot = chain.forkChoice.getFinalizedCheckpointSlot(); @@ -65,7 +63,7 @@ export async function* onBeaconBlocksByRange( // re-org there's no need to abort the request // Spec: https://github.com/ethereum/consensus-specs/blob/a1e46d1ae47dd9d097725801575b46907c12a1f8/specs/eip4844/p2p-interface.md#blobssidecarsbyrange-v1 - const blockBytes = await unfinalized.getBinary(fromHex(block.blockRoot)); + const blockBytes = await chain.getSerializedBlockByRoot(block.blockRoot); if (!blockBytes) { throw new ResponseError( RespStatus.SERVER_ERROR, @@ -74,7 +72,7 @@ export async function* onBeaconBlocksByRange( } yield { - data: blockBytes, + data: blockBytes.block, boundary: chain.config.getForkBoundaryAtEpoch(computeEpochAtSlot(block.slot)), }; } diff --git a/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRoot.ts b/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRoot.ts index 211c03495809..362ae9267a3b 100644 --- a/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRoot.ts +++ b/packages/beacon-node/src/network/reqresp/handlers/beaconBlocksByRoot.ts @@ -1,48 +1,21 @@ import {ResponseOutgoing} from "@lodestar/reqresp"; import {computeEpochAtSlot} from "@lodestar/state-transition"; -import {Slot} from "@lodestar/types"; import {toRootHex} from "@lodestar/utils"; import {IBeaconChain} from "../../../chain/index.js"; -import {IBeaconDb} from "../../../db/index.js"; -import {getSlotFromSignedBeaconBlockSerialized} from "../../../util/sszBytes.js"; import {BeaconBlocksByRootRequest} from "../../../util/types.js"; export async function* onBeaconBlocksByRoot( requestBody: BeaconBlocksByRootRequest, - chain: IBeaconChain, - db: IBeaconDb + chain: IBeaconChain ): AsyncIterable { for (const blockRoot of requestBody) { const root = blockRoot; - const summary = chain.forkChoice.getBlock(root); - let blockBytes: Uint8Array | null = null; - - // finalized block has summary in forkchoice but it stays in blockArchive db - if (summary) { - blockBytes = await db.block.getBinary(root); - } - - let slot: Slot | undefined = undefined; - if (!blockBytes) { - const blockEntry = await db.blockArchive.getBinaryEntryByRoot(root); - if (blockEntry) { - slot = blockEntry.key; - blockBytes = blockEntry.value; - } - } - - if (blockBytes) { - if (slot === undefined) { - const slotFromBytes = getSlotFromSignedBeaconBlockSerialized(blockBytes); - if (slotFromBytes === null) { - throw Error(`Invalid block bytes for block root ${toRootHex(root)}`); - } - slot = slotFromBytes; - } + const block = await chain.getSerializedBlockByRoot(toRootHex(root)); + if (block) { yield { - data: blockBytes, - boundary: chain.config.getForkBoundaryAtEpoch(computeEpochAtSlot(slot)), + data: block.block, + boundary: chain.config.getForkBoundaryAtEpoch(computeEpochAtSlot(block.slot)), }; } } diff --git a/packages/beacon-node/src/network/reqresp/handlers/blobSidecarsByRoot.ts b/packages/beacon-node/src/network/reqresp/handlers/blobSidecarsByRoot.ts index 6416520a3ef6..751f9d4980ee 100644 --- a/packages/beacon-node/src/network/reqresp/handlers/blobSidecarsByRoot.ts +++ b/packages/beacon-node/src/network/reqresp/handlers/blobSidecarsByRoot.ts @@ -2,16 +2,13 @@ import {BLOB_SIDECAR_FIXED_SIZE} from "@lodestar/params"; import {RespStatus, ResponseError, ResponseOutgoing} from "@lodestar/reqresp"; import {computeEpochAtSlot} from "@lodestar/state-transition"; import {RootHex} from "@lodestar/types"; -import {fromHex, toRootHex} from "@lodestar/utils"; +import {toRootHex} from "@lodestar/utils"; import {IBeaconChain} from "../../../chain/index.js"; -import {IBeaconDb} from "../../../db/index.js"; -import {BLOB_SIDECARS_IN_WRAPPER_INDEX} from "../../../db/repositories/blobSidecars.js"; import {BlobSidecarsByRootRequest} from "../../../util/types.js"; export async function* onBlobSidecarsByRoot( requestBody: BlobSidecarsByRootRequest, - chain: IBeaconChain, - db: IBeaconDb + chain: IBeaconChain ): AsyncIterable { const finalizedSlot = chain.forkChoice.getFinalizedBlock().slot; @@ -34,14 +31,13 @@ export async function* onBlobSidecarsByRoot( // Check if we need to load sidecars for a new block root if (lastFetchedSideCars === null || lastFetchedSideCars.blockRoot !== blockRootHex) { - const blobSideCarsBytesWrapped = await db.blobSidecars.getBinary(fromHex(block.blockRoot)); - if (!blobSideCarsBytesWrapped) { + const blobSidecarsBytes = await chain.getSerializedBlobSidecars(block.slot, blockRootHex); + if (!blobSidecarsBytes) { // Handle the same to onBeaconBlocksByRange throw new ResponseError(RespStatus.SERVER_ERROR, `No item for root ${block.blockRoot} slot ${block.slot}`); } - const blobSideCarsBytes = blobSideCarsBytesWrapped.slice(BLOB_SIDECARS_IN_WRAPPER_INDEX); - lastFetchedSideCars = {blockRoot: blockRootHex, bytes: blobSideCarsBytes}; + lastFetchedSideCars = {blockRoot: blockRootHex, bytes: blobSidecarsBytes}; } const blobSidecarBytes = lastFetchedSideCars.bytes.slice( diff --git a/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRange.ts b/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRange.ts index 457eb0296917..3610b0136e8b 100644 --- a/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRange.ts +++ b/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRange.ts @@ -38,7 +38,6 @@ export async function* onDataColumnSidecarsByRange( } const finalized = db.dataColumnSidecarArchive; - const unfinalized = db.dataColumnSidecar; const finalizedSlot = chain.forkChoice.getFinalizedBlock().slot; // Finalized range of columns @@ -92,7 +91,11 @@ export async function* onDataColumnSidecarsByRange( // at the time of the start of the request. Spec is clear the chain of columns must be consistent, but on // re-org there's no need to abort the request // Spec: https://github.com/ethereum/consensus-specs/blob/ad36024441cf910d428d03f87f331fbbd2b3e5f1/specs/fulu/p2p-interface.md#L425-L429 - const dataColumnSidecars = await unfinalized.getManyBinary(fromHex(block.blockRoot), availableColumns); + const dataColumnSidecars = await chain.getSerializedDataColumnSidecars( + block.slot, + block.blockRoot, + availableColumns + ); const unavailableColumnIndices: ColumnIndex[] = []; for (let i = 0; i < dataColumnSidecars.length; i++) { diff --git a/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRoot.ts b/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRoot.ts index 24c88fd499d9..8e598ebe5a4d 100644 --- a/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRoot.ts +++ b/packages/beacon-node/src/network/reqresp/handlers/dataColumnSidecarsByRoot.ts @@ -61,11 +61,7 @@ export async function* onDataColumnSidecarsByRoot( continue; } - const dataColumns = block - ? // Non-finalized sidecars are stored by block root - await db.dataColumnSidecar.getManyBinary(blockRoot, availableColumns) - : // Finalized sidecars are archived and stored by slot - await db.dataColumnSidecarArchive.getManyBinary(slot, availableColumns); + const dataColumns = await chain.getSerializedDataColumnSidecars(slot, blockRootHex, availableColumns); const unavailableColumnIndices: ColumnIndex[] = []; for (let i = 0; i < dataColumns.length; i++) { diff --git a/packages/beacon-node/src/network/reqresp/handlers/index.ts b/packages/beacon-node/src/network/reqresp/handlers/index.ts index 985a34c8f40b..9777bbf23964 100644 --- a/packages/beacon-node/src/network/reqresp/handlers/index.ts +++ b/packages/beacon-node/src/network/reqresp/handlers/index.ts @@ -42,12 +42,12 @@ export function getReqRespHandlers({db, chain}: {db: IBeaconDb; chain: IBeaconCh [ReqRespMethod.BeaconBlocksByRoot]: (req) => { const fork = chain.config.getForkName(chain.clock.currentSlot); const body = BeaconBlocksByRootRequestType(fork, chain.config).deserialize(req.data); - return onBeaconBlocksByRoot(body, chain, db); + return onBeaconBlocksByRoot(body, chain); }, [ReqRespMethod.BlobSidecarsByRoot]: (req) => { const fork = chain.config.getForkName(chain.clock.currentSlot); const body = BlobSidecarsByRootRequestType(fork, chain.config).deserialize(req.data); - return onBlobSidecarsByRoot(body, chain, db); + return onBlobSidecarsByRoot(body, chain); }, [ReqRespMethod.BlobSidecarsByRange]: (req) => { const body = ssz.deneb.BlobSidecarsByRangeRequest.deserialize(req.data); diff --git a/packages/beacon-node/src/sync/range/chain.ts b/packages/beacon-node/src/sync/range/chain.ts index b0d07b1011e8..911ce93b5bb0 100644 --- a/packages/beacon-node/src/sync/range/chain.ts +++ b/packages/beacon-node/src/sync/range/chain.ts @@ -585,7 +585,6 @@ export class SyncChain { if (!res.err) { batch.processingSuccess(); - this.pruneBlockInputs(batch.getBlocks()); // If the processed batch is not empty, validate previous AwaitingValidation blocks. if (blocks.length > 0) { diff --git a/packages/beacon-node/src/sync/range/range.ts b/packages/beacon-node/src/sync/range/range.ts index 311311beae80..dbd5213ae4ef 100644 --- a/packages/beacon-node/src/sync/range/range.ts +++ b/packages/beacon-node/src/sync/range/range.ts @@ -188,9 +188,6 @@ export class RangeSync extends (EventEmitter as {new (): RangeSyncEmitter}) { // when this runs, syncing is the most important thing and gossip is not likely to run // so we can utilize worker threads to verify signatures blsVerifyOnMainThread: false, - // we want to be safe to only persist blocks after verifying it to avoid any attacks that may cause our DB - // to grow too much - eagerPersistBlock: false, }; if (this.opts?.disableProcessAsChainSegment) { diff --git a/packages/beacon-node/src/sync/unknownBlock.ts b/packages/beacon-node/src/sync/unknownBlock.ts index ba507a039d58..ad3e5cbc2ad9 100644 --- a/packages/beacon-node/src/sync/unknownBlock.ts +++ b/packages/beacon-node/src/sync/unknownBlock.ts @@ -423,8 +423,6 @@ export class BlockInputSync { // see https://github.com/ChainSafe/lodestar/issues/5650 ignoreIfFinalized: true, blsVerifyOnMainThread: true, - // block is validated with correct root, we want to process it as soon as possible - eagerPersistBlock: true, }) ); @@ -434,7 +432,6 @@ export class BlockInputSync { if (!res.err) { // no need to update status to "processed", delete anyway this.pendingBlocks.delete(pendingBlock.blockInput.blockRootHex); - this.chain.seenBlockInputCache.prune(pendingBlock.blockInput.blockRootHex); // Send child blocks to the processor for (const descendantBlock of getDescendantBlocks(pendingBlock.blockInput.blockRootHex, this.pendingBlocks)) { diff --git a/packages/db/src/abstractPrefixedRepository.ts b/packages/db/src/abstractPrefixedRepository.ts index 744d91255d59..76cf6b58e0d2 100644 --- a/packages/db/src/abstractPrefixedRepository.ts +++ b/packages/db/src/abstractPrefixedRepository.ts @@ -172,6 +172,34 @@ export abstract class PrefixedRepository { await this.db.batch(batchWithKeys, this.dbReqOpts); } + async values(prefix: P | P[]): Promise { + const result: T[] = []; + for (const p of Array.isArray(prefix) ? prefix : [prefix]) { + for await (const vb of this.db.valuesStream({ + gte: this.wrapKey(this.getMinKeyRaw(p)), + lte: this.wrapKey(this.getMaxKeyRaw(p)), + bucketId: this.bucketId, + })) { + result.push(this.decodeValue(vb)); + } + } + return result; + } + + async valuesBinary(prefix: P | P[]): Promise { + const result: Uint8Array[] = []; + for (const p of Array.isArray(prefix) ? prefix : [prefix]) { + for await (const vb of this.db.valuesStream({ + gte: this.wrapKey(this.getMinKeyRaw(p)), + lte: this.wrapKey(this.getMaxKeyRaw(p)), + bucketId: this.bucketId, + })) { + result.push(vb); + } + } + return result; + } + async *valuesStream(prefix: P | P[]): AsyncIterable { for (const p of Array.isArray(prefix) ? prefix : [prefix]) { for await (const vb of this.db.valuesStream({ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 539049bf6aa6..6960b30e9239 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1233,10 +1233,6 @@ packages: resolution: {integrity: sha512-4A/SCr/2KLd5jrtOMFzaKjVtAei3+2r/NChoBNoZ3EyP/+GlhoaEGoWOZUmFmoITP7zOJyHIMm+DYRd8o3PvHA==} engines: {node: '>=6.9.0'} - '@babel/helper-validator-identifier@7.25.9': - resolution: {integrity: sha512-Ed61U6XJc3CVRfkERJWDz4dJwKe7iLmmJsbOGu9wSloNSFttHV0I8g6UAgb7qnK5ly5bGLPd4oXZlxCdANBOWQ==} - engines: {node: '>=6.9.0'} - '@babel/helper-validator-identifier@7.28.5': resolution: {integrity: sha512-qSs4ifwzKJSV39ucNjsvc6WVHs6b7S03sOh2OcHF9UHfVPqWWALUsNUVzhSBiItjRZoLHx7nIarVjqKVusUZ1Q==} engines: {node: '>=6.9.0'} @@ -7543,8 +7539,6 @@ snapshots: '@babel/helper-string-parser@7.25.9': {} - '@babel/helper-validator-identifier@7.25.9': {} - '@babel/helper-validator-identifier@7.28.5': {} '@babel/parser@7.26.7': @@ -7554,7 +7548,7 @@ snapshots: '@babel/types@7.26.7': dependencies: '@babel/helper-string-parser': 7.25.9 - '@babel/helper-validator-identifier': 7.25.9 + '@babel/helper-validator-identifier': 7.28.5 '@bcoe/v8-coverage@1.0.2': {}