From f113a25c9e2c6e475a62d7afe8bb2412f05365c8 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 27 Aug 2026 17:59:49 -0300 Subject: [PATCH] feat(fast-inbox): wait for the local archiver before rejecting a block proposal for an unsynced inbox bucket A proposer only consumes buckets at least one Ethereum slot old, so a bucket a validator cannot resolve is almost always its own archiver trailing L1, not a divergence. Rejecting on the spot with `bucket_unknown` lost an attestation for a pure race. The handler now forces an archiver sync and re-runs the whole metadata check every half second until it resolves or the attestation deadline passes. A hash mismatch on a known bucket gets one forced sync and one re-check (this node may be the stale side of an L1 reorg); every other reason still rejects immediately. The deadline-bounded sync waits in the handler now go through a shared `awaitLocalSync` helper, which the new wait reuses; the checkpoint last-block wait keeps its own copy because it must still attempt one lookup after the deadline has passed. --- .../src/proposal_handler.test.ts | 138 ++++++++++++- .../validator-client/src/proposal_handler.ts | 181 +++++++++++++----- .../src/streaming_inbox_checks.ts | 11 +- 3 files changed, 279 insertions(+), 51 deletions(-) diff --git a/yarn-project/validator-client/src/proposal_handler.test.ts b/yarn-project/validator-client/src/proposal_handler.test.ts index 265e1eceb053..6722d0f484ba 100644 --- a/yarn-project/validator-client/src/proposal_handler.test.ts +++ b/yarn-project/validator-client/src/proposal_handler.test.ts @@ -970,7 +970,7 @@ describe('ProposalHandler checkpoint validation', () => { }); /** Genesis-parent streaming block proposal at slot 1, with the handler wired to reach the streaming checks. */ - async function setupStreamingProposal(bucketRef: InboxBucketRef | undefined) { + async function setupStreamingProposal(bucketRef: InboxBucketRef | undefined, options: { nowMs?: number } = {}) { const proposal = ValidatedBlockProposal( await makeBlockProposal({ blockHeader: makeBlockHeader(1, { slotNumber: SlotNumber(1) }), @@ -987,8 +987,8 @@ describe('ProposalHandler checkpoint validation', () => { const txProvider = mock(); txProvider.getTxsForBlockProposal.mockResolvedValue({ txs: [], missingTxs: [] } as any); - // Well past the minimum bucket age (one 12s Ethereum slot) for a bucket opened at t=100. - dateProvider.setTime(1_000_000); + // Well past the minimum bucket age (one 12s Ethereum slot) for a bucket opened at t=100, unless overridden. + dateProvider.setTime(options.nowMs ?? 1_000_000); const blockHandler = new ProposalHandler( checkpointsBuilder, @@ -1086,6 +1086,138 @@ describe('ProposalHandler checkpoint validation', () => { expect.anything(), ); }); + + // A bucket the proposer already consumed is on L1 by construction, so a bucket this node cannot resolve is + // (usually) local archiver lag, not a divergence: the handler forces a sync and re-checks until the + // attestation deadline instead of dropping the attestation on the spot. + describe('bucket sync wait', () => { + // attestation_deadline(slot=1) = 1*24 + 24 - 8 = 40s. Waits run on a real timer against the remaining + // budget read off the fake clock, so holding it 2s short of the deadline keeps the tests short. + const DEADLINE_MS = 40_000; + const WAIT_BUDGET_MS = 2_000; + const BEFORE_DEADLINE_MS = DEADLINE_MS - WAIT_BUDGET_MS; + const PAST_DEADLINE_MS = DEADLINE_MS + 1_000; + const WAIT_INTERVAL_MS = 500; + + /** A bucket old enough to be lag-eligible at {@link BEFORE_DEADLINE_MS}. */ + const eligibleBucket = (overrides: Partial = {}) => bucket({ timestamp: 10n, ...overrides }); + + /** Wires the parent-bucket lookup and the bundle read for a proposal that consumes `eligibleBucket()`. */ + function mockAcceptedSurroundings() { + l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockResolvedValue( + eligibleBucket({ seq: 0n, totalMsgCount: 0n, msgCount: 0 }), + ); + l1ToL2MessageSource.getL1ToL2MessagesBetweenBuckets.mockResolvedValue([new Fr(1000), new Fr(1001)]); + } + + it('attests once the referenced bucket shows up on a later archiver sync', async () => { + const ref = new InboxBucketRef(1n, 10n, new Fr(0xabc)); + const { proposal, blockHandler } = await setupStreamingProposal(ref, { nowMs: BEFORE_DEADLINE_MS }); + mockAcceptedSurroundings(); + // Unknown on arrival, synced by the time the wait re-checks. + l1ToL2MessageSource.getInboxBucket.mockResolvedValueOnce(undefined).mockResolvedValue(eligibleBucket()); + jest.spyOn(blockHandler, 'reexecuteTransactions').mockResolvedValue({ block: undefined } as any); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result.isValid).toBe(true); + expect(result.blockNumber).toEqual(BlockNumber(INITIAL_L2_BLOCK_NUM)); + }); + + it('rejects with bucket_unknown when the bucket never syncs, no earlier than the deadline', async () => { + const ref = new InboxBucketRef(1n, 10n, new Fr(0xabc)); + const { proposal, blockHandler, txProvider } = await setupStreamingProposal(ref, { + nowMs: BEFORE_DEADLINE_MS, + }); + mockAcceptedSurroundings(); + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(undefined); + + const startMs = Date.now(); + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + const elapsedMs = Date.now() - startMs; + + expect(result).toEqual({ + isValid: false, + blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM), + reason: 'bucket_unknown', + }); + // The wait runs out the remaining budget and gives up within one retry interval of the deadline. + expect(elapsedMs).toBeGreaterThanOrEqual(WAIT_BUDGET_MS - 100); + expect(elapsedMs).toBeLessThan(WAIT_BUDGET_MS + 2 * WAIT_INTERVAL_MS); + // Waiting never buys the proposer any network work: the rejection still happens before tx collection. + expect(txProvider.getTxsForBlockProposal).not.toHaveBeenCalled(); + }); + + it('rejects immediately without syncing when the attestation deadline has already passed', async () => { + const ref = new InboxBucketRef(1n, 10n, new Fr(0xabc)); + const { proposal, blockHandler } = await setupStreamingProposal(ref, { nowMs: PAST_DEADLINE_MS }); + mockAcceptedSurroundings(); + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(undefined); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result).toEqual({ + isValid: false, + blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM), + reason: 'bucket_unknown', + }); + // With no budget left there is nothing to wait for, so the archiver is not poked at all. + expect(blockSource.syncImmediate).not.toHaveBeenCalled(); + }); + + it('rejects immediately without syncing when the proposal carries no bucket reference', async () => { + const { proposal, blockHandler } = await setupStreamingProposal(undefined, { + nowMs: BEFORE_DEADLINE_MS, + }); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result).toEqual({ + isValid: false, + blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM), + reason: 'bucket_unknown', + }); + expect(blockSource.syncImmediate).not.toHaveBeenCalled(); + }); + + it('rejects a hash mismatch that survives one forced sync, without looping', async () => { + const ref = new InboxBucketRef(1n, 10n, new Fr(0xdead)); + const { proposal, blockHandler } = await setupStreamingProposal(ref, { nowMs: BEFORE_DEADLINE_MS }); + mockAcceptedSurroundings(); + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(eligibleBucket({ inboxRollingHash: new Fr(0xabc) })); + + const startMs = Date.now(); + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + const elapsedMs = Date.now() - startMs; + + expect(result).toEqual({ + isValid: false, + blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM), + reason: 'bucket_hash_mismatch', + }); + // A persistent mismatch is a divergence from L1, not local lag: one sync, one re-check, no retry loop. + expect(blockSource.syncImmediate).toHaveBeenCalledTimes(1); + expect(elapsedMs).toBeLessThan(WAIT_INTERVAL_MS); + }); + + it('attests when the forced sync replaces our stale bucket with the proposed one', async () => { + // This validator held the orphaned side of an L1 reorg; the forced sync rolls it back and re-syncs. + const ref = new InboxBucketRef(1n, 10n, new Fr(0xabc)); + const { proposal, blockHandler } = await setupStreamingProposal(ref, { nowMs: BEFORE_DEADLINE_MS }); + mockAcceptedSurroundings(); + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(eligibleBucket({ inboxRollingHash: new Fr(0xbad) })); + blockSource.syncImmediate.mockImplementation(() => { + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(eligibleBucket()); + return Promise.resolve(); + }); + jest.spyOn(blockHandler, 'reexecuteTransactions').mockResolvedValue({ block: undefined } as any); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result.isValid).toBe(true); + expect(result.blockNumber).toEqual(BlockNumber(INITIAL_L2_BLOCK_NUM)); + }); + }); }); // Streaming Inbox: the checkpoint handler enforces the last-block minimum-consumption (censorship) rule before diff --git a/yarn-project/validator-client/src/proposal_handler.ts b/yarn-project/validator-client/src/proposal_handler.ts index ec348375861b..5b6073292f2e 100644 --- a/yarn-project/validator-client/src/proposal_handler.ts +++ b/yarn-project/validator-client/src/proposal_handler.ts @@ -577,7 +577,7 @@ export class ProposalHandler { // Streaming Inbox: run the metadata checks before committing to any network work. They are point lookups // against our own Inbox view, so a proposal carrying a bucket reference that does not resolve locally is // rejected without a proposer being able to make us spend the validation window collecting its txs. - const streamingMetadata = await this.checkStreamingBlockMetadata(proposal, blockNumber, parentBlock); + const streamingMetadata = await this.awaitStreamingBlockMetadata(proposal, blockNumber, parentBlock, proposalInfo); if (!streamingMetadata.accepted) { this.log.warn(`Streaming Inbox block acceptance check failed, skipping processing`, { reason: streamingMetadata.reason, @@ -729,6 +729,40 @@ export class ProposalHandler { } } + /** + * Re-runs `resolve` against this node's local view, forcing an archiver L1 sync before every attempt, until it + * yields a value or the slot's attestation deadline passes. Returns `undefined` when the deadline had already + * passed on entry (nothing is forced in that case) or when it passes while waiting; anything other than the + * timeout propagates. Callers own their own logging and whatever they fall back to on `undefined`, and check + * the deadline themselves when they need to tell "no budget on entry" apart from "timed out while waiting". + */ + private async awaitLocalSync( + slotNumber: SlotNumber, + what: string, + resolve: () => Promise, + ): Promise { + const deadline = this.getReexecutionDeadline(slotNumber); + if (deadline.getTime() - this.dateProvider.now() <= 0) { + return undefined; + } + try { + return await retryUntil( + async () => { + await this.blockSource.syncImmediate(); + return await resolve(); + }, + what, + { deadline, dateProvider: this.dateProvider }, + 0.5, + ); + } catch (err) { + if (err instanceof TimeoutError) { + return undefined; + } + throw err; + } + } + private async getParentBlock(proposal: BlockProposal): Promise<'genesis' | BlockData | undefined> { const parentArchive = proposal.blockHeader.lastArchive.root; const { genesisArchiveRoot } = await this.blockSource.getGenesisValues(); @@ -737,28 +771,23 @@ export class ProposalHandler { return 'genesis'; } - const deadline = this.getReexecutionDeadline(proposal.slotNumber); - const timeoutDurationMs = deadline.getTime() - this.dateProvider.now(); - try { - return ( - (await this.blockSource.getBlockData({ archive: parentArchive })) ?? - (timeoutDurationMs <= 0 - ? undefined - : await retryUntil( - () => - this.blockSource.syncImmediate().then(() => this.blockSource.getBlockData({ archive: parentArchive })), - 'force archiver sync', - { deadline, dateProvider: this.dateProvider }, - 0.5, - )) + const parentBlock = await this.blockSource.getBlockData({ archive: parentArchive }); + if (parentBlock !== undefined) { + return parentBlock; + } + if (this.getReexecutionDeadline(proposal.slotNumber).getTime() - this.dateProvider.now() <= 0) { + return undefined; + } + const synced = await this.awaitLocalSync(proposal.slotNumber, 'force archiver sync', () => + this.blockSource.getBlockData({ archive: parentArchive }), ); - } catch (err) { - if (err instanceof TimeoutError) { + if (synced === undefined) { this.log.debug(`Timed out getting parent block by archive root`, { parentArchive }); - } else { - this.log.error('Error getting parent block by archive root', err, { parentArchive }); } + return synced; + } catch (err) { + this.log.error('Error getting parent block by archive root', err, { parentArchive }); return undefined; } } @@ -784,8 +813,7 @@ export class ProposalHandler { // A different block already occupies this number: it may be a stale fork being pruned during a reorg, not a // genuine duplicate. Wait for the local prune rather than permanently rejecting the proposal. - const deadline = this.getReexecutionDeadline(slotNumber); - if (deadline.getTime() - this.dateProvider.now() <= 0) { + if (this.getReexecutionDeadline(slotNumber).getTime() - this.dateProvider.now() <= 0) { return existingBlock; } @@ -795,29 +823,19 @@ export class ProposalHandler { proposalArchive: proposalArchive.toString(), }); - try { - const { block } = await retryUntil( - async () => { - await this.blockSource.syncImmediate(); - const block = await this.blockSource.getBlockData({ number: blockNumber }); - // Resolve once the existing block is gone (pruned) or has been replaced by one matching the - // proposal — the same condition as the early return above. A matching block is returned so the - // caller still treats it as a genuine duplicate; an `undefined` (pruned) block lets the proposal - // be processed. Wrap in an object so the `undefined` case is still a truthy retry result. - return block === undefined || block.archive.root.equals(proposalArchive) ? { block } : undefined; - }, - `prune of stale block ${blockNumber}`, - { deadline, dateProvider: this.dateProvider }, - 0.5, - ); - return block; - } catch (err) { - if (err instanceof TimeoutError) { - this.log.warn(`Timed out waiting for stale block ${blockNumber} to be pruned`, { blockNumber }); - return existingBlock; - } - throw err; + const pruned = await this.awaitLocalSync(slotNumber, `prune of stale block ${blockNumber}`, async () => { + const block = await this.blockSource.getBlockData({ number: blockNumber }); + // Resolve once the existing block is gone (pruned) or has been replaced by one matching the + // proposal — the same condition as the early return above. A matching block is returned so the + // caller still treats it as a genuine duplicate; an `undefined` (pruned) block lets the proposal + // be processed. Wrap in an object so the `undefined` case is still a truthy retry result. + return block === undefined || block.archive.root.equals(proposalArchive) ? { block } : undefined; + }); + if (pruned === undefined) { + this.log.warn(`Timed out waiting for stale block ${blockNumber} to be pruned`, { blockNumber }); + return existingBlock; } + return pruned.block; } private computeCheckpointNumber( @@ -969,6 +987,80 @@ export class ProposalHandler { } } + /** + * Runs the streaming-Inbox metadata checks, waiting out a local sync lag. A bucket the proposer consumed is at + * least one Ethereum slot old, so it is on L1 by the time the proposal arrives: a bucket this node cannot + * resolve is almost always its own archiver trailing L1, not a divergence. That case (and the equivalent one + * where the block before the checkpoint's first block has not synced) forces an archiver sync and re-checks + * every half second until it resolves or the attestation deadline passes, instead of dropping the attestation + * on the spot. A hash mismatch on a bucket we do know gets exactly one forced sync and one re-check, because + * this node may be the stale side of an L1 reorg and that sync performs the rollback; a mismatch that survives + * it will not resolve by waiting. Every other reason is a structural rejection and returns immediately. + * + * The wait is bounded by the same consensus deadline as the other sync waits here, so a proposer referencing a + * bucket that never appears can at most make validators poll their own archiver for the remainder of its own + * slot — which it could waste anyway by not proposing. + */ + private async awaitStreamingBlockMetadata( + proposal: BlockProposal, + blockNumber: BlockNumber, + parentBlock: 'genesis' | BlockData, + proposalInfo: LogData, + ): Promise { + const first = await this.checkStreamingBlockMetadata(proposal, blockNumber, parentBlock); + const bucketRef = proposal.bucketRef; + if (first.accepted || bucketRef === undefined) { + return first; + } + + const slotNumber = proposal.slotNumber; + const bucketSeq = bucketRef.bucketSeq; + const outOfBudget = this.getReexecutionDeadline(slotNumber).getTime() - this.dateProvider.now() <= 0; + + if (first.reason === 'bucket_hash_mismatch') { + if (outOfBudget) { + return first; + } + await this.blockSource.syncImmediate(); + const rechecked = await this.checkStreamingBlockMetadata(proposal, blockNumber, parentBlock); + if (!rechecked.accepted && rechecked.reason === 'bucket_hash_mismatch') { + this.log.warn(`Inbox bucket ${bucketSeq} still disagrees with the proposal after forcing an archiver sync`, { + reason: 'bucket_hash_mismatch_after_sync', + bucketSeq, + expected: bucketRef.inboxRollingHash.toString(), + actual: (await this.l1ToL2MessageSource.getInboxBucket(bucketSeq))?.inboxRollingHash.toString(), + ...proposalInfo, + }); + } + return rechecked; + } + + if (first.reason !== 'bucket_unknown') { + return first; + } + + this.log.info(`Referenced Inbox bucket ${bucketSeq} not synced locally, awaiting archiver sync`, { + bucketSeq, + ...proposalInfo, + }); + const timer = new Timer(); + const resolved = await this.awaitLocalSync(slotNumber, `inbox bucket ${bucketSeq}`, async () => { + const result = await this.checkStreamingBlockMetadata(proposal, blockNumber, parentBlock); + return !result.accepted && result.reason === 'bucket_unknown' ? undefined : result; + }); + if (resolved === undefined) { + this.log.warn(`Timed out waiting for Inbox bucket ${bucketSeq} to sync, rejecting proposal`, { + reason: 'bucket_sync_timeout', + slot: slotNumber, + bucketSeq, + waitedMs: timer.ms(), + ...proposalInfo, + }); + return first; + } + return resolved; + } + /** * Runs the streaming-Inbox per-block metadata checks for a block proposal, returning the bucket range its message * bundle derives from or a rejection reason. The parent block's consumed total and the checkpoint's starting total @@ -988,7 +1080,8 @@ export class ProposalHandler { ); if (checkpointStartTotalMsgCount === undefined) { // The block before the checkpoint's first block has not synced locally, so the per-checkpoint cap origin is - // unavailable: treat as an unknown local view. There is no bounded wait for the missing block yet. + // unavailable: treat as an unknown local view. Like an unknown bucket this is local lag rather than a + // divergence, and `awaitStreamingBlockMetadata` waits it out by re-running the whole check after a sync. return { accepted: false, reason: 'bucket_unknown' }; } const nowSeconds = BigInt(Math.floor(this.dateProvider.now() / 1000)); diff --git a/yarn-project/validator-client/src/streaming_inbox_checks.ts b/yarn-project/validator-client/src/streaming_inbox_checks.ts index 2d2f1effa0dd..861c32530b2a 100644 --- a/yarn-project/validator-client/src/streaming_inbox_checks.ts +++ b/yarn-project/validator-client/src/streaming_inbox_checks.ts @@ -92,9 +92,11 @@ export type StreamingBlockCheckResult = * Mirrors the L1 acceptance conditions: * * 1. **Exists**: the referenced bucket resolves in this node's own Inbox view, and its consensus rolling hash matches - * the reference. An unknown bucket is an immediate reject here (there is no bounded wait yet); a hash - * mismatch means the wire reference disagrees with the local bucket. The reference is trusted only as a `bucketSeq` - * lookup hint — timestamp and message counts are read from the locally resolved bucket, never from the wire. + * the reference. An unknown bucket is an immediate reject here; the caller decides whether it is worth waiting for + * a local sync and re-running the checks (the validator's proposal handler does, bounded by the attestation + * deadline). A hash mismatch means the wire reference disagrees with the local bucket. The reference is trusted + * only as a `bucketSeq` lookup hint — timestamp and message counts are read from the locally resolved bucket, + * never from the wire. * 2. **Moves forward**: the bucket's cumulative total is at least the parent block's, so consumption never rewinds. * Equal totals mean the block consumes nothing (empty bundle). * 3. **Not too new**: the bucket is at least `minBucketAgeSeconds` old at validation time @@ -107,7 +109,8 @@ export type StreamingBlockCheckResult = * Because this phase is cheap and needs nothing off the network, a caller can run it before committing to any * expensive work on a proposal — notably before collecting the proposal's transactions over P2P. * - * The reject branch is a single function so a future bounded wait can wrap `bucket_unknown`. + * The reject branch is a single function so a caller can re-run the whole check after forcing a local sync, which + * is how the validator's proposal handler turns a `bucket_unknown` into a bounded wait. */ export async function checkStreamingBlockProposalMetadata( input: StreamingBlockMetadataCheckInput,