feat(fast-inbox): sequencer streaming message selection behind STREAMING_INBOX (A-1382)

Behind the shared `streamingInbox` flag (env `STREAMING_INBOX`, default off), the
sequencer selects L1-to-L2 messages from the Inbox rolling-hash buckets per block
instead of consuming the whole checkpoint's messages up front.

- Adds a pure `inbox_bucket_selector` that mirrors `ProposeLib.validateInboxConsumption`:
  newest lag-eligible bucket, per-block/per-checkpoint cap walk-back with cap-escape,
  and a last-block cutoff floor (cutoff = toTimestamp(slot-1) - INBOX_LAG_SECONDS).
- Adds `INBOX_LAG_SECONDS` (12) to constants.nr, regenerated into the TS constants.
- Threads a per-block message bundle through the checkpoint builder interface and an
  optional signed bucket reference onto block proposals.
- Flag off is byte-identical: the new paths are gated at checkpoint start and per block.

Author: spalladino

noir-projects/noir-protocol-circuits/crates/types/src/constants.nr

@@ -69,6 +69,11 @@ pub global L1_TO_L2_MSG_SUBTREE_ROOT_SIBLING_PATH_LENGTH: u32 =
 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
+// 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;
 // Maximum number of subtrees a L2ToL1Msg unbalanced tree can have. Used when calculating the out hash of a tx.
 pub global MAX_L2_TO_L1_MSG_SUBTREES_PER_TX: u32 = 3; // ceil(log2(MAX_L2_TO_L1_MSGS_PER_TX))
 

yarn-project/constants/src/constants.gen.ts

@@ -31,6 +31,7 @@ export const NULLIFIER_SUBTREE_ROOT_SIBLING_PATH_LENGTH = 36;
 export const L1_TO_L2_MSG_SUBTREE_ROOT_SIBLING_PATH_LENGTH = 26;
 export const MAX_L1_TO_L2_MSGS_PER_BLOCK = 1024;
 export const MAX_L1_TO_L2_MSGS_PER_CHECKPOINT = 1024;
+export const INBOX_LAG_SECONDS = 12;
 export const MAX_L2_TO_L1_MSG_SUBTREES_PER_TX = 3;
 export const MAX_NOTE_HASHES_PER_TX = 64;
 export const MAX_NULLIFIERS_PER_TX = 64;

yarn-project/foundation/src/config/env_var.ts

@@ -382,6 +382,7 @@ export type EnvVar =
   | 'FISHERMAN_MODE'
   | 'MAX_ALLOWED_ETH_CLIENT_DRIFT_SECONDS'
   | 'MAX_BLOCKS_PER_CHECKPOINT'
+  | 'STREAMING_INBOX'
   | 'LEGACY_BLS_CLI'
   | 'DEBUG_FORCE_TX_PROOF_VERIFICATION'
   | 'VALIDATOR_ALLOW_EPHEMERAL_SIGNING_PROTECTION'

yarn-project/prover-client/src/light/lightweight_checkpoint_builder.ts

