diff --git a/noir-projects/noir-protocol-circuits/crates/types/src/constants.nr b/noir-projects/noir-protocol-circuits/crates/types/src/constants.nr index 0595238903a3..4f5c31a9823a 100644 --- a/noir-projects/noir-protocol-circuits/crates/types/src/constants.nr +++ b/noir-projects/noir-protocol-circuits/crates/types/src/constants.nr @@ -70,7 +70,7 @@ pub global MAX_L1_TO_L2_MSGS_PER_BLOCK: u32 = 1024; // Cap on L1-to-L2 messages consumed by a single checkpoint. Semantically today's NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP. pub global MAX_L1_TO_L2_MSGS_PER_CHECKPOINT: u32 = 1024; // Minimum bucket age, in seconds, at the start of a checkpoint's build frame for its consumption to be mandatory under -// the streaming Inbox censorship cutoff (AZIP-22 Fast Inbox). One L1 slot: validators cannot be required to act on +// the streaming Inbox censorship cutoff. One L1 slot: validators cannot be required to act on // buckets they may not yet have seen. L2 consensus policy consumed by the sequencer and validator (and mirrored by // ProposeLib on L1); the protocol circuits do not read it. pub global INBOX_LAG_SECONDS: u32 = 12; diff --git a/yarn-project/archiver/src/errors.ts b/yarn-project/archiver/src/errors.ts index 565a2716e9ba..29f4de4e84b6 100644 --- a/yarn-project/archiver/src/errors.ts +++ b/yarn-project/archiver/src/errors.ts @@ -127,6 +127,17 @@ export class InboxBucketNotSyncedError extends Error { } } +/** + * Thrown when a cumulative Inbox message count does not resolve to a bucket boundary this archiver has synced, either + * because the count sits inside a bucket or because the bucket is not synced yet. + */ +export class InboxBucketBoundaryNotSyncedError extends Error { + constructor(public readonly totalMsgCount: bigint) { + super(`No synced Inbox bucket ends at cumulative message count ${totalMsgCount}`); + this.name = 'InboxBucketBoundaryNotSyncedError'; + } +} + /** Thrown when a proposed checkpoint number is stale (already processed). */ export class ProposedCheckpointStaleError extends Error { constructor( diff --git a/yarn-project/archiver/src/modules/data_source_base.ts b/yarn-project/archiver/src/modules/data_source_base.ts index 10da66d12bc4..9d2256c7baa6 100644 --- a/yarn-project/archiver/src/modules/data_source_base.ts +++ b/yarn-project/archiver/src/modules/data_source_base.ts @@ -332,10 +332,18 @@ export abstract class ArchiverDataSourceBase return this.stores.messages.getInboxBucket(seq); } + public getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise { + return this.stores.messages.getInboxBucketByTotalMsgCount(totalMsgCount); + } + public getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { return this.stores.messages.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive); } + public getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise { + return this.stores.messages.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount); + } + private async getPublishedCheckpointFromCheckpointData(checkpoint: CheckpointData): Promise { const blocksForCheckpoint = await this.stores.blocks.getBlocksForCheckpoint(checkpoint.checkpointNumber); if (!blocksForCheckpoint) { diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index 983624f51b8e..c7191ba61199 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -8,7 +8,11 @@ import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; import { updateInboxRollingHash } from '@aztec/stdlib/messaging'; import '@aztec/stdlib/testing/jest'; -import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js'; +import { + InboxBucketBoundaryNotSyncedError, + InboxBucketNotSyncedError, + L1ToL2MessagesNotReadyError, +} from '../errors.js'; import { type InboxMessage, updateRollingHash } from '../structs/inbox_message.js'; import { makeInboxMessage, @@ -401,6 +405,28 @@ describe('MessageStore', () => { expect(await messageStore.getInboxBucket(4n)).toBeUndefined(); }); + it('resolves a bucket by its cumulative message total', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + + // Each bucket boundary (cumulative totals 3, 5, 6) resolves to its bucket. + expect((await messageStore.getInboxBucketByTotalMsgCount(3n))?.seq).toEqual(1n); + expect((await messageStore.getInboxBucketByTotalMsgCount(5n))?.seq).toEqual(2n); + expect((await messageStore.getInboxBucketByTotalMsgCount(6n))?.seq).toEqual(3n); + // A total inside a bucket (not on a boundary) does not resolve. + expect(await messageStore.getInboxBucketByTotalMsgCount(4n)).toBeUndefined(); + // A total past the last synced bucket does not resolve. + expect(await messageStore.getInboxBucketByTotalMsgCount(7n)).toBeUndefined(); + }); + + it('synthesizes the genesis sentinel bucket (sequence 0, total 0) which is never ingested', async () => { + // With real messages present but no ingested sequence-0 snapshot, both lookups still resolve genesis. + await messageStore.addL1ToL2MessageBuckets(makeBucketedMessages(threeBucketSpec)); + + expect(await messageStore.getInboxBucket(0n)).toMatchObject({ seq: 0n, totalMsgCount: 0n, msgCount: 0 }); + expect(await messageStore.getInboxBucketByTotalMsgCount(0n)).toMatchObject({ seq: 0n, totalMsgCount: 0n }); + }); + it('rejects a bucket delivered without the messages already stored for it', async () => { const msgs = makeBucketedMessages(threeBucketSpec); await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 2)); @@ -492,6 +518,40 @@ describe('MessageStore', () => { await expect(messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).rejects.toThrow(/Invalid Inbox bucket range/); }); + it('returns messages between cumulative leaf counts', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + const leaves = msgs.map(m => m.leaf); + + // Bucket boundaries sit at cumulative counts 0 (genesis), 3, 5 and 6. + expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 6n)).toEqual(leaves); + expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 3n)).toEqual(leaves.slice(0, 3)); + expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(3n, 5n)).toEqual(leaves.slice(3, 5)); + expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 6n)).toEqual(leaves.slice(5)); + // An empty range consumes nothing, at a bucket boundary or at genesis. + expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 5n)).toEqual([]); + expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 0n)).toEqual([]); + }); + + it('throws when a leaf count does not land on a synced bucket boundary', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + + // Counts inside a bucket and past the last synced bucket both fail rather than returning a partial range. + await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 4n)).rejects.toThrow( + InboxBucketBoundaryNotSyncedError, + ); + await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(4n, 6n)).rejects.toThrow( + InboxBucketBoundaryNotSyncedError, + ); + await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(3n, 9n)).rejects.toThrow( + InboxBucketBoundaryNotSyncedError, + ); + await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 3n)).rejects.toThrow( + /Invalid Inbox leaf count range/, + ); + }); + it('rewinds buckets when messages are removed', async () => { const msgs = makeBucketedMessages(threeBucketSpec); await messageStore.addL1ToL2MessageBuckets(msgs); diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index f23135da5e72..6054453ae608 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -15,7 +15,11 @@ import { } from '@aztec/kv-store'; import { type InboxBucket, InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging'; -import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js'; +import { + InboxBucketBoundaryNotSyncedError, + InboxBucketNotSyncedError, + L1ToL2MessagesNotReadyError, +} from '../errors.js'; import { type InboxMessage, deserializeInboxMessage, @@ -85,6 +89,19 @@ function groupMessagesByBucket(messages: InboxMessage[]): IncomingBucket[] { return buckets; } +// The genesis sentinel bucket: sequence 0 with a zero rolling hash and no messages, mirroring the +// on-chain Inbox's base case. The archiver never ingests a snapshot for it (no message is absorbed into sequence 0), so +// it is synthesized on read. Consumers use its sequence number and zero total; its deploy-time timestamp is not tracked +// here and is unused. +const GENESIS_INBOX_BUCKET: InboxBucket = { + seq: 0n, + inboxRollingHash: Fr.ZERO, + totalMsgCount: 0n, + timestamp: 0n, + msgCount: 0, + lastMessageIndex: 0n, +}; + export class MessageStoreError extends Error { constructor( message: string, @@ -487,17 +504,65 @@ export class MessageStore { } /** - * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced (AZIP-22 Fast - * Inbox). + * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced. Sequence 0 is + * the genesis sentinel: the on-chain Inbox reserves it as the "consumed nothing" base case + * and never absorbs a message into it, so the archiver ingests no snapshot for it; it is synthesized here (rolling + * hash 0, total 0) so streaming consumers can resolve a genesis parent or an empty checkpoint's last-consumed bucket. */ public async getInboxBucket(seq: bigint): Promise { const snapshot = await this.getBucketSnapshotBySeq(seq); - return snapshot && this.toInboxBucket(seq, snapshot); + if (snapshot !== undefined) { + return this.toInboxBucket(seq, snapshot); + } + return seq === 0n ? GENESIS_INBOX_BUCKET : undefined; + } + + /** + * Returns the Inbox bucket whose cumulative message total equals `totalMsgCount`, or undefined if no synced bucket + * sits on that boundary. Sequence 0 (total 0) is the genesis sentinel base case; otherwise the + * message at global index `totalMsgCount - 1` is the last message of the bucket with that cumulative total, so its + * `bucketSeq` resolves the bucket. A total that does not land on a bucket boundary returns undefined. + */ + public async getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise { + if (totalMsgCount === 0n) { + return this.getInboxBucket(0n); + } + const buffer = await this.#l1ToL2Messages.getAsync(this.indexToKey(totalMsgCount - 1n)); + if (buffer === undefined) { + return undefined; + } + const bucket = await this.getInboxBucket(deserializeInboxMessage(buffer).bucketSeq); + return bucket !== undefined && bucket.totalMsgCount === totalMsgCount ? bucket : undefined; + } + + /** + * Returns the message leaves in the cumulative Inbox message-count range `[startLeafCount, endLeafCount)`, in + * insertion order. The bounds are compact L1-to-L2 tree leaf counts, which every block header + * carries, so consumers can ask for the messages a block or checkpoint consumed without resolving buckets + * themselves. Both bounds must land on a bucket boundary this archiver has synced; it throws otherwise, since a + * caller asking for a range always expects the messages in it. + */ + public async getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise { + if (startLeafCount > endLeafCount) { + throw new Error(`Invalid Inbox leaf count range [${startLeafCount}, ${endLeafCount})`); + } + const startBucket = await this.getBucketAtBoundary(startLeafCount); + const endBucket = await this.getBucketAtBoundary(endLeafCount); + return this.getL1ToL2MessagesBetweenBuckets(startBucket.seq, endBucket.seq); + } + + /** Resolves the bucket ending at the given cumulative message count, failing loudly if there is none. */ + private async getBucketAtBoundary(totalMsgCount: bigint): Promise { + const bucket = await this.getInboxBucketByTotalMsgCount(totalMsgCount); + if (bucket === undefined) { + throw new InboxBucketBoundaryNotSyncedError(totalMsgCount); + } + return bucket; } /** * Returns the latest Inbox bucket opened at or before the given L1 timestamp, or undefined if every synced bucket - * was opened strictly after it (AZIP-22 Fast Inbox). + * was opened strictly after it. */ public async getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { // Bucket timestamps are non-decreasing in sequence number, so the bucket we want is the highest sequence indexed @@ -519,8 +584,8 @@ export class MessageStore { } /** - * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion order - * (AZIP-22 Fast Inbox). Both bounds must name buckets this archiver has synced, so that an empty result means the + * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion order. + * Both bounds must name buckets this archiver has synced, so that an empty result means the * range holds no messages rather than hiding an unsynced bound; callers route the * `InboxBucketNotSyncedError` to their own catch-up handling. Sequence 0 is the genesis base case and always * resolves: the range then starts at the first message of the Inbox. diff --git a/yarn-project/archiver/src/structs/inbox_message.ts b/yarn-project/archiver/src/structs/inbox_message.ts index 2e88e69cfe78..137ee14b1044 100644 --- a/yarn-project/archiver/src/structs/inbox_message.ts +++ b/yarn-project/archiver/src/structs/inbox_message.ts @@ -12,9 +12,9 @@ export type InboxMessage = { l1BlockHash: Buffer32; /** Legacy 128-bit keccak rolling hash of all messages inserted up to and including this one. */ rollingHash: Buffer16; - /** Consensus rolling hash (truncated sha256 chain) of all messages up to and including this one (AZIP-22 Fast Inbox). */ + /** Consensus rolling hash (truncated sha256 chain) of all messages up to and including this one. */ inboxRollingHash: Fr; - /** Sequence number of the Inbox bucket this message was absorbed into (AZIP-22 Fast Inbox). */ + /** Sequence number of the Inbox bucket this message was absorbed into. */ bucketSeq: bigint; /** L1 block timestamp at which this message's bucket was opened; the bucket's recency key, in seconds. */ bucketTimestamp: bigint; diff --git a/yarn-project/archiver/src/test/fake_l1_state.ts b/yarn-project/archiver/src/test/fake_l1_state.ts index 28226f12020b..1137038f2a01 100644 --- a/yarn-project/archiver/src/test/fake_l1_state.ts +++ b/yarn-project/archiver/src/test/fake_l1_state.ts @@ -143,7 +143,7 @@ export class FakeL1State { private checkpoints: CheckpointData[] = []; private messages: MessageData[] = []; private messagesRollingHash: Buffer16 = Buffer16.ZERO; - // Consensus rolling-hash and bucket-ring state, mirroring the on-chain Inbox (AZIP-22 Fast Inbox). + // Consensus rolling-hash and bucket-ring state, mirroring the on-chain Inbox. private messagesConsensusRollingHash: Fr = Fr.ZERO; private currentBucketSeq: bigint = 0n; private currentBucketTimestamp: bigint = 0n; diff --git a/yarn-project/archiver/src/test/mock_archiver.ts b/yarn-project/archiver/src/test/mock_archiver.ts index 4449f9eff9be..33bbcb78db72 100644 --- a/yarn-project/archiver/src/test/mock_archiver.ts +++ b/yarn-project/archiver/src/test/mock_archiver.ts @@ -37,9 +37,17 @@ export class MockArchiver extends MockL2BlockSource implements L2BlockSource, L1 return this.messageSource.getInboxBucket(seq); } + getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise { + return this.messageSource.getInboxBucketByTotalMsgCount(totalMsgCount); + } + getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { return this.messageSource.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive); } + + getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise { + return this.messageSource.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount); + } } /** diff --git a/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts b/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts index 75b6ee1cb859..7a8940a12a23 100644 --- a/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts +++ b/yarn-project/archiver/src/test/mock_l1_to_l2_message_source.ts @@ -38,6 +38,13 @@ export class MockL1ToL2MessageSource implements L1ToL2MessageSource { return Promise.resolve(this.buckets.get(seq)); } + getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise { + if (totalMsgCount === 0n) { + return Promise.resolve(this.buckets.get(0n)); + } + return Promise.resolve([...this.buckets.values()].find(bucket => bucket.totalMsgCount === totalMsgCount)); + } + getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise { const atOrBefore = [...this.buckets.values()] .filter(bucket => bucket.timestamp <= timestamp) @@ -52,6 +59,15 @@ export class MockL1ToL2MessageSource implements L1ToL2MessageSource { return Promise.resolve(seqs.flatMap(seq => this.messagesPerBucket.get(seq) ?? [])); } + async getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise { + const startBucket = await this.getInboxBucketByTotalMsgCount(startLeafCount); + const endBucket = await this.getInboxBucketByTotalMsgCount(endLeafCount); + if (startBucket === undefined || endBucket === undefined) { + throw new Error(`No mocked Inbox bucket boundary at ${startLeafCount} or ${endLeafCount}`); + } + return this.getL1ToL2MessagesBetweenBuckets(startBucket.seq, endBucket.seq); + } + getBlockNumber() { return Promise.resolve(BlockNumber(this.blockNumber)); } diff --git a/yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts b/yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts index 8a14968a9288..f9faa376374e 100644 --- a/yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts +++ b/yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts @@ -208,7 +208,7 @@ export class LightweightCheckpointBuilder { timings.insertSideEffects = msInsertSideEffects; } - // Streaming Inbox (AZIP-22 Fast Inbox): insert this block's L1-to-L2 message bundle before reading the end state, + // Streaming Inbox: insert this block's L1-to-L2 message bundle before reading the end state, // so the block header's L1-to-L2 tree snapshot reflects it. First-in-checkpoint bundles are padded to // NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP (matching the legacy per-checkpoint insertion and the world-state // synchronizer); non-first bundles are appended compactly. The logical (unpadded) messages are accumulated only diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts index b8c9cf995d25..029a3c62bd67 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts @@ -52,7 +52,12 @@ import { type ResolvedSequencerConfig, type WorldStateSynchronizer, } from '@aztec/stdlib/interfaces/server'; -import { InboxBucketRef, type L1ToL2MessageSource, computeInHashFromL1ToL2Messages } from '@aztec/stdlib/messaging'; +import { + InboxBucketRef, + type L1ToL2MessageSource, + computeInHashFromL1ToL2Messages, + getInboxCutoffTimestamp, +} from '@aztec/stdlib/messaging'; import type { BlockProposal, BlockProposalOptions, @@ -105,7 +110,7 @@ type CheckpointProposalResult = { }; /** - * Running state of streaming Inbox message selection across the blocks of one checkpoint (AZIP-22 Fast Inbox). + * Running state of streaming Inbox message selection across the blocks of one checkpoint. * Consumption starts from the parent checkpoint's last-consumed bucket and advances one block at a time. */ type StreamingCheckpointState = { @@ -1119,7 +1124,7 @@ export class CheckpointProposalJob implements Traceable { /** * Selects this block's streaming-Inbox message bundle against the current consumption cursor, mirroring the L1 - * predicate in `ProposeLib.validateInboxConsumption` (AZIP-22 Fast Inbox). Does not mutate the cursor; the caller + * predicate in `ProposeLib.validateInboxConsumption`. Does not mutate the cursor; the caller * advances it only after the block builds successfully. */ private selectStreamingBundle( @@ -1127,9 +1132,7 @@ export class CheckpointProposalJob implements Traceable { isLastBlock: boolean, nowSeconds: number, ): Promise { - // Mirror ProposeLib's cutoff exactly: buildFrameStart = toTimestamp(slot - 1), cutoff = buildFrameStart - lag. - const buildFrameStart = getTimestampForSlot(SlotNumber(this.targetSlot - 1), this.l1Constants); - const cutoffTimestamp = buildFrameStart - BigInt(INBOX_LAG_SECONDS); + const cutoffTimestamp = getInboxCutoffTimestamp(this.targetSlot, this.l1Constants, INBOX_LAG_SECONDS); return selectInboxBucketForBlock({ messageSource: this.l1ToL2MessageSource, now: BigInt(Math.floor(nowSeconds)), diff --git a/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.test.ts b/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.test.ts index 3e811527d29f..4ccd4d9b0eaa 100644 --- a/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.test.ts +++ b/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.test.ts @@ -62,7 +62,8 @@ function makeSource(specs: TestBucketSpec[]): { const GENESIS_PARENT = { seq: 0n, totalMsgCount: 0n }; -// Pinned cross-layer values from A-1371-resolution §13: genesisTime=100000, slotDuration=36, INBOX_LAG_SECONDS=12. +// Pinned cross-layer values shared with the L1 Foundry harness: genesisTime=100000, slotDuration=36, +// INBOX_LAG_SECONDS=12. const GENESIS_TIME = 100000n; const SLOT_DURATION = 36n; const LAG = 12n; diff --git a/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.ts b/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.ts index e56c4bcaef3f..b8a984bcb89a 100644 --- a/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.ts +++ b/yarn-project/sequencer-client/src/sequencer/inbox_bucket_selector.ts @@ -1,7 +1,7 @@ import type { Fr } from '@aztec/foundation/curves/bn254'; import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging'; -/** The subset of the archiver's Inbox-bucket queries the selector needs (AZIP-22 Fast Inbox). */ +/** The subset of the archiver's Inbox-bucket queries the selector needs. */ export type InboxBucketSource = Pick< L1ToL2MessageSource, 'getInboxBucket' | 'getLatestInboxBucketAtOrBefore' | 'getL1ToL2MessagesBetweenBuckets' @@ -57,7 +57,7 @@ export type InboxBucketSelection = /** * Selects the newest Inbox bucket a block streams from, mirroring the L1 consumption predicate in - * `ProposeLib.validateInboxConsumption` (AZIP-22 Fast Inbox). The policy, per block: + * `ProposeLib.validateInboxConsumption`. The policy, per block: * * 1. Pick the newest lag-eligible bucket: the newest bucket opened at or before `now - lagSeconds`. On the * checkpoint's last block, also consider the cutoff bucket (newest opened at or before `cutoffTimestamp`) and take diff --git a/yarn-project/stdlib/src/interfaces/archiver.test.ts b/yarn-project/stdlib/src/interfaces/archiver.test.ts index 11658128468e..9a56a0fd5eb8 100644 --- a/yarn-project/stdlib/src/interfaces/archiver.test.ts +++ b/yarn-project/stdlib/src/interfaces/archiver.test.ts @@ -224,11 +224,21 @@ describe('ArchiverApiSchema', () => { expect(result).toMatchObject({ seq: 2n, msgCount: 3 }); }); + it('getInboxBucketByTotalMsgCount', async () => { + const result = await context.client.getInboxBucketByTotalMsgCount(3n); + expect(result).toMatchObject({ seq: 2n, totalMsgCount: 3n, msgCount: 3 }); + }); + it('getL1ToL2MessagesBetweenBuckets', async () => { const result = await context.client.getL1ToL2MessagesBetweenBuckets(0n, 3n); expect(result).toEqual([expect.any(Fr)]); }); + it('getL1ToL2MessagesBetweenLeafCounts', async () => { + const result = await context.client.getL1ToL2MessagesBetweenLeafCounts(0n, 3n); + expect(result).toEqual([expect.any(Fr)]); + }); + it('registerContractFunctionSignatures', async () => { await context.client.registerContractFunctionSignatures(['test()']); }); @@ -589,11 +599,27 @@ class MockArchiver implements ArchiverApi { lastMessageIndex: 2n, }); } + getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise { + expect(typeof totalMsgCount).toEqual('bigint'); + return Promise.resolve({ + seq: 2n, + inboxRollingHash: Fr.random(), + totalMsgCount, + timestamp: 100n, + msgCount: 3, + lastMessageIndex: 2n, + }); + } getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { expect(typeof fromExclusive).toEqual('bigint'); expect(typeof toInclusive).toEqual('bigint'); return Promise.resolve([Fr.random()]); } + getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise { + expect(typeof startLeafCount).toEqual('bigint'); + expect(typeof endLeafCount).toEqual('bigint'); + return Promise.resolve([Fr.random()]); + } getL1Constants(): Promise { return Promise.resolve(EmptyL1RollupConstants); } diff --git a/yarn-project/stdlib/src/interfaces/archiver.ts b/yarn-project/stdlib/src/interfaces/archiver.ts index d1934e803b7b..3728c39090cc 100644 --- a/yarn-project/stdlib/src/interfaces/archiver.ts +++ b/yarn-project/stdlib/src/interfaces/archiver.ts @@ -143,10 +143,18 @@ export const ArchiverApiSchema: ApiSchemaFor = { output: InboxBucketSchema.optional(), }), getInboxBucket: z.function({ input: z.tuple([schemas.BigInt]), output: InboxBucketSchema.optional() }), + getInboxBucketByTotalMsgCount: z.function({ + input: z.tuple([schemas.BigInt]), + output: InboxBucketSchema.optional(), + }), getL1ToL2MessagesBetweenBuckets: z.function({ input: z.tuple([schemas.BigInt, schemas.BigInt]), output: z.array(schemas.Fr), }), + getL1ToL2MessagesBetweenLeafCounts: z.function({ + input: z.tuple([schemas.BigInt, schemas.BigInt]), + output: z.array(schemas.Fr), + }), getDebugFunctionName: z.function({ input: z.tuple([schemas.AztecAddress, schemas.FunctionSelector]), output: optional(z.string()), diff --git a/yarn-project/stdlib/src/interfaces/validator.ts b/yarn-project/stdlib/src/interfaces/validator.ts index f17e96e2a475..a7d1ae245daf 100644 --- a/yarn-project/stdlib/src/interfaces/validator.ts +++ b/yarn-project/stdlib/src/interfaces/validator.ts @@ -83,7 +83,10 @@ export type ValidatorClientConfig = ValidatorHASignerConfig & }; export type ValidatorClientFullConfig = ValidatorClientConfig & - Pick & + Pick< + SequencerConfig, + 'txPublicSetupAllowListExtend' | 'broadcastInvalidBlockProposal' | 'maxBlocksPerCheckpoint' | 'streamingInbox' + > & // `blockDurationMs` is optional on the loose `SequencerConfig` but is always populated via the shared // `numberConfigHelper(3000)` mapping, so it is required on the fully-resolved validator config. Required> & @@ -134,6 +137,7 @@ export const ValidatorClientFullConfigSchema = zodFor; /** - * Reference to a settled Inbox rolling-hash bucket, carried alongside a block proposal (AZIP-22 Fast Inbox) so a + * Reference to a settled Inbox rolling-hash bucket, carried alongside a block proposal so a * validator can look the bucket up in its own Inbox view and derive the consumed-message bundle itself, rather than * trusting a proposer-supplied message list. Pins the bucket by its dense sequence number and recency-key timestamp * and asserts the expected consensus rolling hash. A wrong reference can only cause a lookup miss or hash mismatch; it diff --git a/yarn-project/stdlib/src/messaging/inbox_consumption.test.ts b/yarn-project/stdlib/src/messaging/inbox_consumption.test.ts new file mode 100644 index 000000000000..36d1f7653b0f --- /dev/null +++ b/yarn-project/stdlib/src/messaging/inbox_consumption.test.ts @@ -0,0 +1,95 @@ +import { SlotNumber } from '@aztec/foundation/branded-types'; + +import { describe, expect, it } from '@jest/globals'; + +import type { L1RollupConstants } from '../epoch-helpers/index.js'; +import { getInboxCutoffTimestamp, isInboxConsumptionSufficient } from './inbox_consumption.js'; + +// Cross-layer test vectors shared with the `ProposeInboxConsumptionTest` Foundry harness. +// The same vectors are asserted against `ProposeLib.validateInboxConsumption` on L1; keeping them identical here makes +// L1, TS, and the design doc agree on the cutoff formula and the mandatory-consumption boundary. +const GENESIS_TIME = 100_000n; +const SLOT_DURATION = 36; +const LAG_SECONDS = 12; + +const l1Constants = { l1GenesisTime: GENESIS_TIME, slotDuration: SLOT_DURATION } as Pick< + L1RollupConstants, + 'l1GenesisTime' | 'slotDuration' +>; + +describe('inbox_consumption', () => { + describe('getInboxCutoffTimestamp', () => { + // buildFrameStart(S) = 100000 + (S - 1) * 36; cutoff(S) = buildFrameStart(S) - 12. + it.each([ + [1, 99_988n], + [2, 100_024n], + [10, 100_312n], + [11, 100_348n], + ])('matches the A-1371 §13 cutoff table for slot %i', (slot, expectedCutoff) => { + expect(getInboxCutoffTimestamp(SlotNumber(slot), l1Constants, LAG_SECONDS)).toBe(expectedCutoff); + }); + }); + + describe('isInboxConsumptionSufficient', () => { + const base = { cutoffTimestamp: 100_312n, checkpointStartTotalMsgCount: 0n, perCheckpointCap: 1024 }; + + it('is sufficient when there is no next bucket (consumed everything)', () => { + expect(isInboxConsumptionSufficient({ ...base, nextBucket: undefined })).toBe(true); + }); + + // Boundary at S=10 (cutoff = 100312): a bucket opened exactly at the cutoff is mandatory (strict `>`). + it('is insufficient when the next bucket opened exactly at the cutoff is left unconsumed', () => { + expect(isInboxConsumptionSufficient({ ...base, nextBucket: { timestamp: 100_312n, totalMsgCount: 5n } })).toBe( + false, + ); + }); + + // Boundary at S=10: a bucket opened at cutoff + 1 is past the cutoff and need not be consumed. + it('is sufficient when the next bucket opened at cutoff + 1 is left unconsumed', () => { + expect(isInboxConsumptionSufficient({ ...base, nextBucket: { timestamp: 100_313n, totalMsgCount: 5n } })).toBe( + true, + ); + }); + + it('is sufficient via cap-escape when consuming through the next bucket would exceed the per-checkpoint cap', () => { + // Next bucket is at/before the cutoff (would otherwise be mandatory), but consuming through it consumes + // 1025 > 1024 messages, so leaving it unconsumed is allowed (the cap-escape branch). + expect( + isInboxConsumptionSufficient({ + ...base, + nextBucket: { timestamp: 100_000n, totalMsgCount: 1025n }, + }), + ).toBe(true); + }); + + it('is insufficient when consuming through the next bucket exactly reaches the per-checkpoint cap', () => { + // Delta of exactly 1024 does not escape (strict `>` on the cap), so a mandatory bucket must still be consumed. + expect( + isInboxConsumptionSufficient({ + ...base, + nextBucket: { timestamp: 100_000n, totalMsgCount: 1024n }, + }), + ).toBe(false); + }); + + it('measures the cap-escape delta from the checkpoint start, not from zero', () => { + // With a non-zero checkpoint start, only the delta consumed this checkpoint counts against the cap. + expect( + isInboxConsumptionSufficient({ + cutoffTimestamp: 100_312n, + checkpointStartTotalMsgCount: 500n, + perCheckpointCap: 1024, + nextBucket: { timestamp: 100_000n, totalMsgCount: 1524n }, + }), + ).toBe(false); // delta = 1024, not an escape + expect( + isInboxConsumptionSufficient({ + cutoffTimestamp: 100_312n, + checkpointStartTotalMsgCount: 500n, + perCheckpointCap: 1024, + nextBucket: { timestamp: 100_000n, totalMsgCount: 1525n }, + }), + ).toBe(true); // delta = 1025, escapes + }); + }); +}); diff --git a/yarn-project/stdlib/src/messaging/inbox_consumption.ts b/yarn-project/stdlib/src/messaging/inbox_consumption.ts new file mode 100644 index 000000000000..d2649dba8579 --- /dev/null +++ b/yarn-project/stdlib/src/messaging/inbox_consumption.ts @@ -0,0 +1,53 @@ +import { SlotNumber } from '@aztec/foundation/branded-types'; + +import { type L1RollupConstants, getTimestampForSlot } from '../epoch-helpers/index.js'; +import type { InboxBucket } from './inbox_bucket.js'; + +/** + * Censorship cutoff timestamp for a checkpoint proposed in `slot`, mirroring the cutoff in + * `ProposeLib.validateInboxConsumption`. A checkpoint proposed in slot `S` is built during slot `S - 1`, so + * `buildFrameStart(S) = toTimestamp(S - 1)` and `cutoff(S) = buildFrameStart(S) - lagSeconds`. Buckets opened at or + * before the cutoff are mandatory to consume by the checkpoint's last block; the strict `>` on the L1 "past cutoff" + * test (see {@link isInboxConsumptionSufficient}) makes a bucket opened exactly at the cutoff mandatory. + * + * This is the single source of truth for the cutoff formula shared by the sequencer's streaming bucket selection and + * the validator's last-block censorship check. + */ +export function getInboxCutoffTimestamp( + slot: SlotNumber, + l1Constants: Pick, + lagSeconds: number, +): bigint { + return getTimestampForSlot(SlotNumber(slot - 1), l1Constants) - BigInt(lagSeconds); +} + +/** + * Whether a checkpoint whose last-consumed bucket is immediately followed by `nextBucket` meets the censorship floor, + * mirroring the mandatory-consumption assert in `ProposeLib.validateInboxConsumption`. + * Consumption is sufficient when the first unconsumed bucket: + * - does not exist (the checkpoint consumed everything the Inbox has), or + * - was opened strictly after the cutoff (`timestamp > cutoffTimestamp`), or + * - consuming through it would exceed the per-checkpoint cap (the cap-escape). + * + * The strict `>` matches L1: a bucket opened exactly at the cutoff must be consumed. This is the single source of + * truth for the minimum-consumption / cap-escape rule shared by the sequencer's selection floor and the validator. + */ +export function isInboxConsumptionSufficient(input: { + /** The first unconsumed bucket (the one after the checkpoint's last-consumed bucket), or undefined if none exists. */ + nextBucket: Pick | undefined; + /** Censorship cutoff timestamp from {@link getInboxCutoffTimestamp}. */ + cutoffTimestamp: bigint; + /** Cumulative Inbox message count consumed as of the parent checkpoint; the per-checkpoint cap origin. */ + checkpointStartTotalMsgCount: bigint; + /** Maximum number of messages the checkpoint may consume in total (`MAX_L1_TO_L2_MSGS_PER_CHECKPOINT`). */ + perCheckpointCap: number; +}): boolean { + const { nextBucket, cutoffTimestamp, checkpointStartTotalMsgCount, perCheckpointCap } = input; + if (nextBucket === undefined) { + return true; + } + if (nextBucket.timestamp > cutoffTimestamp) { + return true; + } + return nextBucket.totalMsgCount - checkpointStartTotalMsgCount > BigInt(perCheckpointCap); +} diff --git a/yarn-project/stdlib/src/messaging/index.ts b/yarn-project/stdlib/src/messaging/index.ts index da0c32490353..aad4f1a91990 100644 --- a/yarn-project/stdlib/src/messaging/index.ts +++ b/yarn-project/stdlib/src/messaging/index.ts @@ -1,6 +1,7 @@ export * from './append_l1_to_l2_messages.js'; export * from './in_hash.js'; export * from './inbox_bucket.js'; +export * from './inbox_consumption.js'; export * from './inbox_leaf.js'; export * from './inbox_rolling_hash.js'; export * from './l1_to_l2_message_bundle.js'; diff --git a/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts b/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts index e30b78ecfdd8..82610dbb461a 100644 --- a/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts +++ b/yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts @@ -27,21 +27,32 @@ export interface L1ToL2MessageSource { /** * Returns the latest Inbox bucket opened at or before the given L1 timestamp, or undefined if no such bucket * exists (i.e., every synced bucket was opened strictly after the timestamp). Used by the sequencer/validator - * to resolve the censorship cutoff and message-lag boundaries (AZIP-22 Fast Inbox). + * to resolve the censorship cutoff and message-lag boundaries. * @param timestamp - The L1 timestamp (in seconds) to look up at-or-before. */ getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise; /** - * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced (AZIP-22 Fast - * Inbox). Validators use this to resolve the bucket a proposal references and check its rolling hash. + * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced. Validators use + * this to resolve the bucket a proposal references and check its rolling hash. * @param seq - The bucket sequence number. */ getInboxBucket(seq: bigint): Promise; + /** + * Returns the Inbox bucket whose cumulative message total equals `totalMsgCount`, or undefined if no synced bucket + * sits on that boundary. A block's or checkpoint's L1-to-L2 tree leaf count equals the cumulative total of the last + * bucket it consumed (messages are indexed compactly, with no padding), so validators use this to resolve the + * bucket a block consumed through from the block's leaf count, without trusting a wire hint. `totalMsgCount === 0` + * resolves the genesis sentinel bucket (sequence 0); a total that does not land on a bucket boundary returns + * undefined. + * @param totalMsgCount - The cumulative Inbox message count (leaf count) to resolve to a bucket boundary. + */ + getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise; + /** * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion - * order, for streaming message-bundle derivation (AZIP-22 Fast Inbox). Both bounds must name buckets the source + * order, for streaming message-bundle derivation. Both bounds must name buckets the source * has synced; it throws otherwise, so that an empty result means the range holds no messages instead of hiding an * unsynced bound. Callers that can tolerate an unsynced source resolve both bounds first, or map the failure to * their own catch-up handling. @@ -50,6 +61,16 @@ export interface L1ToL2MessageSource { */ getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise; + /** + * Returns the message leaves in the cumulative Inbox message-count range `[startLeafCount, endLeafCount)`, in + * insertion order. The bounds are compact L1-to-L2 tree leaf counts, which every block header + * carries, so a consumer can ask for the messages a block or checkpoint consumed without resolving Inbox buckets + * itself. Both bounds must land on a bucket boundary the source has synced; it throws otherwise. + * @param startLeafCount - The cumulative Inbox message count the range starts at, inclusive. + * @param endLeafCount - The cumulative Inbox message count the range ends at, exclusive. + */ + getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise; + /** * Returns the tips of the L2 chain. */ diff --git a/yarn-project/stdlib/src/p2p/block_proposal.ts b/yarn-project/stdlib/src/p2p/block_proposal.ts index 13b76b08e938..750fde3f77ed 100644 --- a/yarn-project/stdlib/src/p2p/block_proposal.ts +++ b/yarn-project/stdlib/src/p2p/block_proposal.ts @@ -90,9 +90,9 @@ export class BlockProposal extends Gossipable implements Signable { public readonly signedTxs?: SignedTxs, /** - * Reference to the Inbox bucket this block proposes to consume (AZIP-22 Fast Inbox). Optional pre-flip: the - * sequencer leaves it unset until the streaming Inbox is enabled, at which point validators derive the consumed - * message bundle from it. Covered by the proposal signature (part of `getPayloadToSign`). + * Reference to the Inbox bucket this block proposes to consume, when the proposer commits to one. Validators + * resolve it against their own Inbox view and derive the consumed message bundle from it rather than trusting a + * proposer-supplied message list. Covered by the proposal signature (part of `getPayloadToSign`). */ public readonly bucketRef?: InboxBucketRef, ) { diff --git a/yarn-project/stdlib/src/p2p/checkpoint_proposal.ts b/yarn-project/stdlib/src/p2p/checkpoint_proposal.ts index 6700563587f3..126e24b56dc4 100644 --- a/yarn-project/stdlib/src/p2p/checkpoint_proposal.ts +++ b/yarn-project/stdlib/src/p2p/checkpoint_proposal.ts @@ -71,8 +71,8 @@ export type CheckpointLastBlock = Omit & { /** The signed transactions in the last block (optional, for DA guarantees) */ signedTxs?: SignedTxs; /** - * Reference to the Inbox bucket the last block proposes to consume (AZIP-22 Fast Inbox). Optional pre-flip; when set, - * its rolling hash must equal the checkpoint header's `inboxRollingHash` (enforced at construction). + * Reference to the Inbox bucket the last block proposes to consume. When set, its rolling hash must equal the + * checkpoint header's `inboxRollingHash` (enforced at construction). */ bucketRef?: InboxBucketRef; }; diff --git a/yarn-project/validator-client/src/checkpoint_builder.ts b/yarn-project/validator-client/src/checkpoint_builder.ts index 291298b3a574..81ff4d0e4450 100644 --- a/yarn-project/validator-client/src/checkpoint_builder.ts +++ b/yarn-project/validator-client/src/checkpoint_builder.ts @@ -375,6 +375,10 @@ export class FullNodeCheckpointsBuilder implements ICheckpointsBuilder { fork: MerkleTreeWriteOperations, existingBlocks: L2Block[] = [], bindings?: LoggerBindings, + // Streaming Inbox (AZIP-22 Fast Inbox): when true the fresh-checkpoint path inserts messages per block (via + // `buildBlock`'s `l1ToL2Messages`) instead of the whole checkpoint up front; `l1ToL2Messages` here must be empty. + // The resume path never inserts messages up front, so this only affects `startCheckpoint`. + insertMessagesPerBlock: boolean = false, ): Promise { const stateReference = await fork.getStateReference(); const archiveTree = await fork.getTreeInfo(MerkleTreeId.ARCHIVE); @@ -389,6 +393,7 @@ export class FullNodeCheckpointsBuilder implements ICheckpointsBuilder { previousInboxRollingHash, fork, bindings, + insertMessagesPerBlock, ); } diff --git a/yarn-project/validator-client/src/config.ts b/yarn-project/validator-client/src/config.ts index 38d914a6d6af..6b130b40190f 100644 --- a/yarn-project/validator-client/src/config.ts +++ b/yarn-project/validator-client/src/config.ts @@ -21,9 +21,9 @@ export type { ValidatorClientConfig }; export const DEFAULT_MAX_GOSSIP_CLOCK_DISPARITY_MS = 500; export const validatorClientConfigMappings: ConfigMappingsType< - ValidatorClientConfig & Pick + ValidatorClientConfig & Pick > = { - ...pickConfigMappings(sharedSequencerConfigMappings, ['blockDurationMs']), + ...pickConfigMappings(sharedSequencerConfigMappings, ['blockDurationMs', 'streamingInbox']), validatorPrivateKeys: { env: 'VALIDATOR_PRIVATE_KEYS', description: 'List of private keys of the validators participating in attestation duties', @@ -123,8 +123,9 @@ export const validatorClientConfigMappings: ConfigMappingsType< * Note: If an environment variable is not set, the default value is used. * @returns The validator configuration. */ -export function getProverEnvVars(): ValidatorClientConfig & Pick { - return getConfigFromMappings>( +export function getProverEnvVars(): ValidatorClientConfig & + Pick { + return getConfigFromMappings>( validatorClientConfigMappings, ); } diff --git a/yarn-project/validator-client/src/proposal_handler.test.ts b/yarn-project/validator-client/src/proposal_handler.test.ts index f3a414719329..ee89c5182f20 100644 --- a/yarn-project/validator-client/src/proposal_handler.test.ts +++ b/yarn-project/validator-client/src/proposal_handler.test.ts @@ -15,8 +15,8 @@ import type { BlockData, L2Block, L2BlockSink, L2BlockSource } from '@aztec/stdl import { type Checkpoint, CheckpointReexecutionTracker, type ProposedCheckpointData } from '@aztec/stdlib/checkpoint'; import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers'; import type { ITxProvider, ValidatorClientFullConfig, WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server'; -import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging'; -import { accumulateCheckpointOutHashes } from '@aztec/stdlib/messaging'; +import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging'; +import { InboxBucketRef, accumulateCheckpointOutHashes } from '@aztec/stdlib/messaging'; import { CheckpointHeader } from '@aztec/stdlib/rollup'; import { TEST_COORDINATION_SIGNATURE_CONTEXT, @@ -798,4 +798,217 @@ describe('ProposalHandler checkpoint validation', () => { }); }); }); + + // AZIP-22 Fast Inbox: with `streamingInbox` on, a block proposal's L1-to-L2 bundle is derived from its bucket + // reference and gated by the four acceptance checks, replacing the legacy per-checkpoint inHash comparison. + describe('handleBlockProposal streaming inbox checks (flag on)', () => { + const bucket = (overrides: Partial = {}): InboxBucket => ({ + seq: 1n, + inboxRollingHash: new Fr(0xabc), + totalMsgCount: 2n, + timestamp: 100n, + msgCount: 2, + lastMessageIndex: 1n, + ...overrides, + }); + + /** Genesis-parent streaming block proposal at slot 1, with the handler wired to reach the streaming checks. */ + async function setupStreamingProposal(bucketRef: InboxBucketRef | undefined) { + const proposal = await makeBlockProposal({ + blockHeader: makeBlockHeader(1, { slotNumber: SlotNumber(1) }), + archiveRoot: Fr.random(), + txHashes: [], + bucketRef, + }); + blockSource.getGenesisValues.mockResolvedValue({ + genesisArchiveRoot: proposal.blockHeader.lastArchive.root, + } as any); + blockSource.getBlockData.mockResolvedValue(undefined); + + const blockProposalValidator = mock(); + blockProposalValidator.validate.mockResolvedValue({ result: 'accept' } as any); + const txProvider = mock(); + txProvider.getTxsForBlockProposal.mockResolvedValue({ txs: [], missingTxs: [] } as any); + + // Well past the 12s lag for a bucket opened at t=100. + dateProvider.setTime(1_000_000); + + const blockHandler = new ProposalHandler( + checkpointsBuilder, + mock(), + blockSource, + l1ToL2MessageSource, + txProvider, + blockProposalValidator, + epochCache, + consensusTimetable, + { ...config, streamingInbox: true } as ValidatorClientFullConfig, + mock(), + new CheckpointReexecutionTracker(), + metrics, + dateProvider, + ); + return { proposal, blockHandler }; + } + + it('rejects (without re-executing) when the referenced bucket is unknown', async () => { + const ref = new InboxBucketRef(1n, 100n, new Fr(0xabc)); + const { proposal, blockHandler } = await setupStreamingProposal(ref); + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(undefined); + const reexecuteSpy = jest.spyOn(blockHandler, 'reexecuteTransactions'); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result).toEqual({ + isValid: false, + blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM), + reason: 'bucket_unknown', + }); + expect(reexecuteSpy).not.toHaveBeenCalled(); + }); + + it('rejects (without re-executing) when the resolved bucket hash disagrees with the reference', async () => { + const ref = new InboxBucketRef(1n, 100n, new Fr(0xdead)); + const { proposal, blockHandler } = await setupStreamingProposal(ref); + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(bucket({ inboxRollingHash: new Fr(0xabc) })); + const reexecuteSpy = jest.spyOn(blockHandler, 'reexecuteTransactions'); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result).toEqual({ + isValid: false, + blockNumber: BlockNumber(INITIAL_L2_BLOCK_NUM), + reason: 'bucket_hash_mismatch', + }); + expect(reexecuteSpy).not.toHaveBeenCalled(); + }); + + it('re-executes with the bundle derived from the buckets when the checks pass', async () => { + const ref = new InboxBucketRef(1n, 100n, new Fr(0xabc)); + const { proposal, blockHandler } = await setupStreamingProposal(ref); + const derivedBundle = [new Fr(1000), new Fr(1001)]; + l1ToL2MessageSource.getInboxBucket.mockResolvedValue(bucket()); + l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockResolvedValue( + bucket({ seq: 0n, totalMsgCount: 0n, msgCount: 0 }), + ); + l1ToL2MessageSource.getL1ToL2MessagesBetweenBuckets.mockResolvedValue(derivedBundle); + const reexecuteSpy = jest + .spyOn(blockHandler, 'reexecuteTransactions') + .mockResolvedValue({ block: undefined } as any); + + const result = await blockHandler.handleBlockProposal(proposal, {} as any, true); + + expect(result.isValid).toBe(true); + // The block re-executes with the derived per-block bundle and the streaming flag set. + expect(reexecuteSpy).toHaveBeenCalledWith( + proposal, + BlockNumber(INITIAL_L2_BLOCK_NUM), + CheckpointNumber.INITIAL, + [], + derivedBundle, + expect.anything(), + expect.anything(), + true, + ); + }); + }); + + // AZIP-22 Fast Inbox: with `streamingInbox` on, the checkpoint handler enforces the last-block minimum-consumption + // (censorship) rule before attesting. + describe('checkpoint proposal last-block censorship (flag on)', () => { + /** Two-block checkpoint at slot 10 whose last block consumed through leaf count `lastBlockTotal`. */ + function setupCensorshipMocks(lastBlockTotal: number) { + const archiveRoot = Fr.random(); + const blockWithLeafCount = (leafCount: number, archive: Fr, number: number) => + ({ + archive: new AppendOnlyTreeSnapshot(archive, number), + number, + checkpointNumber: CheckpointNumber(1), + header: { + globalVariables: GlobalVariables.empty({ slotNumber: SlotNumber(10) }), + state: { l1ToL2MessageTree: { nextAvailableLeafIndex: leafCount } }, + }, + }) as unknown as L2Block; + + blockSource.getBlockData.mockResolvedValue({ + header: makeBlockHeader(), + checkpointNumber: CheckpointNumber(1), + } as any); + blockSource.getBlocksForSlot.mockResolvedValue([ + blockWithLeafCount(0, Fr.random(), 1), + blockWithLeafCount(lastBlockTotal, archiveRoot, 2), + ]); + return { archiveRoot }; + } + + async function makeSlot10Proposal(archiveRoot: Fr) { + return ( + await makeCheckpointProposal({ + checkpointHeader: makeCheckpointHeader(0, { slotNumber: SlotNumber(10) }), + archiveRoot, + }) + ).toCore(); + } + + it('refuses to attest when a mandatory bucket (at or before the cutoff) is left unconsumed', async () => { + config = { ...config, streamingInbox: true } as ValidatorClientFullConfig; + handler.updateConfig(config); + const { archiveRoot } = setupCensorshipMocks(2); + // cutoff(slot=10) with l1GenesisTime=0, slotDuration=24, lag=12 is (10-1)*24 - 12 = 204. + l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockResolvedValue({ + seq: 1n, + totalMsgCount: 2n, + } as InboxBucket); + // The next (first unconsumed) bucket opened at t=100 <= cutoff 204 is mandatory and was left unconsumed. + l1ToL2MessageSource.getInboxBucket.mockResolvedValue({ + seq: 2n, + totalMsgCount: 5n, + timestamp: 100n, + } as InboxBucket); + + const result = await handler.handleCheckpointProposal(await makeSlot10Proposal(archiveRoot), proposalInfo); + + expect(result).toEqual({ + isValid: false, + reason: 'inbox_consumption_insufficient', + checkpointNumber: CheckpointNumber(1), + }); + }); + + it('does not reject on censorship when the first unconsumed bucket is past the cutoff', async () => { + config = { ...config, streamingInbox: true } as ValidatorClientFullConfig; + handler.updateConfig(config); + const { archiveRoot } = setupCensorshipMocks(2); + l1ToL2MessageSource.getInboxBucketByTotalMsgCount.mockResolvedValue({ + seq: 1n, + totalMsgCount: 2n, + } as InboxBucket); + // Next bucket opened at t=205 > cutoff 204: not mandatory, so the censorship check passes and validation + // proceeds past it to the checkpoint rebuild (which mismatches here, an unrelated reason). + l1ToL2MessageSource.getInboxBucket.mockResolvedValue({ + seq: 2n, + totalMsgCount: 5n, + timestamp: 205n, + } as InboxBucket); + checkpointsBuilder.getFork.mockResolvedValue({ [Symbol.asyncDispose]: jest.fn() } as any); + const mockBuilder = mock(); + mockBuilder.completeCheckpoint.mockResolvedValue({ + header: CheckpointHeader.empty(), + archive: new AppendOnlyTreeSnapshot(Fr.ZERO, 0), + getCheckpointOutHash: () => Fr.random(), + blocks: [], + number: CheckpointNumber(1), + } as unknown as Checkpoint); + checkpointsBuilder.openCheckpoint.mockResolvedValue(mockBuilder); + + const result = await handler.handleCheckpointProposal(await makeSlot10Proposal(archiveRoot), proposalInfo); + + // The censorship rule passed; the checkpoint is rejected later by the rebuild, not the consumption check. + expect(result).toEqual({ + isValid: false, + reason: 'checkpoint_header_mismatch', + checkpointNumber: CheckpointNumber(1), + }); + }); + }); }); diff --git a/yarn-project/validator-client/src/proposal_handler.ts b/yarn-project/validator-client/src/proposal_handler.ts index 96f20aada864..722a0d9b86d5 100644 --- a/yarn-project/validator-client/src/proposal_handler.ts +++ b/yarn-project/validator-client/src/proposal_handler.ts @@ -1,7 +1,12 @@ import type { Archiver } from '@aztec/archiver'; import type { BlobClientInterface } from '@aztec/blob-client/client'; import { type Blob, encodeCheckpointBlobDataFromBlocks, getBlobsPerL1Block } from '@aztec/blob-lib'; -import { INITIAL_L2_BLOCK_NUM } from '@aztec/constants'; +import { + INBOX_LAG_SECONDS, + INITIAL_L2_BLOCK_NUM, + MAX_L1_TO_L2_MSGS_PER_BLOCK, + MAX_L1_TO_L2_MSGS_PER_CHECKPOINT, +} from '@aztec/constants'; import type { EpochCache } from '@aztec/epoch-cache'; import { validateFeeAssetPriceModifier } from '@aztec/ethereum/contracts'; import { @@ -39,6 +44,8 @@ import { type L1ToL2MessageSource, accumulateCheckpointOutHashes, computeInHashFromL1ToL2Messages, + getInboxCutoffTimestamp, + isInboxConsumptionSufficient, } from '@aztec/stdlib/messaging'; import type { BlockProposal, CheckpointAttestation, CheckpointProposalCore } from '@aztec/stdlib/p2p'; import type { ConsensusTimetable } from '@aztec/stdlib/timetable'; @@ -55,6 +62,11 @@ import { type TelemetryClient, type Tracer, getTelemetryClient } from '@aztec/te import type { FullNodeCheckpointsBuilder } from './checkpoint_builder.js'; import type { ValidatorMetrics } from './metrics.js'; +import { + type StreamingBlockCheckReason, + type StreamingBlockCheckResult, + checkStreamingBlockProposal, +} from './streaming_inbox_checks.js'; export type BlockProposalValidationFailureReason = | 'invalid_signature' @@ -62,6 +74,8 @@ export type BlockProposalValidationFailureReason = | 'parent_block_not_found' | 'parent_block_wrong_slot' | 'in_hash_mismatch' + // Streaming Inbox (AZIP-22 Fast Inbox) per-block acceptance failures, gated behind `streamingInbox`. + | StreamingBlockCheckReason | 'global_variables_mismatch' | 'block_number_already_exists' | 'txs_not_available' @@ -109,6 +123,8 @@ export type CheckpointProposalValidationFailureReason = | 'checkpoint_header_mismatch' | 'archive_mismatch' | 'out_hash_mismatch' + // Streaming Inbox (AZIP-22 Fast Inbox) last-block censorship failure, gated behind `streamingInbox`. + | 'inbox_consumption_insufficient' | 'checkpoint_validation_failed'; /** @@ -134,6 +150,7 @@ const CHECKPOINT_VALIDATION_REASON_TO_OUTCOME: Record< checkpoint_header_mismatch: 'invalid', archive_mismatch: 'invalid', out_hash_mismatch: 'invalid', + inbox_consumption_insufficient: 'invalid', checkpoint_validation_failed: 'invalid', }; @@ -196,6 +213,9 @@ export const SLASHABLE_CHECKPOINT_PROPOSAL_VALIDATION_RESULT: Record< ['last_block_archive_mismatch']: true, // disabled + // Streaming Inbox last-block censorship: new and flag-gated (default off), so keep it out of slashing while the + // path lands; L1 `propose` is the authoritative reject (Rollup__UnconsumedInboxMessages) pre-flip. + ['inbox_consumption_insufficient']: false, ['invalid_signature']: false, ['last_block_not_found']: false, ['block_fetch_error']: false, @@ -556,17 +576,36 @@ export class ProposalHandler { const checkpointNumber = checkpointResult.checkpointNumber; proposalInfo.checkpointNumber = checkpointNumber; - // Check that I have the same set of l1ToL2Messages as the proposal - const l1ToL2Messages = await this.l1ToL2MessageSource.getL1ToL2Messages(checkpointNumber); - const computedInHash = computeInHashFromL1ToL2Messages(l1ToL2Messages); - const proposalInHash = proposal.inHash; - if (!computedInHash.equals(proposalInHash)) { - this.log.warn(`L1 to L2 messages in hash mismatch, skipping processing`, { - proposalInHash: proposalInHash.toString(), - computedInHash: computedInHash.toString(), - ...proposalInfo, - }); - return { isValid: false, blockNumber, reason: 'in_hash_mismatch' }; + // Resolve this block's L1-to-L2 message bundle. Under the streaming Inbox (AZIP-22 Fast Inbox) the block consumes + // a per-block bundle derived from its proposal bucket reference, gated by the four acceptance checks; the legacy + // flow compares the whole checkpoint's `inHash` and inserts all its messages up front. Flag off ⇒ byte-identical + // to before. + const streamingInbox = this.config.streamingInbox === true; + let l1ToL2Messages: Fr[]; + if (streamingInbox) { + const streamingResult = await this.runStreamingBlockChecks(proposal, blockNumber, parentBlock); + if (!streamingResult.accepted) { + this.log.warn(`Streaming Inbox block acceptance check failed, skipping processing`, { + reason: streamingResult.reason, + bucketRef: proposal.bucketRef?.toInspect(), + ...proposalInfo, + }); + return { isValid: false, blockNumber, reason: streamingResult.reason }; + } + l1ToL2Messages = streamingResult.bundle; + } else { + // Check that I have the same set of l1ToL2Messages as the proposal + l1ToL2Messages = await this.l1ToL2MessageSource.getL1ToL2Messages(checkpointNumber); + const computedInHash = computeInHashFromL1ToL2Messages(l1ToL2Messages); + const proposalInHash = proposal.inHash; + if (!computedInHash.equals(proposalInHash)) { + this.log.warn(`L1 to L2 messages in hash mismatch, skipping processing`, { + proposalInHash: proposalInHash.toString(), + computedInHash: computedInHash.toString(), + ...proposalInfo, + }); + return { isValid: false, blockNumber, reason: 'in_hash_mismatch' }; + } } // Check that all of the transactions in the proposal are available @@ -606,6 +645,7 @@ export class ProposalHandler { l1ToL2Messages, previousCheckpointOutHashes, previousInboxRollingHash, + streamingInbox, ); } catch (error) { this.log.error(`Error reexecuting txs while processing block proposal`, error, proposalInfo); @@ -886,6 +926,102 @@ export class ProposalHandler { } } + /** + * Runs the streaming-Inbox per-block acceptance checks for a block proposal and returns the derived L1-to-L2 + * message bundle for re-execution, or a rejection reason. The parent block's consumed total and the checkpoint's + * starting total are derived from L1-to-L2 tree leaf counts; a parent whose count does not sit on a bucket boundary + * is rejected inside {@link checkStreamingBlockProposal}. + */ + private async runStreamingBlockChecks( + proposal: BlockProposal, + blockNumber: BlockNumber, + parentBlock: 'genesis' | BlockData, + ): Promise { + const parentTotalMsgCount = this.getConsumedMsgTotal(parentBlock); + const checkpointStartTotalMsgCount = await this.resolveCheckpointStartTotal( + blockNumber, + proposal.indexWithinCheckpoint, + parentTotalMsgCount, + ); + 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. + return { accepted: false, reason: 'bucket_unknown' }; + } + const nowSeconds = BigInt(Math.floor(this.dateProvider.now() / 1000)); + return checkStreamingBlockProposal({ + messageSource: this.l1ToL2MessageSource, + bucketRef: proposal.bucketRef, + parentTotalMsgCount, + checkpointStartTotalMsgCount, + nowSeconds, + lagSeconds: INBOX_LAG_SECONDS, + perBlockCap: MAX_L1_TO_L2_MSGS_PER_BLOCK, + perCheckpointCap: MAX_L1_TO_L2_MSGS_PER_CHECKPOINT, + }); + } + + /** A block's L1-to-L2 message tree leaf count: the cumulative Inbox message count it consumed through. */ + private blockLeafCount(block: BlockData | L2Block): bigint { + return BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex); + } + + /** The cumulative Inbox message count consumed through a block: its L1-to-L2 tree leaf count (0 at genesis). */ + private getConsumedMsgTotal(block: 'genesis' | BlockData): bigint { + return block === 'genesis' ? 0n : this.blockLeafCount(block); + } + + /** + * The cumulative Inbox message count consumed as of the parent checkpoint (the per-checkpoint cap origin). For a + * checkpoint's first block this is the parent block's total; for a later block it is the leaf count of the block + * before the checkpoint's first block. Returns undefined when that block has not synced locally. + */ + private resolveCheckpointStartTotal( + blockNumber: BlockNumber, + indexWithinCheckpoint: number, + parentTotalMsgCount: bigint, + ): Promise { + return indexWithinCheckpoint === 0 + ? Promise.resolve(parentTotalMsgCount) + : this.getPreBlockConsumedTotal(blockNumber - indexWithinCheckpoint); + } + + /** + * The cumulative Inbox message count consumed as of the block immediately before `firstBlockNumber` (its L1-to-L2 + * tree leaf count): 0 when that block is genesis, undefined when it has not synced locally. + */ + private async getPreBlockConsumedTotal(firstBlockNumber: number): Promise { + const preBlockNumber = firstBlockNumber - 1; + if (preBlockNumber < INITIAL_L2_BLOCK_NUM) { + return 0n; + } + const preBlock = await this.blockSource.getBlockData({ number: BlockNumber(preBlockNumber) }); + return preBlock === undefined ? undefined : this.blockLeafCount(preBlock); + } + + /** + * Enforces the streaming-Inbox last-block minimum-consumption (censorship) rule for a checkpoint, mirroring + * `ProposeLib.validateInboxConsumption`: the first bucket the checkpoint left unconsumed must be absent, past the + * cutoff, or a cap-escape. Returns true (sufficient) when the checkpoint's consumption cannot be resolved against + * the local Inbox view, deferring to L1 `propose` as the authoritative reject. + */ + private async isLastBlockConsumptionSufficient(slot: SlotNumber, blocks: L2Block[]): Promise { + const lastBlockTotal = this.blockLeafCount(blocks[blocks.length - 1]); + const checkpointStartTotal = await this.getPreBlockConsumedTotal(blocks[0].number); + const lastConsumedBucket = await this.l1ToL2MessageSource.getInboxBucketByTotalMsgCount(lastBlockTotal); + if (checkpointStartTotal === undefined || lastConsumedBucket === undefined) { + return true; + } + const nextBucket = await this.l1ToL2MessageSource.getInboxBucket(lastConsumedBucket.seq + 1n); + const cutoffTimestamp = getInboxCutoffTimestamp(slot, this.epochCache.getL1Constants(), INBOX_LAG_SECONDS); + return isInboxConsumptionSufficient({ + nextBucket, + cutoffTimestamp, + checkpointStartTotalMsgCount: checkpointStartTotal, + perCheckpointCap: MAX_L1_TO_L2_MSGS_PER_CHECKPOINT, + }); + } + async reexecuteTransactions( proposal: BlockProposal, blockNumber: BlockNumber, @@ -894,6 +1030,9 @@ export class ProposalHandler { l1ToL2Messages: Fr[], previousCheckpointOutHashes: Fr[], previousInboxRollingHash: Fr, + // Streaming Inbox (AZIP-22 Fast Inbox): when true, `l1ToL2Messages` is this block's per-block bundle, inserted + // into the fork during `buildBlock` rather than the whole checkpoint's messages up front. + streamingInbox: boolean = false, ): Promise { const { blockHeader, txHashes } = proposal; @@ -935,17 +1074,19 @@ export class ProposalHandler { gasFees: blockHeader.globalVariables.gasFees, }; - // Create checkpoint builder with prior blocks + // Create checkpoint builder with prior blocks. Under the streaming Inbox the checkpoint-wide message list is empty + // and this block's bundle is inserted per block (below); the legacy flow inserts the whole checkpoint up front. const checkpointBuilder = await this.checkpointsBuilder.openCheckpoint( checkpointNumber, constants, 0n, // only takes effect in the following checkpoint. - l1ToL2Messages, + streamingInbox ? [] : l1ToL2Messages, previousCheckpointOutHashes, previousInboxRollingHash, fork, priorBlocks, this.log.getBindings(), + streamingInbox, ); // Build the new block @@ -961,6 +1102,7 @@ export class ProposalHandler { expectedEndState: blockHeader.state, maxTransactions: this.config.validateMaxTxsPerBlock, maxBlockGas, + l1ToL2Messages: streamingInbox ? l1ToL2Messages : undefined, }); const { block, failedTxs } = result; @@ -1171,6 +1313,17 @@ export class ProposalHandler { const constants = this.extractCheckpointConstants(firstBlock); const checkpointNumber = firstBlock.checkpointNumber; + // Streaming Inbox: on the last block of a checkpoint, enforce the minimum-consumption + // (censorship) rule before attesting. Reject (no attestation) if a mandatory bucket was left unconsumed. Flag off, + // this is skipped and behavior is byte-identical. + if (this.config.streamingInbox === true && !(await this.isLastBlockConsumptionSufficient(slot, blocks))) { + this.log.warn(`Streaming Inbox last-block censorship check failed, refusing to attest`, { + ...proposalInfo, + checkpointNumber, + }); + return { isValid: false, reason: 'inbox_consumption_insufficient', checkpointNumber }; + } + // Get L1-to-L2 messages for this checkpoint const l1ToL2Messages = await this.l1ToL2MessageSource.getL1ToL2Messages(checkpointNumber); diff --git a/yarn-project/validator-client/src/streaming_inbox_checks.test.ts b/yarn-project/validator-client/src/streaming_inbox_checks.test.ts new file mode 100644 index 000000000000..1608f2bcd2b7 --- /dev/null +++ b/yarn-project/validator-client/src/streaming_inbox_checks.test.ts @@ -0,0 +1,286 @@ +import { Fr } from '@aztec/foundation/curves/bn254'; +import type { InboxBucket } from '@aztec/stdlib/messaging'; +import { InboxBucketRef } from '@aztec/stdlib/messaging'; + +import { describe, expect, it } from '@jest/globals'; + +import { + type StreamingBlockCheckInput, + type StreamingInboxBucketSource, + checkStreamingBlockProposal, +} from './streaming_inbox_checks.js'; + +const LAG_SECONDS = 12; +const PER_BLOCK_CAP = 1024; +const PER_CHECKPOINT_CAP = 1024; +const NOW = 10_000n; + +/** + * In-memory Inbox-bucket view mirroring the archiver store's index semantics: buckets keyed by sequence number, a flat + * leaves array indexed by global message index, and `getL1ToL2MessagesBetweenBuckets` slicing that array by the + * `(from, to]` bucket range (start = fromBucket.lastMessageIndex + 1, genesis when from == 0). + */ +class FakeInboxView implements StreamingInboxBucketSource { + private readonly buckets = new Map(); + private readonly leaves: Fr[] = []; + + constructor() { + // Genesis sentinel bucket 0 {total 0}, as the archiver store holds it (the "consumed nothing" base case). + this.buckets.set(0n, { + seq: 0n, + inboxRollingHash: Fr.ZERO, + totalMsgCount: 0n, + timestamp: 0n, + msgCount: 0, + lastMessageIndex: 0n, + }); + } + + /** Appends a bucket of `msgCount` leaves opened at `timestamp`, with a rolling hash derived from `seq`. */ + addBucket(seq: number, msgCount: number, timestamp: number, rollingHash?: Fr): InboxBucket { + const priorTotal = this.leaves.length; + for (let i = 0; i < msgCount; i++) { + this.leaves.push(new Fr(1000 + priorTotal + i)); + } + const totalMsgCount = BigInt(this.leaves.length); + const bucket: InboxBucket = { + seq: BigInt(seq), + inboxRollingHash: rollingHash ?? new Fr(500 + seq), + totalMsgCount, + timestamp: BigInt(timestamp), + msgCount, + lastMessageIndex: totalMsgCount - 1n, + }; + this.buckets.set(BigInt(seq), bucket); + return bucket; + } + + getInboxBucket(seq: bigint): Promise { + return Promise.resolve(this.buckets.get(seq)); + } + + getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise { + if (totalMsgCount === 0n) { + return Promise.resolve(this.buckets.get(0n)); + } + return Promise.resolve([...this.buckets.values()].find(b => b.totalMsgCount === totalMsgCount)); + } + + getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise { + const toBucket = this.buckets.get(toInclusive); + if (toBucket === undefined) { + return Promise.resolve([]); + } + let startIndex = 0n; + if (fromExclusive > 0n) { + const fromBucket = this.buckets.get(fromExclusive); + if (fromBucket === undefined) { + return Promise.resolve([]); + } + startIndex = fromBucket.lastMessageIndex + 1n; + } + return Promise.resolve(this.leaves.slice(Number(startIndex), Number(toBucket.lastMessageIndex + 1n))); + } +} + +function refFor(bucket: InboxBucket, rollingHash = bucket.inboxRollingHash): InboxBucketRef { + return new InboxBucketRef(bucket.seq, bucket.timestamp, rollingHash); +} + +function baseInput(overrides: Partial): StreamingBlockCheckInput { + return { + messageSource: new FakeInboxView(), + bucketRef: undefined, + parentTotalMsgCount: 0n, + checkpointStartTotalMsgCount: 0n, + nowSeconds: NOW, + lagSeconds: LAG_SECONDS, + perBlockCap: PER_BLOCK_CAP, + perCheckpointCap: PER_CHECKPOINT_CAP, + ...overrides, + }; +} + +describe('checkStreamingBlockProposal', () => { + describe('check 1: bucket exists and hash matches', () => { + it('rejects a proposal with no bucket reference', async () => { + const result = await checkStreamingBlockProposal(baseInput({ bucketRef: undefined })); + expect(result).toEqual({ accepted: false, reason: 'bucket_unknown' }); + }); + + it('rejects promptly when the referenced bucket is unknown (no waiting)', async () => { + const view = new FakeInboxView(); + const start = Date.now(); + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: new InboxBucketRef(7n, 100n, new Fr(1)) }), + ); + expect(result).toEqual({ accepted: false, reason: 'bucket_unknown' }); + // The happy path rejects immediately; there is no bounded wait yet. Assert it did not sleep. + expect(Date.now() - start).toBeLessThan(500); + }); + + it('rejects when the resolved bucket hash differs from the reference', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 3, 100); + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(bucket, new Fr(999)) }), + ); + expect(result).toEqual({ accepted: false, reason: 'bucket_hash_mismatch' }); + }); + }); + + describe('check 2: consumption moves forward', () => { + it('rejects when the bucket total is behind the parent block', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 3, 100); // total 3 + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(bucket), parentTotalMsgCount: 5n }), + ); + expect(result).toEqual({ accepted: false, reason: 'bucket_moves_backwards' }); + }); + + it('accepts an empty-consumption block that reuses the parent bucket (empty bundle)', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 3, 100); // total 3 + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(bucket), parentTotalMsgCount: 3n }), + ); + expect(result).toEqual({ accepted: true, bundle: [] }); + }); + }); + + describe('check 3: bucket is at least lagSeconds old', () => { + it('accepts a bucket exactly lagSeconds old (inclusive boundary)', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 2, Number(NOW) - LAG_SECONDS); // timestamp == now - lag + const result = await checkStreamingBlockProposal(baseInput({ messageSource: view, bucketRef: refFor(bucket) })); + expect(result.accepted).toBe(true); + }); + + it('rejects a bucket one second too new', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 2, Number(NOW) - LAG_SECONDS + 1); + const result = await checkStreamingBlockProposal(baseInput({ messageSource: view, bucketRef: refFor(bucket) })); + expect(result).toEqual({ accepted: false, reason: 'bucket_too_new' }); + }); + }); + + describe('check 4: caps', () => { + it('accepts a block consuming exactly the per-block cap', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, PER_BLOCK_CAP, 100); + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(bucket), perBlockCap: PER_BLOCK_CAP }), + ); + expect(result.accepted).toBe(true); + }); + + it('rejects a block consuming one over the per-block cap', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 4, 100); + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(bucket), perBlockCap: 3 }), + ); + expect(result).toEqual({ accepted: false, reason: 'bundle_over_block_cap' }); + }); + + it('rejects when the running checkpoint total exceeds the per-checkpoint cap', async () => { + const view = new FakeInboxView(); + view.addBucket(1, 3, 100); // total 3, the checkpoint's earlier consumption + const bucket = view.addBucket(2, 3, 100); // total 6 + const result = await checkStreamingBlockProposal( + baseInput({ + messageSource: view, + bucketRef: refFor(bucket), + parentTotalMsgCount: 3n, + checkpointStartTotalMsgCount: 0n, + perCheckpointCap: 5, // 6 - 0 = 6 > 5 + }), + ); + expect(result).toEqual({ accepted: false, reason: 'checkpoint_over_msg_cap' }); + }); + }); + + describe('bundle derivation', () => { + it('derives the bundle for a genesis-parent first block', async () => { + const view = new FakeInboxView(); + const bucket = view.addBucket(1, 3, 100); // leaves at global indices 0..2 + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(bucket), parentTotalMsgCount: 0n }), + ); + expect(result).toEqual({ accepted: true, bundle: [new Fr(1000), new Fr(1001), new Fr(1002)] }); + }); + + it('derives the bundle spanning multiple buckets since the parent', async () => { + const view = new FakeInboxView(); + view.addBucket(1, 2, 100); // parent consumed through here, total 2 + view.addBucket(2, 2, 100); // total 4 + const proposed = view.addBucket(3, 1, 100); // total 5 + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(proposed), parentTotalMsgCount: 2n }), + ); + // Bundle = leaves at global indices 2,3,4 (buckets 2 and 3), derived after resolving the parent bucket (seq 1). + expect(result).toEqual({ accepted: true, bundle: [new Fr(1002), new Fr(1003), new Fr(1004)] }); + }); + + it('rejects when the parent leaf count does not sit on a bucket boundary (padded legacy parent)', async () => { + const view = new FakeInboxView(); + view.addBucket(1, 2, 100); // total 2 + const proposed = view.addBucket(2, 2, 100); // total 4 + // Parent leaf count 3 is between bucket boundaries (2 and 4): unresolvable. + const result = await checkStreamingBlockProposal( + baseInput({ messageSource: view, bucketRef: refFor(proposed), parentTotalMsgCount: 3n }), + ); + expect(result).toEqual({ accepted: false, reason: 'parent_bucket_unresolved' }); + }); + }); + + describe('running-total accumulation across a checkpoint', () => { + it('accumulates the per-checkpoint total across three blocks against a fixed start', async () => { + // Checkpoint starts after bucket 1 (total 2). Three blocks each consume 2 messages: totals 4, 6, 8. + const view = new FakeInboxView(); + view.addBucket(1, 2, 100); // checkpoint start total 2 + const b2 = view.addBucket(2, 2, 100); // total 4 + const b3 = view.addBucket(3, 2, 100); // total 6 + const b4 = view.addBucket(4, 2, 100); // total 8 + const checkpointStart = 2n; + + // Block 1 (parent = bucket 1): checkpoint delta 4 - 2 = 2. + const r1 = await checkStreamingBlockProposal( + baseInput({ + messageSource: view, + bucketRef: refFor(b2), + parentTotalMsgCount: 2n, + checkpointStartTotalMsgCount: checkpointStart, + perCheckpointCap: 6, + }), + ); + expect(r1.accepted).toBe(true); + + // Block 3 (parent = bucket 3): checkpoint delta 8 - 2 = 6, exactly at the cap. + const r3 = await checkStreamingBlockProposal( + baseInput({ + messageSource: view, + bucketRef: refFor(b4), + parentTotalMsgCount: 6n, + checkpointStartTotalMsgCount: checkpointStart, + perCheckpointCap: 6, + }), + ); + expect(r3.accepted).toBe(true); + + // Same block against a tighter cap of 5: the accumulated delta 6 now exceeds it. + const r3Tight = await checkStreamingBlockProposal( + baseInput({ + messageSource: view, + bucketRef: refFor(b4), + parentTotalMsgCount: 6n, + checkpointStartTotalMsgCount: checkpointStart, + perCheckpointCap: 5, + }), + ); + expect(r3Tight).toEqual({ accepted: false, reason: 'checkpoint_over_msg_cap' }); + void b3; + }); + }); +}); diff --git a/yarn-project/validator-client/src/streaming_inbox_checks.ts b/yarn-project/validator-client/src/streaming_inbox_checks.ts new file mode 100644 index 000000000000..1859ead4d90d --- /dev/null +++ b/yarn-project/validator-client/src/streaming_inbox_checks.ts @@ -0,0 +1,132 @@ +import type { Fr } from '@aztec/foundation/curves/bn254'; +import type { InboxBucketRef, L1ToL2MessageSource } from '@aztec/stdlib/messaging'; + +/** + * Reason a streaming-Inbox block proposal fails the per-block acceptance checks. Follows the + * handler's existing `{ isValid, reason }` string style. + */ +export type StreamingBlockCheckReason = + | 'bucket_unknown' + | 'bucket_hash_mismatch' + | 'parent_bucket_unresolved' + | 'bucket_moves_backwards' + | 'bucket_too_new' + | 'bundle_over_block_cap' + | 'checkpoint_over_msg_cap'; + +/** The subset of the archiver's Inbox-bucket queries the per-block streaming checks need. */ +export type StreamingInboxBucketSource = Pick< + L1ToL2MessageSource, + 'getInboxBucket' | 'getInboxBucketByTotalMsgCount' | 'getL1ToL2MessagesBetweenBuckets' +>; + +/** Inputs to the per-block streaming Inbox acceptance checks. */ +export type StreamingBlockCheckInput = { + /** Archiver Inbox-bucket queries (resolved against this node's own Inbox view). */ + messageSource: StreamingInboxBucketSource; + /** The proposal's bucket reference: `bucketSeq` is the lookup hint, `inboxRollingHash` the expected commitment. */ + bucketRef: InboxBucketRef | undefined; + /** Cumulative Inbox message count consumed through the parent block (its L1-to-L2 tree leaf count; 0 at genesis). */ + parentTotalMsgCount: bigint; + /** Cumulative Inbox message count consumed as of the parent checkpoint; the per-checkpoint cap origin. */ + checkpointStartTotalMsgCount: bigint; + /** Validation-time wall clock in seconds; the lag-eligibility anchor. */ + nowSeconds: bigint; + /** Minimum bucket age in seconds (`INBOX_LAG_SECONDS`) for a bucket to be lag-eligible. */ + lagSeconds: number; + /** Maximum number of messages this block may consume (`MAX_L1_TO_L2_MSGS_PER_BLOCK`). */ + perBlockCap: number; + /** Maximum number of messages the checkpoint may consume in total (`MAX_L1_TO_L2_MSGS_PER_CHECKPOINT`). */ + perCheckpointCap: number; +}; + +/** Result of the per-block streaming Inbox acceptance checks. */ +export type StreamingBlockCheckResult = + | { + /** All four checks passed; `bundle` is the message-leaf bundle this block consumes, for re-execution. */ + accepted: true; + bundle: Fr[]; + } + | { + /** A check failed; `reason` mirrors the L1 acceptance condition that would have rejected the proposal. */ + accepted: false; + reason: StreamingBlockCheckReason; + }; + +/** + * Runs the per-block acceptance checks a validator applies to a streaming block proposal, and + * derives the message-leaf bundle the block consumes (for re-execution). 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. + * 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 `lagSeconds` old at validation time (`timestamp <= now - lagSeconds`, + * inclusive — a bucket exactly `lagSeconds` old is eligible, matching L1's strict `>` "too new" test). + * 4. **Caps**: the per-block message count and the running per-checkpoint total fit their respective caps. + * + * The reject branch is a single function so a future bounded wait can wrap `bucket_unknown`. + */ +export async function checkStreamingBlockProposal(input: StreamingBlockCheckInput): Promise { + const { + messageSource, + bucketRef, + parentTotalMsgCount, + checkpointStartTotalMsgCount, + nowSeconds, + lagSeconds, + perBlockCap, + perCheckpointCap, + } = input; + + // A streaming proposal must carry a bucket reference to derive its bundle from. + if (bucketRef === undefined) { + return { accepted: false, reason: 'bucket_unknown' }; + } + + // Check 1: exists in our own Inbox view, and the resolved rolling hash matches the reference. + const bucket = await messageSource.getInboxBucket(bucketRef.bucketSeq); + if (bucket === undefined) { + return { accepted: false, reason: 'bucket_unknown' }; + } + if (!bucket.inboxRollingHash.equals(bucketRef.inboxRollingHash)) { + return { accepted: false, reason: 'bucket_hash_mismatch' }; + } + + // Check 2: consumption moves forward relative to the parent block. + if (bucket.totalMsgCount < parentTotalMsgCount) { + return { accepted: false, reason: 'bucket_moves_backwards' }; + } + + // Check 3: the bucket is at least `lagSeconds` old at validation time. + if (bucket.timestamp > nowSeconds - BigInt(lagSeconds)) { + return { accepted: false, reason: 'bucket_too_new' }; + } + + // Check 4a: the per-block message count fits the per-block cap. + const blockCount = bucket.totalMsgCount - parentTotalMsgCount; + if (blockCount > BigInt(perBlockCap)) { + return { accepted: false, reason: 'bundle_over_block_cap' }; + } + + // Check 4b: the running per-checkpoint total fits the per-checkpoint cap. + const checkpointCount = bucket.totalMsgCount - checkpointStartTotalMsgCount; + if (checkpointCount > BigInt(perCheckpointCap)) { + return { accepted: false, reason: 'checkpoint_over_msg_cap' }; + } + + // Derive the message bundle for re-execution: the leaves consumed between the parent bucket and the proposed one. + // The parent bucket is the one whose cumulative total equals the parent block's leaf count (messages are indexed + // compactly, with no padding); a parent whose count does not sit on a bucket boundary is unresolvable. + const parentBucket = await messageSource.getInboxBucketByTotalMsgCount(parentTotalMsgCount); + if (parentBucket === undefined) { + return { accepted: false, reason: 'parent_bucket_unresolved' }; + } + const bundle = + parentBucket.seq === bucket.seq + ? [] + : await messageSource.getL1ToL2MessagesBetweenBuckets(parentBucket.seq, bucket.seq); + return { accepted: true, bundle }; +}