feat(fast-inbox): archiver syncs Inbox buckets (A-1379)

Part of the Fast Inbox stack (AZIP-22). Makes the archiver track the L1 Inbox rolling-hash buckets introduced in A-1377, validate the full-width consensus rolling-hash chain, and expose the bucket queries the sequencer/validator will consume (A-1382/A-1383). Purely additive: the legacy per-checkpoint `getL1ToL2Messages` path is unchanged.

## What's added

- `InboxMessage` gains `inboxRollingHash: Fr`, `bucketSeq`, and `bucketTimestamp` (the L1 block timestamp that keys the bucket). `mapLogInboxMessage` propagates them; the ethereum `MessageSent` log now also carries the emitting L1 block's timestamp.
- `MessageStore` persists a per-bucket snapshot (`{ inboxRollingHash, totalMsgCount, timestamp, msgCount, lastMessageIndex }`) keyed by bucket sequence, plus a timestamp→sequence index for at-or-before lookups. Snapshots are written as messages are inserted and rewound (deleted / boundary-bucket recomputed) on reorg removal, all inside the existing kv-store transactions.
- `addL1ToL2Messages` now validates the full-width consensus rolling hash (`updateInboxRollingHash`) alongside the legacy 128-bit keccak chain. Both run until the streaming inbox flips on (A-1388 removes the legacy one).
- Three queries on `L1ToL2MessageSource` (implemented by the archiver, supported by the mock, exposed over the archiver RPC schema): `getLatestInboxBucketAtOrBefore(timestamp)`, `getInboxBucket(seq)`, and `getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive)`. New `InboxBucket` type + zod schema in stdlib messaging.

## Notes