@@ -70,9 +70,14 @@ export class LightweightCheckpointBuilder {
     db: MerkleTreeWriteOperations,
     bindings?: LoggerBindings,
     feeAssetPriceModifier: bigint = 0n,
+    // Streaming Inbox (AZIP-22 Fast Inbox): messages are inserted per block via `addBlock`, so `l1ToL2Messages` here
+    // is empty and the up-front checkpoint-wide insertion is skipped.
+    insertMessagesPerBlock: boolean = false,
   ): Promise<LightweightCheckpointBuilder> {
-    // Insert l1-to-l2 messages into the tree.
-    await appendL1ToL2MessagesToTree(db, l1ToL2Messages);
+    // Insert l1-to-l2 messages into the tree (legacy flow: the whole checkpoint's messages up front).
+    if (!insertMessagesPerBlock) {
+      await appendL1ToL2MessagesToTree(db, l1ToL2Messages);
+    }
 
     return new LightweightCheckpointBuilder(
       checkpointNumber,
@@ -172,7 +177,7 @@ export class LightweightCheckpointBuilder {
   public async addBlock(
     globalVariables: GlobalVariables,
     txs: ProcessedTx[],
-    opts: { insertTxsEffects?: boolean; expectedEndState?: StateReference } = {},
+    opts: { insertTxsEffects?: boolean; expectedEndState?: StateReference; l1ToL2Messages?: Fr[] } = {},
   ): Promise<{ block: L2Block; timings: Record<string, number> }> {
     const timings: Record<string, number> = {};
     const isFirstBlock = this.blocks.length === 0;
@@ -203,6 +208,20 @@ 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,
+    // 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 rolling hash and inHash are then recomputed over
+    // the accumulated logical (unpadded) messages at checkpoint completion.
+    if (opts.l1ToL2Messages !== undefined) {
+      if (isFirstBlock) {
+        await appendL1ToL2MessagesToTree(this.db, opts.l1ToL2Messages);
+      } else {
+        await this.db.appendLeaves(MerkleTreeId.L1_TO_L2_MESSAGE_TREE, opts.l1ToL2Messages);
+      }
+      this.l1ToL2Messages.push(...opts.l1ToL2Messages);
+    }
+
     const [msGetEndState, endState] = await elapsed(() => this.db.getStateReference());
     timings.getEndState = msGetEndState;
 

yarn-project/sequencer-client/src/config.ts

@@ -68,6 +68,7 @@ export const DefaultSequencerConfig = {
   skipPushProposedBlocksToArchiver: false,
   skipPublishingCheckpointsPercent: 0,
   maxBlocksPerCheckpoint: DEFAULT_MAX_BLOCKS_PER_CHECKPOINT,
+  streamingInbox: false,
 } satisfies ResolvedSequencerConfig;
 
 /**

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts

@@ -40,7 +40,7 @@ import {
   type ResolvedSequencerConfig,
   type WorldStateSynchronizer,
 } from '@aztec/stdlib/interfaces/server';
-import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging';
+import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging';
 import { BlockProposal, CheckpointProposal, type CoordinationSignatureContext } from '@aztec/stdlib/p2p';
 import { CheckpointHeader } from '@aztec/stdlib/rollup';
 import type { ProposerTimetable, SubslotSelection } from '@aztec/stdlib/timetable';
@@ -1394,6 +1394,60 @@ describe('CheckpointProposalJob', () => {
     });
   });
 
+  describe('streaming inbox (flag on)', () => {
+    beforeEach(() => {
+      job.setTimetable(makeProposerTimetable({ l1Constants, blockDurationMs: 3000 }));
+      job.updateConfig({ streamingInbox: true });
+    });
+
+    it('streams per-block bucket bundles, advances the reference, and skips the bulk message fetch', async () => {
+      jest
+        .spyOn(job.getTimetable(), 'selectNextSubslot')
+        .mockReturnValueOnce(subslot(10, 0, false))
+        .mockReturnValueOnce(subslot(18, 1, true))
+        .mockReturnValue(noSubslot());
+
+      // The archiver returns the newest synced bucket (seq 2) for any lag/cutoff lookup, so the first block
+      // consumes through it and the second block (cursor already at seq 2) consumes nothing and reuses the ref.
+      const bucket: InboxBucket = {
+        seq: 2n,
+        inboxRollingHash: new Fr(99),
+        totalMsgCount: 5n,
+        timestamp: 0n,
+        msgCount: 2,
+        lastMessageIndex: 4n,
+        isOpen: false,
+      };
+      const bundle = Array.from({ length: 5 }, (_, i) => new Fr(i + 1));
+      l1ToL2MessageSource.getLatestInboxBucketAtOrBefore.mockResolvedValue(bucket);
+      l1ToL2MessageSource.getL1ToL2MessagesBetweenBuckets.mockResolvedValue(bundle);
+
+      const { lastBlock } = await setupMultipleBlocks(2, [2, 1]);
+      validatorClient.collectAttestations.mockResolvedValue(getAttestations(lastBlock));
+
+      const checkpoint = await job.executeAndAwait();
+
+      expect(checkpoint).toBeDefined();
+
+      // Streaming skips the bulk per-checkpoint fetch and tells the builder to insert messages per block.
+      expect(l1ToL2MessageSource.getL1ToL2Messages).not.toHaveBeenCalled();
+      expect(checkpointsBuilder.startCheckpointCalls).toHaveLength(1);
+      expect(checkpointsBuilder.startCheckpointCalls[0].l1ToL2Messages).toEqual([]);
+      expect(checkpointsBuilder.startCheckpointCalls[0].insertMessagesPerBlock).toBe(true);
+
+      // The first block consumes the selected bundle; the second consumes nothing.
+      expect(checkpointBuilder.buildBlockCalls).toHaveLength(2);
+      expect(checkpointBuilder.buildBlockCalls[0].opts.l1ToL2Messages).toEqual(bundle);
+      expect(checkpointBuilder.buildBlockCalls[1].opts.l1ToL2Messages).toEqual([]);
+
+      // Both block proposals carry the selected bucket reference (the second reuses the first's).
+      const bucketRefArgs = validatorClient.createBlockProposal.mock.calls.map(call => call[8]);
+      expect(bucketRefArgs).toHaveLength(2);
+      expect(bucketRefArgs[0]?.bucketSeq).toBe(2n);
+      expect(bucketRefArgs[1]?.bucketSeq).toBe(2n);
+    });
+  });
+
   describe('build single block', () => {
     it('does not build a block if not enough valid txs are collected', async () => {
       // We have enough txs, but not enough valid ones