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 ed12daa15ab0..a43320eae36e 100644 --- a/packages/beacon-node/src/api/impl/beacon/blocks/index.ts +++ b/packages/beacon-node/src/api/impl/beacon/blocks/index.ts @@ -128,7 +128,8 @@ export function getBeaconBlockApi({ forkName: fork, sampledColumns: chain.custodyConfig.sampledColumns, custodyColumns: chain.custodyConfig.custodyColumns, - timeCreatedSec: seenTimestampSec, + seenTimestampSec, + source: PayloadEnvelopeInputSource.api, }); } diff --git a/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/payloadEnvelopeInput.ts b/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/payloadEnvelopeInput.ts index 7eeee4f33240..b69aae865c6f 100644 --- a/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/payloadEnvelopeInput.ts +++ b/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/payloadEnvelopeInput.ts @@ -9,6 +9,7 @@ import { ColumnWithSource, CreateFromBidProps, CreateFromBlockProps, + PayloadEnvelopeInputSource, SourceMeta, } from "./types.js"; @@ -71,6 +72,7 @@ export class PayloadEnvelopeInput { readonly bid: gloas.ExecutionPayloadBid; readonly versionedHashes: VersionedHashes; readonly daOutOfRange: boolean; + readonly source: PayloadEnvelopeInputSource; private columnsCache = new Map(); @@ -95,6 +97,7 @@ export class PayloadEnvelopeInput { custodyColumns: ColumnIndex[]; timeCreatedSec: number; daOutOfRange: boolean; + source: PayloadEnvelopeInputSource; }) { this.blockRootHex = props.blockRootHex; this.slot = props.slot; @@ -106,6 +109,7 @@ export class PayloadEnvelopeInput { this.custodyColumns = props.custodyColumns; this.timeCreatedSec = props.timeCreatedSec; this.daOutOfRange = props.daOutOfRange; + this.source = props.source; this.payloadEnvelopeDataPromise = createPromise(); this.allDataPromise = createPromise(); this.columnsDataPromise = createPromise(); @@ -133,8 +137,9 @@ export class PayloadEnvelopeInput { bid, sampledColumns: props.sampledColumns, custodyColumns: props.custodyColumns, - timeCreatedSec: props.timeCreatedSec, + timeCreatedSec: props.seenTimestampSec, daOutOfRange: props.daOutOfRange, + source: props.source, }); } @@ -153,8 +158,9 @@ export class PayloadEnvelopeInput { bid: props.bid, sampledColumns: props.sampledColumns, custodyColumns: props.custodyColumns, - timeCreatedSec: props.timeCreatedSec, + timeCreatedSec: props.seenTimestampSec, daOutOfRange: props.daOutOfRange, + source: props.source, }); } diff --git a/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/types.ts b/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/types.ts index 0b2a544d613b..a2ffa2ed9a79 100644 --- a/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/types.ts +++ b/packages/beacon-node/src/chain/blocks/payloadEnvelopeInput/types.ts @@ -7,9 +7,24 @@ export enum PayloadEnvelopeInputSource { engine = "engine", byRange = "req_resp_by_range", byRoot = "req_resp_by_root", + // Data-column reconstruction (KZG cell recovery), NOT a cache reload recovery = "recovery", + // Entry reconstructed from the hot DB by SeenPayloadEnvelopeInput.getOrReload + reload = "reload", + // Entry seeded from a checkpoint anchor state's latestExecutionPayloadBid at chain init + anchorState = "anchor_state", } +/** + * Reason a PayloadEnvelopeInput is evicted from SeenPayloadEnvelopeInput. Used for the `pruned` metric + * label and the eviction log. + * - belowParent: pruned below the new head's parent (canonical, FULL, all-columns) + * - finalized: below the finalized slot + * - prune: explicit prune by root + * - cap: insertion-order backstop cap (MAX_PAYLOAD_ENVELOPE_INPUT_CACHE_SIZE) + */ +export type PayloadEnvelopeInputPruneReason = "belowParent" | "finalized" | "prune" | "cap"; + export type SourceMeta = { source: PayloadEnvelopeInputSource; seenTimestampSec: number; @@ -20,13 +35,12 @@ export type ColumnWithSource = SourceMeta & { columnSidecar: gloas.DataColumnSidecar; }; -export type CreateFromBlockProps = { +export type CreateFromBlockProps = SourceMeta & { blockRootHex: RootHex; block: SignedBeaconBlock; forkName: ForkName; sampledColumns: ColumnIndex[]; custodyColumns: ColumnIndex[]; - timeCreatedSec: number; daOutOfRange: boolean; }; @@ -35,7 +49,7 @@ export type CreateFromBlockProps = { * the chain from a checkpoint anchor state — we have the bid via the state but not the * full SignedBeaconBlock). */ -export type CreateFromBidProps = { +export type CreateFromBidProps = SourceMeta & { blockRootHex: RootHex; slot: number; forkName: ForkName; @@ -43,7 +57,6 @@ export type CreateFromBidProps = { bid: gloas.ExecutionPayloadBid; sampledColumns: ColumnIndex[]; custodyColumns: ColumnIndex[]; - timeCreatedSec: number; daOutOfRange: boolean; }; diff --git a/packages/beacon-node/src/chain/chain.ts b/packages/beacon-node/src/chain/chain.ts index b2a4ad758f88..9e04a4c30da8 100644 --- a/packages/beacon-node/src/chain/chain.ts +++ b/packages/beacon-node/src/chain/chain.ts @@ -79,6 +79,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 {PayloadEnvelopeInputSource} from "./blocks/payloadEnvelopeInput/index.js"; import {PayloadEnvelopeProcessor} from "./blocks/payloadEnvelopeProcessor.js"; import {ImportPayloadOpts} from "./blocks/types.js"; import {persistBlockInput} from "./blocks/writeBlockInputToDb.js"; @@ -451,6 +452,9 @@ export class BeaconChain implements IBeaconChain { chainEvents: emitter, signal, serializedCache: this.serializedCache, + db, + seenBlockInputCache: this.seenBlockInputCache, + custodyConfig: this.custodyConfig, metrics, logger, }); @@ -466,7 +470,8 @@ export class BeaconChain implements IBeaconChain { bid: anchorBid, sampledColumns: this.custodyConfig.sampledColumns, custodyColumns: this.custodyConfig.custodyColumns, - timeCreatedSec: Math.floor(Date.now() / 1000), + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.anchorState, }); } diff --git a/packages/beacon-node/src/chain/seenCache/seenPayloadEnvelopeInput.ts b/packages/beacon-node/src/chain/seenCache/seenPayloadEnvelopeInput.ts index 8d21fde1764d..a327ee3693aa 100644 --- a/packages/beacon-node/src/chain/seenCache/seenPayloadEnvelopeInput.ts +++ b/packages/beacon-node/src/chain/seenCache/seenPayloadEnvelopeInput.ts @@ -1,18 +1,32 @@ import {ChainForkConfig} from "@lodestar/config"; import {CheckpointWithHex, IForkChoice, PayloadStatus, ProtoBlock} from "@lodestar/fork-choice"; +import {ForkPostGloas, SLOTS_PER_EPOCH, isForkPostGloas} from "@lodestar/params"; import {computeStartSlotAtEpoch} from "@lodestar/state-transition"; -import {RootHex} from "@lodestar/types"; -import {Logger} from "@lodestar/utils"; +import {RootHex, SignedBeaconBlock} from "@lodestar/types"; +import {Logger, fromHex} from "@lodestar/utils"; +import {IBeaconDb} from "../../db/index.js"; import {Metrics} from "../../metrics/metrics.js"; +import {MAX_LOOK_AHEAD_EPOCHS} from "../../sync/constants.js"; import {IClock} from "../../util/clock.js"; +import {CustodyConfig} from "../../util/dataColumns.js"; import {SerializedCache} from "../../util/serializedCache.js"; import {isDaOutOfRange} from "../blocks/blockInput/index.js"; -import {CreateFromBidProps, CreateFromBlockProps, PayloadEnvelopeInput} from "../blocks/payloadEnvelopeInput/index.js"; +import { + CreateFromBidProps, + CreateFromBlockProps, + PayloadEnvelopeInput, + PayloadEnvelopeInputPruneReason, + PayloadEnvelopeInputSource, +} from "../blocks/payloadEnvelopeInput/index.js"; import {ChainEvent, ChainEventEmitter} from "../emitter.js"; +import {SeenBlockInput} from "./seenGossipBlockInput.js"; export type {PayloadEnvelopeInputState} from "../blocks/payloadEnvelopeInput/index.js"; export {PayloadEnvelopeInput} from "../blocks/payloadEnvelopeInput/index.js"; +// Bound this cache by this max size, this is the same to SeenBlockInput +const MAX_PAYLOAD_ENVELOPE_INPUT_CACHE_SIZE = (MAX_LOOK_AHEAD_EPOCHS + 1) * SLOTS_PER_EPOCH; + export type SeenPayloadEnvelopeInputModules = { config: ChainForkConfig; clock: IClock; @@ -20,6 +34,9 @@ export type SeenPayloadEnvelopeInputModules = { chainEvents: ChainEventEmitter; signal: AbortSignal; serializedCache: SerializedCache; + db: IBeaconDb; + seenBlockInputCache: SeenBlockInput; + custodyConfig: CustodyConfig; metrics: Metrics | null; logger?: Logger; }; @@ -40,9 +57,14 @@ export class SeenPayloadEnvelopeInput { private readonly chainEvents: ChainEventEmitter; private readonly signal: AbortSignal; private readonly serializedCache: SerializedCache; + private readonly db: IBeaconDb; + private readonly seenBlockInputCache: SeenBlockInput; + private readonly custodyConfig: CustodyConfig; private readonly metrics: Metrics | null; private readonly logger?: Logger; private payloadInputs = new Map(); + // Dedup concurrent DB reloads of the same root so callers share one reconstructed object. + private readonly reloading = new Map>(); constructor({ config, @@ -51,6 +73,9 @@ export class SeenPayloadEnvelopeInput { chainEvents, signal, serializedCache, + db, + seenBlockInputCache, + custodyConfig, metrics, logger, }: SeenPayloadEnvelopeInputModules) { @@ -60,6 +85,9 @@ export class SeenPayloadEnvelopeInput { this.chainEvents = chainEvents; this.signal = signal; this.serializedCache = serializedCache; + this.db = db; + this.seenBlockInputCache = seenBlockInputCache; + this.custodyConfig = custodyConfig; this.metrics = metrics; this.logger = logger; @@ -86,7 +114,7 @@ export class SeenPayloadEnvelopeInput { let deletedCount = 0; for (const [, input] of this.payloadInputs) { if (input.slot < finalizedSlot) { - this.evictPayloadInput(input); + this.evictPayloadInput(input, "finalized"); deletedCount++; } } @@ -110,12 +138,13 @@ export class SeenPayloadEnvelopeInput { const daOutOfRange = isDaOutOfRange(this.config, props.forkName, props.block.message.slot, this.clock.currentEpoch); const input = PayloadEnvelopeInput.createFromBlock({...props, daOutOfRange}); this.payloadInputs.set(props.blockRootHex, input); - this.metrics?.seenCache.payloadEnvelopeInput.created.inc(); + this.metrics?.seenCache.payloadEnvelopeInput.created.inc({source: props.source}); this.logger?.verbose("SeenPayloadEnvelopeInput.add created new entry", { slot: input.slot, root: props.blockRootHex, daOutOfRange, }); + this.pruneToMaxSize(); return input; } @@ -131,12 +160,13 @@ export class SeenPayloadEnvelopeInput { const daOutOfRange = isDaOutOfRange(this.config, props.forkName, props.slot, this.clock.currentEpoch); const input = PayloadEnvelopeInput.createFromBid({...props, daOutOfRange}); this.payloadInputs.set(props.blockRootHex, input); - this.metrics?.seenCache.payloadEnvelopeInput.created.inc(); + this.metrics?.seenCache.payloadEnvelopeInput.created.inc({source: props.source}); this.logger?.verbose("SeenPayloadEnvelopeInput.addFromBid created new entry", { slot: input.slot, root: props.blockRootHex, daOutOfRange, }); + this.pruneToMaxSize(); return input; } @@ -144,6 +174,80 @@ export class SeenPayloadEnvelopeInput { return this.payloadInputs.get(blockRootHex); } + /** + * Like `get()`, but on a cache miss reconstruct the shell (bid + versionedHashes) from the block + * in `seenBlockInputCache` or the hot DB. + * This api is meant for BlockInputSync when a late/weird payloads for old blocks + * + * NOTE: the reconstructed entry is always EMPTY even when the block is actually FULL (its + * payload envelope + columns persisted in the DB). The consumer should consult fork choice if it needs to. + */ + async getOrReload(blockRootHex: RootHex): Promise { + const existing = this.payloadInputs.get(blockRootHex); + if (existing !== undefined) { + return existing; + } + + // Without this dedup, two concurrent misses each db.block.get + createFromBlock a DIFFERENT shell + // for the same root; processPayloadEnvelopeJob's WeakMap (keyed by object) can't dedup the import. + const inflight = this.reloading.get(blockRootHex); + if (inflight !== undefined) { + return inflight; + } + const promise = this.reloadFromDb(blockRootHex); + this.reloading.set(blockRootHex, promise); + try { + return await promise; + } finally { + this.reloading.delete(blockRootHex); + } + } + + private async reloadFromDb(blockRootHex: RootHex): Promise { + // Only recover unfinalized, fork-choice-known blocks. Do not read the finalized archive. The gate + // also filters out un-imported blocks that may sit in seenBlockInputCache before validation. + if (!this.forkChoice.hasBlockHex(blockRootHex)) { + return undefined; + } + + // In-memory first: persistBlockInput writes db.block THEN prunes seenBlockInputCache, so a + // just-imported block whose async write (unfinalizedBlockWrites) has not flushed yet is still here; + // older blocks (already pruned from the cache) fall through to the hot db. + const cachedBlockInput = this.seenBlockInputCache.get(blockRootHex); + const block = cachedBlockInput?.hasBlock() + ? cachedBlockInput.getBlock() + : await this.db.block.get(fromHex(blockRootHex)); + if (block == null) { + return undefined; + } + + const forkName = this.config.getForkName(block.message.slot); + if (!isForkPostGloas(forkName)) { + return undefined; + } + + const daOutOfRange = isDaOutOfRange(this.config, forkName, block.message.slot, this.clock.currentEpoch); + const input = PayloadEnvelopeInput.createFromBlock({ + blockRootHex, + block: block as SignedBeaconBlock, + forkName, + sampledColumns: this.custodyConfig.sampledColumns, + custodyColumns: this.custodyConfig.custodyColumns, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.reload, + daOutOfRange, + }); + this.payloadInputs.set(blockRootHex, input); + this.metrics?.seenCache.payloadEnvelopeInput.created.inc({source: PayloadEnvelopeInputSource.reload}); + this.logger?.verbose("SeenPayloadEnvelopeInput.getOrReload reconstructed entry from db", { + slot: input.slot, + root: blockRootHex, + }); + // Set above (entry is at the back of the insertion order), so the cap never evicts what we just reloaded. + this.pruneToMaxSize(); + return input; + } + hasPayload(blockRootHex: RootHex): boolean { return this.payloadInputs.get(blockRootHex)?.hasPayloadEnvelope() ?? false; } @@ -155,8 +259,7 @@ export class SeenPayloadEnvelopeInput { prune(blockRootHex: RootHex): void { const input = this.payloadInputs.get(blockRootHex); if (input) { - this.evictPayloadInput(input); - this.logger?.verbose("SeenPayloadEnvelopeInput.prune deleted", {slot: input.slot, root: blockRootHex}); + this.evictPayloadInput(input, "prune"); } } @@ -170,17 +273,45 @@ export class SeenPayloadEnvelopeInput { // ...and don't evict while columns are still being gathered: writeDataColumnsToDb awaits the // same hasComputedAllData() before persisting. Such entries are pruned by a later call. if (input?.hasComputedAllData()) { - this.evictPayloadInput(input); - this.logger?.verbose("SeenPayloadEnvelopeInput.pruneBelowParent deleted", { - slot: block.slot, - root: block.blockRoot, - }); + this.evictPayloadInput(input, "belowParent"); } } } } - private evictPayloadInput(payloadInput: PayloadEnvelopeInput): void { + /** + * Backstop cap for entries pruneBelowParent/pruneFinalized can't reach (non-canonical forks, + * EMPTY/PENDING entries). Evicts by INSERTION ORDER — the Map iterates oldest-inserted first — so a + * just-reloaded old-slot entry (set at the back) survives while genuinely stale forks/shells are shed. + * Runs after every single insert, so it evicts ~1 per call. Safe because the shared cache is + * non-load-bearing: range sync reads its batch map, gossip/BlockInputSync recover via getOrReload, + * and any evicted heavy data is already in the db. + */ + private pruneToMaxSize(): void { + let evicted = 0; + for (const input of this.payloadInputs.values()) { + if (this.payloadInputs.size <= MAX_PAYLOAD_ENVELOPE_INPUT_CACHE_SIZE) { + break; + } + this.evictPayloadInput(input, "cap"); + evicted++; + } + if (evicted > 0) { + this.logger?.debug("SeenPayloadEnvelopeInput.pruneToMaxSize evicted", { + evicted, + size: this.payloadInputs.size, + max: MAX_PAYLOAD_ENVELOPE_INPUT_CACHE_SIZE, + }); + } + } + + private evictPayloadInput(payloadInput: PayloadEnvelopeInput, reason: PayloadEnvelopeInputPruneReason): void { + this.metrics?.seenCache.payloadEnvelopeInput.pruned.inc({reason}); + this.logger?.debug("SeenPayloadEnvelopeInput evicted", { + slot: payloadInput.slot, + root: payloadInput.blockRootHex, + reason, + }); this.serializedCache.delete(payloadInput.getSerializedCacheKeys()); this.payloadInputs.delete(payloadInput.blockRootHex); } diff --git a/packages/beacon-node/src/metrics/metrics/lodestar.ts b/packages/beacon-node/src/metrics/metrics/lodestar.ts index 7bdf6b2e435d..d1734e155859 100644 --- a/packages/beacon-node/src/metrics/metrics/lodestar.ts +++ b/packages/beacon-node/src/metrics/metrics/lodestar.ts @@ -4,7 +4,10 @@ import {ArchiveStoreTask} from "../../chain/archiveStore/archiveStore.js"; import {FrequencyStateArchiveStep} from "../../chain/archiveStore/strategies/frequencyStateArchiveStrategy.js"; import {BlockInputSource} from "../../chain/blocks/blockInput/index.js"; import {PayloadErrorCode} from "../../chain/blocks/importExecutionPayload.js"; -import {PayloadEnvelopeInputSource} from "../../chain/blocks/payloadEnvelopeInput/index.js"; +import { + PayloadEnvelopeInputPruneReason, + PayloadEnvelopeInputSource, +} from "../../chain/blocks/payloadEnvelopeInput/index.js"; import {JobQueueItemType} from "../../chain/bls/index.js"; import {AttestationErrorCode, BlockErrorCode} from "../../chain/errors/index.js"; import { @@ -1643,9 +1646,15 @@ export function createLodestarMetrics( name: "lodestar_seen_payload_envelope_input_cache_serialized_object_refs", help: "Number of serialized-cache object refs retained by cached PayloadEnvelopeInputs", }), - created: register.counter({ + created: register.counter<{source: PayloadEnvelopeInputSource}>({ name: "lodestar_seen_payload_envelope_input_cache_items_created_total", - help: "Number of PayloadEnvelopeInputs created", + help: "Number of PayloadEnvelopeInputs created by source", + labelNames: ["source"], + }), + pruned: register.counter<{reason: PayloadEnvelopeInputPruneReason}>({ + name: "lodestar_seen_payload_envelope_input_cache_items_pruned_total", + help: "Number of PayloadEnvelopeInputs evicted by reason", + labelNames: ["reason"], }), }, }, diff --git a/packages/beacon-node/src/network/processor/gossipHandlers.ts b/packages/beacon-node/src/network/processor/gossipHandlers.ts index aba3f78f646c..efd660b9ef55 100644 --- a/packages/beacon-node/src/network/processor/gossipHandlers.ts +++ b/packages/beacon-node/src/network/processor/gossipHandlers.ts @@ -196,7 +196,8 @@ function getSequentialHandlers(modules: ValidatorFnsModules, options: GossipHand forkName: fork, sampledColumns: chain.custodyConfig.sampledColumns, custodyColumns: chain.custodyConfig.custodyColumns, - timeCreatedSec: seenTimestampSec, + seenTimestampSec, + source: PayloadEnvelopeInputSource.gossip, }); } @@ -640,6 +641,7 @@ function getSequentialHandlers(modules: ValidatorFnsModules, options: GossipHand metrics?.gossipBlock.elapsedTimeTillProcessed.observe(delaySec); if (isForkPostGloas(blockInput.forkName)) { + // we should have the payloadInput in the seen cache so no need getOrReload() here const payloadInput = chain.seenPayloadEnvelopeInputCache.get(blockInput.blockRootHex); // This payloadInput should have been created just after gossip validation if (!payloadInput) { @@ -1189,6 +1191,8 @@ function getSequentialHandlers(modules: ValidatorFnsModules, options: GossipHand chain.validatorMonitor?.registerExecutionPayloadEnvelope(OpSource.gossip, delaySec, signedEnvelope); const blockRootHex = toRootHex(envelope.beaconBlockRoot); + // a gossip payload cannot be `MAX_PAYLOAD_ENVELOPE_INPUT_CACHE_SIZE` slots after the block + // otherwise it'll get to UnknownBlockInput flow const payloadInput = chain.seenPayloadEnvelopeInputCache.get(blockRootHex); if (!payloadInput) { diff --git a/packages/beacon-node/src/sync/unknownBlock.ts b/packages/beacon-node/src/sync/unknownBlock.ts index 6e422795ad2f..090e8526cb11 100644 --- a/packages/beacon-node/src/sync/unknownBlock.ts +++ b/packages/beacon-node/src/sync/unknownBlock.ts @@ -544,15 +544,7 @@ export class BlockInputSync { return {kind: "invalidParentPayload", parentRootHex, parentBlockHashHex}; } - const parentPayloadInput = this.chain.seenPayloadEnvelopeInputCache.get(parentRootHex); - if (parentPayloadInput) { - if (parentPayloadInput.getBlockHashHex() === parentBlockHashHex) { - return {kind: "parentPayload", rootHex: parentRootHex, slot: parentBlock.slot}; - } - - return {kind: "invalidParentPayload", parentRootHex, parentBlockHashHex}; - } - + // Parent is in fork choice but its payload `parentBlockHashHex` isn't revealed yet return {kind: "parentPayload", rootHex: parentRootHex, slot: parentBlock.slot}; } @@ -894,9 +886,9 @@ export class BlockInputSync { this.pendingBlocks.delete(blockRootHex); if (isForkPostGloas(fork)) { - const payloadInput = this.chain.seenPayloadEnvelopeInputCache.get(blockRootHex); + const payloadInput = await this.chain.seenPayloadEnvelopeInputCache.getOrReload(blockRootHex); if (!payloadInput) { - this.logger.warn("PayloadEnvelopeInput not seeded during unknown sync processReadyBlock()", logCtx); + this.logger.debug("PayloadEnvelopeInput unavailable during unknown sync processReadyBlock()", logCtx); } else { // Similar to the gossip path: immediately attempt fetch of data columns from the execution engine. // The bid already carries the kzg commitments, so there is no reason to wait for the payload to arrive. @@ -969,12 +961,12 @@ export class BlockInputSync { */ private async reconcilePayloadEnvelope(pendingPayload: PendingPayloadEnvelope): Promise { const rootHex = getPayloadSyncCacheItemRootHex(pendingPayload); + if (this.chain.forkChoice.hasPayloadHexUnsafe(rootHex)) { this.pendingPayloads.delete(rootHex); return; } - const payloadInput = this.chain.seenPayloadEnvelopeInputCache.get(rootHex); if (!this.chain.forkChoice.hasBlockHex(rootHex)) { // Block not in fork choice yet. payloadInput may be seeded from the block body during download, so a // non-null payloadInput does not imply the block is imported; defer regardless and pull the block first. @@ -990,8 +982,9 @@ export class BlockInputSync { return; } + const payloadInput = await this.chain.seenPayloadEnvelopeInputCache.getOrReload(rootHex); if (!payloadInput) { - this.logger.debug("Missing PayloadEnvelopeInput for known block while reconciling payload envelope", { + this.logger.debug("PayloadEnvelopeInput not yet reloadable for imported block, will retry", { root: rootHex, }); return; @@ -1187,7 +1180,7 @@ export class BlockInputSync { let slot = getPayloadSyncCacheItemSlot(cacheItem); let payloadInput = isPendingPayloadInput(cacheItem) ? cacheItem.payloadInput - : this.chain.seenPayloadEnvelopeInputCache.get(rootHex); + : await this.chain.seenPayloadEnvelopeInputCache.getOrReload(rootHex); let envelope = payloadInput?.hasPayloadEnvelope() ? payloadInput.getPayloadEnvelope() : undefined; let i = 0; @@ -1220,7 +1213,7 @@ export class BlockInputSync { slot = envelope.message.payload.slotNumber; } - payloadInput ??= this.chain.seenPayloadEnvelopeInputCache.get(rootHex); + payloadInput ??= await this.chain.seenPayloadEnvelopeInputCache.getOrReload(rootHex); if (!this.chain.forkChoice.hasBlockHex(rootHex)) { // Block not in fork choice yet. Validating now would throw BLOCK_ROOT_UNKNOWN, so keep the downloaded // envelope and wait for the block body; reconcilePayloadEnvelope validates once the block lands. diff --git a/packages/beacon-node/src/sync/utils/downloadByRange.ts b/packages/beacon-node/src/sync/utils/downloadByRange.ts index 5fa704e3417f..2261c68a411d 100644 --- a/packages/beacon-node/src/sync/utils/downloadByRange.ts +++ b/packages/beacon-node/src/sync/utils/downloadByRange.ts @@ -136,6 +136,8 @@ export function cacheByRangeResponses({ const source = BlockInputSource.byRange; const updatedBatchBlocks = new Map(batchBlocks.map((block) => [block.slot, block])); + const payloadEnvelopes = new Map(existingPayloadEnvelopes); + const blocks = responses.validatedBlocks ?? []; for (let i = 0; i < blocks.length; i++) { const {block, blockRoot} = blocks[i]; @@ -172,15 +174,17 @@ export function cacheByRangeResponses({ // later step) can throw and abort the batch — otherwise a gloas block would sit in // seenBlockInputCache but unseeded here, and payload-by-root sync would later throw "Missing // PayloadEnvelopeInput for known block" (see issue #9306). add() is idempotent. - if (isForkPostGloas(blockInput.forkName)) { - seenPayloadEnvelopeInputCache.add({ + if (isForkPostGloas(blockInput.forkName) && !payloadEnvelopes.has(blockInput.slot)) { + const payloadInput = seenPayloadEnvelopeInputCache.add({ blockRootHex: blockInput.blockRootHex, block: blockInput.getBlock() as SignedBeaconBlock, forkName: blockInput.forkName, sampledColumns: custodyConfig.sampledColumns, custodyColumns: custodyConfig.custodyColumns, - timeCreatedSec: seenTimestampSec, + seenTimestampSec, + source: PayloadEnvelopeInputSource.byRange, }); + payloadEnvelopes.set(blockInput.slot, payloadInput); } } @@ -222,17 +226,22 @@ export function cacheByRangeResponses({ } } - let payloadEnvelopes: Map | null = - existingPayloadEnvelopes !== null ? new Map(existingPayloadEnvelopes) : null; if (downloadedPayloadEnvelopes !== null) { - payloadEnvelopes ??= new Map(); for (const [slot, envelope] of downloadedPayloadEnvelopes) { - const envelopeBlockRootHex = toRootHex(envelope.message.beaconBlockRoot); - const payloadInput = seenPayloadEnvelopeInputCache.get(envelopeBlockRootHex); + // the only PayloadEnvelopeInput miss is the dangling parent, which we can get from the seen cache + let payloadInput = payloadEnvelopes.get(slot); if (payloadInput === undefined) { - // Unreachable given the validatedBlocks loop above seeded an entry for every gloas block in - // the batch. for the parent block, it's populated at BeaconChain init - throw new Error(`Missing PayloadEnvelopeInput for block ${envelopeBlockRootHex}`); + if (updatedBatchBlocks.has(slot)) { + throw new Error( + `Missing PayloadEnvelopeInput for in-batch slot ${slot} root ${toRootHex(envelope.message.beaconBlockRoot)}` + ); + } + payloadInput = seenPayloadEnvelopeInputCache.get(toRootHex(envelope.message.beaconBlockRoot)); + } + if (payloadInput === undefined) { + throw new Error( + `Missing PayloadEnvelopeInput for slot ${slot} root ${toRootHex(envelope.message.beaconBlockRoot)}` + ); } if (!payloadInput.hasPayloadEnvelope()) { @@ -262,7 +271,7 @@ export function cacheByRangeResponses({ // Gloas columns are attached to the matching PayloadEnvelopeInput, NOT to IBlockInput. // Gloas DataColumnSidecar has `slot` directly (no signedBlockHeader). const dataSlot = firstColumn.slot; - const payloadInput = payloadEnvelopes?.get(dataSlot); + const payloadInput = payloadEnvelopes.get(dataSlot); if (!payloadInput) { // Should not happen: we built payloadInputs for all gloas blocks above continue; diff --git a/packages/beacon-node/src/sync/utils/downloadByRoot.ts b/packages/beacon-node/src/sync/utils/downloadByRoot.ts index 654a1ecfd4d5..d78c71992ac4 100644 --- a/packages/beacon-node/src/sync/utils/downloadByRoot.ts +++ b/packages/beacon-node/src/sync/utils/downloadByRoot.ts @@ -13,6 +13,7 @@ import {BlobIndex, ColumnIndex, SignedBeaconBlock, Slot, deneb, fulu} from "@lod import {LodestarError, byteArrayEquals, fromHex, prettyPrintIndices, toHex, toRootHex} from "@lodestar/utils"; import {isBlockInputBlobs, isBlockInputColumns} from "../../chain/blocks/blockInput/blockInput.js"; import {BlockInputSource, IBlockInput} from "../../chain/blocks/blockInput/types.js"; +import {PayloadEnvelopeInputSource} from "../../chain/blocks/payloadEnvelopeInput/index.js"; import {ChainEventEmitter} from "../../chain/emitter.js"; import {IBeaconChain} from "../../chain/interface.js"; import {validateBlockBlobSidecars} from "../../chain/validation/blobSidecar.js"; @@ -122,7 +123,8 @@ export async function downloadByRoot({ forkName: blockInput.forkName, sampledColumns: chain.custodyConfig.sampledColumns, custodyColumns: chain.custodyConfig.custodyColumns, - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.byRoot, }); } diff --git a/packages/beacon-node/test/e2e/sync/unknownBlockSync.test.ts b/packages/beacon-node/test/e2e/sync/unknownBlockSync.test.ts index 2a4d54700c4c..fa0de85686d8 100644 --- a/packages/beacon-node/test/e2e/sync/unknownBlockSync.test.ts +++ b/packages/beacon-node/test/e2e/sync/unknownBlockSync.test.ts @@ -7,6 +7,7 @@ import {SLOTS_PER_EPOCH} from "@lodestar/params"; import {gloas} from "@lodestar/types"; import {BlockInputNoData} from "../../../src/chain/blocks/blockInput/blockInput.js"; import {BlockInputSource} from "../../../src/chain/blocks/blockInput/types.js"; +import {PayloadEnvelopeInputSource} from "../../../src/chain/blocks/payloadEnvelopeInput/index.js"; import {ChainEvent} from "../../../src/chain/emitter.js"; import {BlockError, BlockErrorCode} from "../../../src/chain/errors/index.js"; import {INTEROP_BLOCK_HASH} from "../../../src/node/utils/interop/state.js"; @@ -187,7 +188,8 @@ describe("sync / unknown block sync thru gloas", () => { forkName: headInput.forkName, sampledColumns: bn2.chain.custodyConfig.sampledColumns, custodyColumns: bn2.chain.custodyConfig.custodyColumns, - timeCreatedSec: headInput.getTimeComplete(), + seenTimestampSec: headInput.getTimeComplete(), + source: PayloadEnvelopeInputSource.gossip, }); } const waitForPayloadImported = expectsPayloadImport diff --git a/packages/beacon-node/test/mocks/mockedBeaconChain.ts b/packages/beacon-node/test/mocks/mockedBeaconChain.ts index 86d42bec29f4..5e35df573beb 100644 --- a/packages/beacon-node/test/mocks/mockedBeaconChain.ts +++ b/packages/beacon-node/test/mocks/mockedBeaconChain.ts @@ -172,6 +172,7 @@ vi.mock("../../src/chain/chain.js", async (importActual) => { seenBlockInputCache: new SeenBlockInput(), seenPayloadEnvelopeInputCache: { get: vi.fn(), + getOrReload: vi.fn(), }, seenPayloadEnvelope: vi.fn(), shufflingCache: new ShufflingCache(), diff --git a/packages/beacon-node/test/spec/presets/fast_confirmation.test.ts b/packages/beacon-node/test/spec/presets/fast_confirmation.test.ts index 862dc63a2de7..e33657a430a0 100644 --- a/packages/beacon-node/test/spec/presets/fast_confirmation.test.ts +++ b/packages/beacon-node/test/spec/presets/fast_confirmation.test.ts @@ -49,6 +49,7 @@ import { BlockInputPreData, BlockInputSource, } from "../../../src/chain/blocks/blockInput/index.js"; +import {PayloadEnvelopeInputSource} from "../../../src/chain/blocks/payloadEnvelopeInput/types.ts"; import {AttestationImportOpt, BlobSidecarValidation} from "../../../src/chain/blocks/types.js"; import { verifyExecutionPayloadEnvelope, @@ -289,7 +290,8 @@ const fastConfirmationTest = forkName: fork, sampledColumns: chain.custodyConfig.sampledColumns, custodyColumns: chain.custodyConfig.custodyColumns, - timeCreatedSec: tickTime, + seenTimestampSec: tickTime, + source: PayloadEnvelopeInputSource.gossip, }); } else if (forkSeq >= ForkSeq.fulu) { if (columns === undefined) { diff --git a/packages/beacon-node/test/spec/utils/forkChoiceTestRunner.ts b/packages/beacon-node/test/spec/utils/forkChoiceTestRunner.ts index 566e93dd4b05..d713df432ae9 100644 --- a/packages/beacon-node/test/spec/utils/forkChoiceTestRunner.ts +++ b/packages/beacon-node/test/spec/utils/forkChoiceTestRunner.ts @@ -51,6 +51,7 @@ import { BlockInputPreData, BlockInputSource, } from "../../../src/chain/blocks/blockInput/index.js"; +import {PayloadEnvelopeInputSource} from "../../../src/chain/blocks/payloadEnvelopeInput/index.js"; import {AttestationImportOpt, BlobSidecarValidation} from "../../../src/chain/blocks/types.js"; import { verifyExecutionPayloadEnvelope, @@ -397,7 +398,8 @@ export const forkChoiceTestRunner = forkName: fork, sampledColumns: chain.custodyConfig.sampledColumns, custodyColumns: chain.custodyConfig.custodyColumns, - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.gossip, }); } else if (forkSeq >= ForkSeq.fulu) { if (columns === undefined) { diff --git a/packages/beacon-node/test/unit/api/impl/beacon/blocks/publishExecutionPayloadEnvelope.test.ts b/packages/beacon-node/test/unit/api/impl/beacon/blocks/publishExecutionPayloadEnvelope.test.ts index 6240ddc311d1..601a19895453 100644 --- a/packages/beacon-node/test/unit/api/impl/beacon/blocks/publishExecutionPayloadEnvelope.test.ts +++ b/packages/beacon-node/test/unit/api/impl/beacon/blocks/publishExecutionPayloadEnvelope.test.ts @@ -8,6 +8,7 @@ import {ssz} from "@lodestar/types"; import {fromHex, toRootHex} from "@lodestar/utils"; import {getBeaconBlockApi} from "../../../../../../src/api/impl/beacon/blocks/index.js"; import {PayloadEnvelopeInput} from "../../../../../../src/chain/blocks/payloadEnvelopeInput/payloadEnvelopeInput.js"; +import {PayloadEnvelopeInputSource} from "../../../../../../src/chain/blocks/payloadEnvelopeInput/types.js"; import {SeenBlockProposers} from "../../../../../../src/chain/seenCache/seenBlockProposers.js"; import {ApiTestModules, getApiTestModules} from "../../../../../utils/api.js"; import {generateProtoBlock} from "../../../../../utils/typeGenerator.js"; @@ -53,7 +54,8 @@ describe("api - beacon - publishExecutionPayloadEnvelope", () => { forkName: ForkName.gloas, sampledColumns: [], custodyColumns: [], - timeCreatedSec: 0, + seenTimestampSec: 0, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); const signedEnvelope = ssz.gloas.SignedExecutionPayloadEnvelope.defaultValue(); diff --git a/packages/beacon-node/test/unit/chain/blocks/utils/chainSegment.test.ts b/packages/beacon-node/test/unit/chain/blocks/utils/chainSegment.test.ts index d40312ecde79..9b7b17e81d0f 100644 --- a/packages/beacon-node/test/unit/chain/blocks/utils/chainSegment.test.ts +++ b/packages/beacon-node/test/unit/chain/blocks/utils/chainSegment.test.ts @@ -51,7 +51,8 @@ describe("chain / blocks / utils / chainSegment / assertLinearChainSegment with forkName: ForkName.gloas, sampledColumns: [], custodyColumns: [], - timeCreatedSec: seenTimestampSec, + seenTimestampSec, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); const envelope = ssz.gloas.SignedExecutionPayloadEnvelope.defaultValue(); @@ -203,7 +204,8 @@ describe("chain / blocks / utils / chainSegment / assertLinearChainSegment bound forkName: ForkName.gloas, sampledColumns: [], custodyColumns: [], - timeCreatedSec: seenTimestampSec, + seenTimestampSec, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); const envelope = ssz.gloas.SignedExecutionPayloadEnvelope.defaultValue(); diff --git a/packages/beacon-node/test/unit/chain/blocks/verifyPayloadsDataAvailability.test.ts b/packages/beacon-node/test/unit/chain/blocks/verifyPayloadsDataAvailability.test.ts index 462a1dea2c1b..583fc8c45c4b 100644 --- a/packages/beacon-node/test/unit/chain/blocks/verifyPayloadsDataAvailability.test.ts +++ b/packages/beacon-node/test/unit/chain/blocks/verifyPayloadsDataAvailability.test.ts @@ -36,7 +36,8 @@ function buildPayloadEnvelopeInput({ forkName: ForkName.gloas, sampledColumns, custodyColumns: sampledColumns, - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange, }); @@ -192,7 +193,8 @@ describe("PayloadEnvelopeInput.waitForEnvelopeAndAllData", () => { forkName: ForkName.gloas, sampledColumns, custodyColumns: sampledColumns, - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); diff --git a/packages/beacon-node/test/unit/chain/seenCache/seenPayloadEnvelopeInput.test.ts b/packages/beacon-node/test/unit/chain/seenCache/seenPayloadEnvelopeInput.test.ts index 44d577ee44b0..b989d27aa98d 100644 --- a/packages/beacon-node/test/unit/chain/seenCache/seenPayloadEnvelopeInput.test.ts +++ b/packages/beacon-node/test/unit/chain/seenCache/seenPayloadEnvelopeInput.test.ts @@ -1,14 +1,28 @@ import {beforeEach, describe, expect, it, vi} from "vitest"; import {ExecutionStatus, IForkChoice, PayloadStatus, ProtoBlock} from "@lodestar/fork-choice"; import {testLogger} from "@lodestar/logger/test-utils"; -import {ForkName} from "@lodestar/params"; +import {ForkName, SLOTS_PER_EPOCH} from "@lodestar/params"; import {DataAvailabilityStatus} from "@lodestar/state-transition"; import {RootHex} from "@lodestar/types"; +import {PayloadEnvelopeInputSource} from "../../../../src/chain/blocks/payloadEnvelopeInput/index.js"; import {ChainEventEmitter} from "../../../../src/chain/emitter.js"; +import {SeenBlockInput} from "../../../../src/chain/seenCache/seenGossipBlockInput.js"; import {SeenPayloadEnvelopeInput} from "../../../../src/chain/seenCache/seenPayloadEnvelopeInput.js"; +import {IBeaconDb} from "../../../../src/db/index.js"; +import {MAX_LOOK_AHEAD_EPOCHS} from "../../../../src/sync/constants.js"; +import {CustodyConfig} from "../../../../src/util/dataColumns.js"; import {SerializedCache} from "../../../../src/util/serializedCache.js"; import {getMockedClock} from "../../../mocks/clock.js"; -import {config, generateBlock, generateBlockWithColumnSidecars} from "../../../utils/blocksAndData.js"; +import { + FULU_FORK_EPOCH, + GLOAS_FORK_EPOCH, + config, + generateBlock, + generateBlockWithColumnSidecars, +} from "../../../utils/blocksAndData.js"; + +const GLOAS_SLOT = GLOAS_FORK_EPOCH * SLOTS_PER_EPOCH; +const FULU_SLOT = FULU_FORK_EPOCH * SLOTS_PER_EPOCH; describe("SeenPayloadEnvelopeInput", () => { let cache: SeenPayloadEnvelopeInput; @@ -16,14 +30,20 @@ describe("SeenPayloadEnvelopeInput", () => { let chainEvents: ChainEventEmitter; let forkChoice: IForkChoice; let serializedCache: SerializedCache; + let db: IBeaconDb; + let seenBlockInputCache: SeenBlockInput; beforeEach(() => { chainEvents = new ChainEventEmitter(); abortController = new AbortController(); forkChoice = { getAllAncestorBlocks: vi.fn(), + hasBlockHex: vi.fn(), } as unknown as IForkChoice; serializedCache = new SerializedCache(); + db = {block: {get: vi.fn()}} as unknown as IBeaconDb; + // Default: cache miss so getOrReload exercises the db path; individual tests override get(). + seenBlockInputCache = {get: vi.fn().mockReturnValue(undefined)} as unknown as SeenBlockInput; cache = new SeenPayloadEnvelopeInput({ config, @@ -32,6 +52,9 @@ describe("SeenPayloadEnvelopeInput", () => { chainEvents, signal: abortController.signal, serializedCache, + db, + seenBlockInputCache, + custodyConfig: {sampledColumns: [], custodyColumns: []} as unknown as CustodyConfig, metrics: null, logger: testLogger(), }); @@ -45,7 +68,8 @@ describe("SeenPayloadEnvelopeInput", () => { forkName: ForkName.gloas, sampledColumns: [], custodyColumns: [], - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.gossip, }); return rootHex; } @@ -60,7 +84,8 @@ describe("SeenPayloadEnvelopeInput", () => { forkName: ForkName.gloas, sampledColumns: [0, 1], custodyColumns: [0, 1], - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.gossip, }); return rootHex; } @@ -144,7 +169,8 @@ describe("SeenPayloadEnvelopeInput", () => { forkName: ForkName.gloas, sampledColumns: [], custodyColumns: [], - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.gossip, }; const first = cache.add(props); @@ -172,4 +198,175 @@ describe("SeenPayloadEnvelopeInput", () => { expect(cache.get(rootHex)).toBeDefined(); expect(cache.size()).toBe(1); }); + + describe("getOrReload", () => { + it("returns the in-memory entry without touching fork choice or db", async () => { + const rootHex = addPayloadInput(1); + + const result = await cache.getOrReload(rootHex); + + expect(result).toBe(cache.get(rootHex)); + expect(forkChoice.hasBlockHex).not.toHaveBeenCalled(); + expect(db.block.get).not.toHaveBeenCalled(); + }); + + it("returns undefined and does not read db when the block is not in fork choice", async () => { + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(false); + + const result = await cache.getOrReload(`0x${"ab".repeat(32)}`); + + expect(result).toBeUndefined(); + expect(db.block.get).not.toHaveBeenCalled(); + }); + + it("reconstructs the shell from the hot db when the gloas block is in fork choice", async () => { + const {block, rootHex} = generateBlock({forkName: ForkName.gloas, slot: GLOAS_SLOT}); + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(true); + vi.mocked(db.block.get).mockResolvedValue(block); + + const result = await cache.getOrReload(rootHex); + + expect(result).toBeDefined(); + expect(result?.slot).toBe(GLOAS_SLOT); + // inserted into the cache so subsequent sync get() hits + expect(cache.get(rootHex)).toBe(result); + }); + + it("reconstructs from seenBlockInputCache without reading db when the block is still in memory", async () => { + const {block, rootHex} = generateBlock({forkName: ForkName.gloas, slot: GLOAS_SLOT}); + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(true); + // Block imported but its async db write has not flushed yet: it is still in seenBlockInputCache. + vi.mocked(seenBlockInputCache.get).mockReturnValue({ + hasBlock: () => true, + getBlock: () => block, + } as unknown as ReturnType); + + const result = await cache.getOrReload(rootHex); + + expect(result).toBeDefined(); + expect(result?.slot).toBe(GLOAS_SLOT); + expect(cache.get(rootHex)).toBe(result); + // in-memory hit avoids the disk read + expect(db.block.get).not.toHaveBeenCalled(); + }); + + it("returns undefined during the import-to-persist window (block not yet in hot db)", async () => { + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(true); + vi.mocked(db.block.get).mockResolvedValue(null); + + const result = await cache.getOrReload(`0x${"cd".repeat(32)}`); + + expect(result).toBeUndefined(); + expect(cache.size()).toBe(0); + }); + + it("returns undefined for a pre-gloas block", async () => { + const {block, rootHex} = generateBlock({forkName: ForkName.fulu, slot: FULU_SLOT}); + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(true); + vi.mocked(db.block.get).mockResolvedValue(block); + + const result = await cache.getOrReload(rootHex); + + expect(result).toBeUndefined(); + expect(cache.get(rootHex)).toBeUndefined(); + }); + + it("dedups concurrent reloads of the same root into a single db read and object", async () => { + const {block, rootHex} = generateBlock({forkName: ForkName.gloas, slot: GLOAS_SLOT}); + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(true); + vi.mocked(db.block.get).mockResolvedValue(block); + + const [a, b] = await Promise.all([cache.getOrReload(rootHex), cache.getOrReload(rootHex)]); + + // Without in-flight dedup, each miss would build a divergent shell that the import-path + // WeakMap cannot dedup on object identity. + expect(a).toBe(b); + expect(db.block.get).toHaveBeenCalledTimes(1); + }); + }); + + describe("pruneToMaxSize (insertion-order cap)", () => { + const MAX = (MAX_LOOK_AHEAD_EPOCHS + 1) * SLOTS_PER_EPOCH; + + it("caps size and evicts by insertion order, not slot order", () => { + // Insert MAX + 2 entries in DESCENDING slot order, so insertion order != slot order. + const rootsInInsertionOrder: string[] = []; + for (let i = MAX + 1; i >= 0; i--) { + rootsInInsertionOrder.push(addPayloadInput(GLOAS_SLOT + i)); + } + + expect(cache.size()).toBe(MAX); + // The 2 first-inserted (highest slots) are evicted; a slot-ordered cap would have evicted the + // lowest slots instead. + expect(cache.get(rootsInInsertionOrder[0])).toBeUndefined(); + expect(cache.get(rootsInInsertionOrder[1])).toBeUndefined(); + // Everything inserted afterwards (including the lowest slots) survives. + expect(cache.get(rootsInInsertionOrder[2])).toBeDefined(); + expect(cache.get(rootsInInsertionOrder.at(-1) as string)).toBeDefined(); + }); + + it("keeps a just-reloaded old-slot entry, evicting an older-inserted one instead", async () => { + // Fill exactly to MAX with high slots. + const firstInserted = addPayloadInput(GLOAS_SLOT + MAX); + for (let i = MAX - 1; i >= 1; i--) { + addPayloadInput(GLOAS_SLOT + i); + } + expect(cache.size()).toBe(MAX); + + // getOrReload an old-slot miss: it inserts the reconstructed shell at the BACK, then the cap runs. + const {block, rootHex} = generateBlock({forkName: ForkName.gloas, slot: GLOAS_SLOT}); + vi.mocked(forkChoice.hasBlockHex).mockReturnValue(true); + vi.mocked(db.block.get).mockResolvedValue(block); + const reloaded = await cache.getOrReload(rootHex); + + expect(reloaded).toBeDefined(); + expect(cache.size()).toBe(MAX); + // Anti-thrash: the reloaded old-slot entry survives; the oldest-INSERTED entry was evicted instead. + expect(cache.get(rootHex)).toBe(reloaded); + expect(cache.get(firstInserted)).toBeUndefined(); + }); + + it("increments pruned{reason:cap} on a cap eviction", () => { + const prunedInc = vi.fn(); + const metrics = { + seenCache: { + payloadEnvelopeInput: { + count: {addCollect: vi.fn(), set: vi.fn()}, + serializedObjectRefs: {set: vi.fn()}, + created: {inc: vi.fn()}, + pruned: {inc: prunedInc}, + }, + }, + }; + const cacheWithMetrics = new SeenPayloadEnvelopeInput({ + config, + clock: getMockedClock(), + forkChoice, + chainEvents, + signal: abortController.signal, + serializedCache: new SerializedCache(), + db, + seenBlockInputCache, + custodyConfig: {sampledColumns: [], custodyColumns: []} as unknown as CustodyConfig, + metrics: metrics as unknown as ConstructorParameters[0]["metrics"], + logger: testLogger(), + }); + + for (let i = MAX; i >= 0; i--) { + const {block, rootHex} = generateBlock({forkName: ForkName.gloas, slot: GLOAS_SLOT + i}); + cacheWithMetrics.add({ + blockRootHex: rootHex, + block, + forkName: ForkName.gloas, + sampledColumns: [], + custodyColumns: [], + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.gossip, + }); + } + + expect(cacheWithMetrics.size()).toBe(MAX); + expect(prunedInc).toHaveBeenCalledWith({reason: "cap"}); + }); + }); }); diff --git a/packages/beacon-node/test/unit/sync/range/batch.test.ts b/packages/beacon-node/test/unit/sync/range/batch.test.ts index 66ee427ae0d2..be94021f9758 100644 --- a/packages/beacon-node/test/unit/sync/range/batch.test.ts +++ b/packages/beacon-node/test/unit/sync/range/batch.test.ts @@ -340,7 +340,8 @@ describe("sync / range / batch", async () => { forkName: ForkName.gloas, sampledColumns, custodyColumns: sampledColumns, - timeCreatedSec: seenTimestampSec, + seenTimestampSec, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); if (addEnvelope) { diff --git a/packages/beacon-node/test/unit/sync/unknownBlock.test.ts b/packages/beacon-node/test/unit/sync/unknownBlock.test.ts index 05ca2e503e84..2ce1a5270467 100644 --- a/packages/beacon-node/test/unit/sync/unknownBlock.test.ts +++ b/packages/beacon-node/test/unit/sync/unknownBlock.test.ts @@ -85,7 +85,8 @@ function buildPayloadFixture({ forkName: ForkName.gloas, sampledColumns, custodyColumns: sampledColumns, - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); @@ -488,6 +489,7 @@ describe("sync by UnknownBlockSync", {timeout: 20_000}, () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockReturnValue(undefined), + getOrReload: vi.fn().mockResolvedValue(undefined), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], }; @@ -735,6 +737,7 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockReturnValue(undefined), + getOrReload: vi.fn().mockResolvedValue(undefined), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], } as unknown as IBeaconChain; @@ -821,6 +824,7 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockReturnValue(undefined), + getOrReload: vi.fn().mockResolvedValue(undefined), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], seenBlockInputCache: {prune: vi.fn()} as unknown as SeenBlockInput, @@ -892,6 +896,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -956,6 +963,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -1027,6 +1037,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -1096,6 +1109,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? cachedPayloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? cachedPayloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], seenBlockInputCache: { @@ -1200,6 +1216,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], seenBlockInputCache: { @@ -1262,9 +1281,16 @@ describe("UnknownBlockSync", () => { expect(sendExecutionPayloadEnvelopesByRoot).toHaveBeenCalledWith(peer, [blockRoot]); expect(sendBeaconBlocksByRoot).toHaveBeenCalledWith(peer, [blockRoot]); expect(processBlock).toHaveBeenCalledTimes(1); + // Validation runs exactly once: the first reconcile pass seeds the envelope into payloadInput, so any + // racing scheduler pass sees hasPayloadEnvelope() and skips revalidation. expect(validateGossipExecutionPayloadEnvelope).toHaveBeenCalledOnce(); - expect(processExecutionPayload).toHaveBeenCalledTimes(1); - expect(processExecutionPayload).toHaveBeenCalledWith(payloadInput); + // Two racing scheduler passes may both reach processPayload, but every call carries the SAME payloadInput + // object; chain.processExecutionPayload (via processPayloadEnvelopeJob's WeakMap) dedups them into one + // real import. This mock bypasses that dedup, so assert on object identity rather than an exact count. + expect(processExecutionPayload).toHaveBeenCalled(); + for (const [arg] of processExecutionPayload.mock.calls) { + expect(arg).toBe(payloadInput); + } }); it("downloads the block and retries payload import when EL reports block not in fork choice", async () => { @@ -1298,6 +1324,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], seenBlockInputCache: { @@ -1396,6 +1425,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -1454,6 +1486,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === blockRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], }, @@ -1492,6 +1527,7 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockReturnValue(payloadInput), + getOrReload: vi.fn().mockResolvedValue(payloadInput), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -1561,6 +1597,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === parentRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === parentRootHex ? payloadInput : undefined)), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -1849,7 +1888,8 @@ describe("UnknownBlockSync", () => { forkName: ForkName.gloas, sampledColumns: [], custodyColumns: [], - timeCreatedSec: Date.now() / 1000, + seenTimestampSec: Date.now() / 1000, + source: PayloadEnvelopeInputSource.byRange, daOutOfRange: false, }); @@ -1882,7 +1922,9 @@ describe("UnknownBlockSync", () => { } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], seenBlockInputCache: {prune: seenBlockInputPrune} as unknown as SeenBlockInput, forkChoice: { - hasPayloadHexUnsafe: vi.fn().mockReturnValue(false), + // Parent payload is revealed/imported (FULL) but no variant matches the child's parentBlockHash. + // getMissingBlockDependency defers the drop until the payload lands; only then is the conflict real. + hasPayloadHexUnsafe: vi.fn().mockImplementation((root: string) => root === parentRootHex), hasBlockHex: vi.fn().mockImplementation((root: string) => root === parentRootHex), getBlockHexDefaultStatus: vi .fn() @@ -1952,6 +1994,9 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn().mockImplementation((root: string) => (root === parentRootHex ? payloadInput : undefined)), + getOrReload: vi + .fn() + .mockImplementation((root: string) => (root === parentRootHex ? payloadInput : undefined)), prune: seenPayloadPrune, } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], forkChoice: { @@ -2055,6 +2100,7 @@ describe("UnknownBlockSync", () => { seenPayloadEnvelopeInputCache: { add: vi.fn(), get: vi.fn(), + getOrReload: vi.fn().mockResolvedValue(undefined), prune: vi.fn(), } as unknown as IBeaconChain["seenPayloadEnvelopeInputCache"], seenBlockInputCache: {prune: vi.fn()} as unknown as SeenBlockInput,