feat(fast-inbox): validator streaming acceptance conditions behind STREAMING_INBOX (A-1383)

Behind the shared `streamingInbox` flag (default off), the validator applies the AZIP-22
Fast Inbox acceptance conditions in place of the legacy per-checkpoint inHash comparison.
Flag off, behavior is byte-identical.

Block proposals (`proposal_handler.handleBlockProposal`), via a new pure
`streaming_inbox_checks` module:
- exists: the referenced bucket resolves in the node's own Inbox view and its rolling
  hash matches the reference (unknown bucket is an immediate reject; the bounded wait is
  A-1393);
- moves forward: the bucket's cumulative total is at least the parent block's;
- not too new: the bucket is at least INBOX_LAG_SECONDS old at validation time
  (dateProvider.now, inclusive boundary);
- caps: the per-block count and the running per-checkpoint total fit their caps.
The derived bundle is fed into per-block re-execution (insertMessagesPerBlock threaded
through openCheckpoint/reexecuteTransactions), whose state-ref comparison stays the final
arbiter.

Checkpoint proposals (`validateCheckpointProposal`): the last-block minimum-consumption
(censorship) rule runs before attesting, via the shared `isInboxConsumptionSufficient`
predicate and the corrected cutoff anchor `getTimestampForSlot(slot - 1) -
INBOX_LAG_SECONDS`; a mandatory unconsumed bucket rejects the checkpoint (no attestation).

Parent/last-consumed buckets are resolved from L1-to-L2 tree leaf counts via a new
`getInboxBucketByTotalMsgCount` archiver lookup (compact post-flip indexing); a leaf count
that does not land on a bucket boundary (a pre-flip padded parent) is rejected/deferred,
the flip (A-1384) closes that gap.

The flag is picked up on the validator config via pickConfigMappings of the shared
`streamingInbox` key. New reasons are non-slashable while the path lands. Full tsc -b
cannot run locally (pre-existing noir-protocol-circuits-types breakage); CI is the
typecheck of record for validator-client. Unit tests cover each check, the caps and lag
boundaries, running-total accumulation, the censorship predicate, and the handler wiring.

Author: spalladino

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,20 @@ 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('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

@@ -410,6 +410,25 @@ export class MessageStore {
     return snapshot && this.toInboxBucket(seq, snapshot);
   }
 
+  /**
+   * 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;
+  }
+
   /**
    * 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).

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/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/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

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<CheckpointBuilder> {
     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,
       );
     }