Skip to content

feat(fast-inbox): wait for the local archiver before rejecting a block proposal for an unsynced inbox bucket #25342

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
spalladino wants to merge 1 commit into from
Open

feat(fast-inbox): wait for the local archiver before rejecting a block proposal for an unsynced inbox bucket #25342

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 135 additions & 3 deletions yarn-project/validator-client/src/proposal_handler.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) }),
Expand All @@ -987,8 +987,8 @@ describe('ProposalHandler checkpoint validation', () => {
const txProvider = mock<ITxProvider>();
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,
Expand Down Expand Up @@ -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<InboxBucket> = {}) => 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
Expand Down
181 changes: 137 additions & 44 deletions yarn-project/validator-client/src/proposal_handler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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<T>(
slotNumber: SlotNumber,
what: string,
resolve: () => Promise<T | undefined>,
): Promise<T | undefined> {
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();
Expand All @@ -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;
}
}
Expand All @@ -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;
}

Expand All @@ -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(
Expand Down Expand Up @@ -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<StreamingBlockMetadataCheckResult> {
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
Expand All @@ -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));
Expand Down
Loading
Loading