- No migration: `ARCHIVER_DB_VERSION` is bumped 7 → 8, so nodes resync their L1→L2 messages from L1. Archiver message stores are rebuildable from L1, so this is acceptable on this release line.
- Folds in A-1344: `InboxMessage.l1BlockNumber` is widened from UInt32 to UInt64 (the serialization was being rewritten anyway).
- Field naming: the full-width hash is `inboxRollingHash: Fr` to match the `updateInboxRollingHash` mirror, `CheckpointHeader.inboxRollingHash`, and the decoded `MessageSent` event — the value is a truncated-sha256-to-field element. (The plan's provisional `consensusRollingHash: Buffer32` predates the implemented event.) The legacy `rollingHash: Buffer16` stays until A-1388.

## Out of scope

- Reorg-driven bucket invalidation (a reorg that replays the identical message leaves but re-buckets them across different L1 blocks changes neither the 128-bit nor the full-width leaf-based hash, so `localStateMatches` cannot detect it, and the allowed message-reinsertion path leaves its bucket snapshot stale). This is handled by A-1389, and bucket consumption is gated off by the `streamingInbox` flag pre-flip.
- Sequencer/validator consumption (A-1382/A-1383), proposal bucket-reference types (A-1381), and removing the legacy 128-bit hash (A-1388).

## Testing

- `message_store.test.ts`: bucket snapshotting (multi-message, cross-batch, rollover), full-width chain-mismatch detection, reorg rewinding, and the three queries incl. boundary cases.
- `archiver-sync.test.ts`: end-to-end sync now exercises the full-width chain validation and asserts the bucket queries against the synced state (the fake L1 state mirrors the on-chain bucket assignment).
- `stdlib` archiver interface RPC round-trip tests for the three new methods.


Replaces #24784.

Author: spalladino

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

@@ -259,6 +259,34 @@ describe('Archiver Sync', () => {
       expect(
         (await archiver.getCheckpoints({ from: CheckpointNumber(1), limit: 100 })).map(b => b.checkpoint.number),
       ).toEqual([1, 2, 3]);
+
+      // Inbox buckets: each of the three L1 message blocks opened its own bucket, in insertion order.
+      const t1 = fake.getTimestampAtL1Block(98n);
+      const t2 = fake.getTimestampAtL1Block(2504n);
+      const t3 = fake.getTimestampAtL1Block(2511n);
+
+      expect(await archiver.getInboxBucket(1n)).toMatchObject({
+        seq: 1n,
+        msgCount: 3,
+        totalMsgCount: 3n,
+        timestamp: t1,
+      });
+      expect(await archiver.getInboxBucket(3n)).toMatchObject({
+        seq: 3n,
+        msgCount: 3,
+        totalMsgCount: 9n,
+        timestamp: t3,
+      });
+
+      // At-or-before lookups resolve the latest bucket not opened after the given timestamp.
+      expect((await archiver.getLatestInboxBucketAtOrBefore(t3))!.seq).toEqual(3n);
+      expect((await archiver.getLatestInboxBucketAtOrBefore(t2))!.seq).toEqual(2n);
+      expect(await archiver.getLatestInboxBucketAtOrBefore(t1 - 1n)).toBeUndefined();
+
+      // Messages between buckets, in insertion order.
+      expect(await archiver.getL1ToL2MessagesBetweenBuckets(0n, 3n)).toEqual([...msgs1, ...msgs2, ...msgs3]);
+      expect(await archiver.getL1ToL2MessagesBetweenBuckets(1n, 2n)).toEqual(msgs2);
+      expect(await archiver.getL1ToL2MessagesBetweenBuckets(2n, 3n)).toEqual(msgs3);
     }, 30_000);
 
     it('ignores checkpoint 3 because it has been pruned', async () => {

yarn-project/archiver/src/errors.ts

@@ -116,6 +116,17 @@ export class L1ToL2MessagesNotReadyError extends Error {
   }
 }
 
+/**
+ * Thrown when a query names an Inbox bucket this archiver has not synced. Distinguishes "not synced yet, retry once
+ * L1 sync catches up" from a genuinely empty result.
+ */
+export class InboxBucketNotSyncedError extends Error {
+  constructor(public readonly bucketSeq: bigint) {
+    super(`Inbox bucket ${bucketSeq} has not been synced`);
+    this.name = 'InboxBucketNotSyncedError';
+  }
+}
+
 /** Thrown when a proposed checkpoint number is stale (already processed). */
 export class ProposedCheckpointStaleError extends Error {
   constructor(

yarn-project/archiver/src/l1/data_retrieval.ts

@@ -387,6 +387,9 @@ function mapLogInboxMessage(log: MessageSentLog): InboxMessage {
     l1BlockHash: log.l1BlockHash,
     checkpointNumber: log.args.checkpointNumber,
     rollingHash: log.args.rollingHash,
+    inboxRollingHash: log.args.inboxRollingHash,
+    bucketSeq: log.args.bucketSeq,
+    bucketTimestamp: log.l1BlockTimestamp,
   };
 }
 

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

@@ -40,7 +40,7 @@ import {
 } from '@aztec/stdlib/epoch-helpers';
 import type { L2LogsSource } from '@aztec/stdlib/interfaces/server';
 import type { LogResult, PrivateLogsQuery, PublicLogsQuery } from '@aztec/stdlib/logs';
-import type { L1ToL2MessageSource, L2ToL1MembershipWitness } from '@aztec/stdlib/messaging';
+import type { InboxBucket, L1ToL2MessageSource, L2ToL1MembershipWitness } from '@aztec/stdlib/messaging';
 import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees';
 import type { BlockHeader, IndexedTxEffect, TxHash } from '@aztec/stdlib/tx';
 import type { UInt64 } from '@aztec/stdlib/types';
@@ -324,6 +324,18 @@ export abstract class ArchiverDataSourceBase
     return this.stores.messages.getL1ToL2MessageIndex(l1ToL2Message);
   }
 
+  public getLatestInboxBucketAtOrBefore(timestamp: bigint): Promise<InboxBucket | undefined> {
+    return this.stores.messages.getLatestInboxBucketAtOrBefore(timestamp);
+  }
+
+  public getInboxBucket(seq: bigint): Promise<InboxBucket | undefined> {
+    return this.stores.messages.getInboxBucket(seq);
+  }
+
+  public getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
+    return this.stores.messages.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
+  }
+
   private async getPublishedCheckpointFromCheckpointData(checkpoint: CheckpointData): Promise<PublishedCheckpoint> {
     const blocksForCheckpoint = await this.stores.blocks.getBlocksForCheckpoint(checkpoint.checkpointNumber);
     if (!blocksForCheckpoint) {

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

@@ -495,7 +495,10 @@ export class ArchiverL1Synchronizer implements Traceable {
     );
   }
 
-  /** Retrieves L1 to L2 messages from L1 in batches and stores them. */
+  /**
+   * Retrieves L1 to L2 messages from L1 in batches and stores them. Batches must span whole L1 blocks so that every
+   * message of an Inbox bucket is stored in a single call, which the message store requires.
+   */
   private async retrieveAndStoreMessages(fromL1Block: bigint, toL1Block: bigint): Promise<void> {
     let searchStartBlock: bigint = 0n;
     let searchEndBlock: bigint = fromL1Block;
@@ -508,7 +511,7 @@ export class ArchiverL1Synchronizer implements Traceable {
       this.log.trace(`Retrieving L1 to L2 messages in L1 blocks ${searchStartBlock}-${searchEndBlock}`);
       const messages = await retrieveL1ToL2Messages(this.inbox, searchStartBlock, searchEndBlock);
       const timer = new Timer();
-      await this.stores.messages.addL1ToL2Messages(messages);
+      await this.stores.messages.addL1ToL2MessageBuckets(messages);
       const perMsg = timer.ms() / messages.length;
       this.instrumentation.processNewMessages(messages.length, perMsg);
       for (const msg of messages) {

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

@@ -13,7 +13,7 @@ import { FunctionNamesCache } from './function_names_cache.js';
 import { LogStore } from './log_store.js';
 import { MessageStore } from './message_store.js';
 
-export const ARCHIVER_DB_VERSION = 7;
+export const ARCHIVER_DB_VERSION = 8;
 
 /**
  * Represents the latest L1 block processed by the archiver for various objects in L2.