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

yarn-project/archiver/src/modules/data_source_base.ts

@@ -332,6 +332,10 @@ 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);
   }

yarn-project/archiver/src/store/message_store.test.ts

@@ -389,6 +389,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.addL1ToL2Messages(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.addL1ToL2Messages(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('continues a bucket that spans two insertion batches', async () => {
       const msgs = makeBucketedMessages(threeBucketSpec);
       await messageStore.addL1ToL2Messages(msgs.slice(0, 2));

yarn-project/archiver/src/store/message_store.ts

@@ -54,6 +54,20 @@ function deserializeBucketSnapshot(buffer: Buffer): BucketSnapshot {
   return { inboxRollingHash, totalMsgCount, timestamp, msgCount, lastMessageIndex };
 }
 
+// The genesis sentinel bucket (AZIP-22 Fast Inbox): 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,
+  isOpen: false,
+};
+
 export class MessageStoreError extends Error {
   constructor(
     message: string,
@@ -403,11 +417,35 @@ export class MessageStore {
 
   /**
    * Returns the Inbox bucket with the given sequence number, or undefined if it has not been synced (AZIP-22 Fast
-   * Inbox).
+   * Inbox). 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 (AZIP-22 Fast Inbox). 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 (e.g. a pre-flip padded leaf
+   * count) 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;
   }
 
   /**

yarn-project/archiver/src/test/mock_archiver.ts

@@ -37,6 +37,10 @@ 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);
   }

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)

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,
@@ -1127,9 +1132,7 @@ export class CheckpointProposalJob implements Traceable {
     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/stdlib/src/interfaces/archiver.test.ts

@@ -224,6 +224,11 @@ describe('ArchiverApiSchema', () => {
     expect(result).toMatchObject({ seq: 2n, msgCount: 3, isOpen: true });
   });
 
+  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)]);
@@ -591,6 +596,18 @@ class MockArchiver implements ArchiverApi {
       isOpen: true,
     });
   }
+  getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise<InboxBucket | undefined> {
+    expect(typeof totalMsgCount).toEqual('bigint');
+    return Promise.resolve({
+      seq: 2n,
+      inboxRollingHash: Fr.random(),
+      totalMsgCount,
+      timestamp: 100n,
+      msgCount: 3,
+      lastMessageIndex: 2n,
+      isOpen: true,
+    });
+  }
   getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
     expect(typeof fromExclusive).toEqual('bigint');
     expect(typeof toInclusive).toEqual('bigint');

yarn-project/stdlib/src/interfaces/archiver.ts

@@ -135,6 +135,10 @@ export const ArchiverApiSchema: ApiSchemaFor<ArchiverApi> = {
     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),

yarn-project/stdlib/src/interfaces/validator.ts

@@ -83,7 +83,10 @@ export type ValidatorClientConfig = ValidatorHASignerConfig &
   };
 
 export type ValidatorClientFullConfig = ValidatorClientConfig &
-  Pick<SequencerConfig, 'txPublicSetupAllowListExtend' | 'broadcastInvalidBlockProposal' | 'maxBlocksPerCheckpoint'> &
+  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<Pick<SequencerConfig, 'blockDurationMs'>> &
@@ -134,6 +137,7 @@ export const ValidatorClientFullConfigSchema = zodFor<Omit<ValidatorClientFullCo
     broadcastInvalidBlockProposal: z.boolean().optional(),
     blockDurationMs: z.number().positive(),
     maxBlocksPerCheckpoint: z.number().positive().optional(),
+    streamingInbox: z.boolean().optional(),
     slashBroadcastedInvalidBlockPenalty: schemas.BigInt,
     slashBroadcastedInvalidCheckpointProposalPenalty: schemas.BigInt,
     slashDuplicateProposalPenalty: schemas.BigInt,

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 pinned by A-1371 decision 13 (the `ProposeInboxConsumptionTest` Foundry harness values).
+// 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. Pinned in A-1371 §13.
+    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);
+    });
+
+    // A-1371 §13 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,
+      );
+    });
+
+    // A-1371 §13 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
+    });
+  });
+});

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` (AZIP-22 Fast Inbox), 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<L1RollupConstants, 'l1GenesisTime' | 'slotDuration'>,
+  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` (AZIP-22 Fast Inbox).
+ * 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<InboxBucket, 'timestamp' | 'totalMsgCount'> | 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);
+}

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

yarn-project/stdlib/src/messaging/l1_to_l2_message_source.ts

@@ -39,6 +39,17 @@ export interface L1ToL2MessageSource {
    */
   getInboxBucket(seq: bigint): Promise<InboxBucket | undefined>;
 
+  /**
+   * Returns the Inbox bucket whose cumulative message total equals `totalMsgCount`, or undefined if no synced bucket
+   * sits on that boundary (AZIP-22 Fast Inbox). A block's or checkpoint's L1-to-L2 tree leaf count equals the
+   * cumulative total of the last bucket it consumed (post-flip compact indexing), 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 (e.g. a pre-flip
+   * padded leaf count) returns undefined.
+   * @param totalMsgCount - The cumulative Inbox message count (leaf count) to resolve to a bucket boundary.
+   */
+  getInboxBucketByTotalMsgCount(totalMsgCount: bigint): Promise<InboxBucket | undefined>;
+
   /**
    * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion
    * order, for streaming message-bundle derivation (AZIP-22 Fast Inbox). Returns an empty array if the upper