feat(fast-inbox): validator streaming acceptance conditions, flag off (A-1383)

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;

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(

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<InboxBucket | undefined> {
+    return this.stores.messages.getInboxBucketByTotalMsgCount(totalMsgCount);
+  }
+
   public getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
     return this.stores.messages.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
   }
 
+  public getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise<Fr[]> {
+    return this.stores.messages.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount);
+  }
+
   private async getPublishedCheckpointFromCheckpointData(checkpoint: CheckpointData): Promise<PublishedCheckpoint> {
     const blocksForCheckpoint = await this.stores.blocks.getBlocksForCheckpoint(checkpoint.checkpointNumber);
     if (!blocksForCheckpoint) {

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);

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<InboxBucket | undefined> {
     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<InboxBucket | undefined> {
+    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<Fr[]> {
+    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<InboxBucket> {
+    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<InboxBucket | undefined> {
     // 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.

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;

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;

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<InboxBucket | undefined> {
+    return this.messageSource.getInboxBucketByTotalMsgCount(totalMsgCount);
+  }
+
   getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
     return this.messageSource.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
   }
+
+  getL1ToL2MessagesBetweenLeafCounts(startLeafCount: bigint, endLeafCount: bigint): Promise<Fr[]> {
+    return this.messageSource.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount);
+  }
 }
 
 /**

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<InboxBucket | undefined> {
+    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<InboxBucket | undefined> {
     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<Fr[]> {
+    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));
   }

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

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,17 +1124,15 @@ 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(
     state: StreamingCheckpointState,
     isLastBlock: boolean,
     nowSeconds: number,
   ): Promise<InboxBucketSelection> {
-    // 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)),

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;