refactor(fast-inbox): store inbox buckets as complete units in the archiver (A-1379)

The message store used to snapshot a bucket incrementally, once per absorbed
message, threading the in-progress bucket state across insertion batches. Every
bucket is opened and closed within a single L1 block and the synchronizer
retrieves Inbox logs in whole-L1-block ranges, so a bucket's messages always
arrive together: group the batch by bucket sequence and derive one snapshot per
bucket from its complete message set instead.

The completeness that makes this correct is now enforced rather than assumed.
A batch that continues a stored bucket from the middle, or that opens a bucket
older than the newest stored one, is rejected: either would produce a snapshot
disagreeing with the messages it covers. Re-delivering a stored bucket from its
first message stays allowed, since an L1 reorg replaces a bucket's tail and the
re-sync that follows replays the whole L1 block.

Renames `addL1ToL2Messages` to `addL1ToL2MessageBuckets` to say what the method
now requires of its caller, drops `isOpen` from `InboxBucket` (it had no
consumers outside the store that produced it), and records each bucket's L1
block number and first message index in the snapshot for rollbacks and range
queries to work off bucket records alone.

Author: spalladino

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

@@ -276,9 +276,7 @@ describe('Archiver Sync', () => {
         msgCount: 3,
         totalMsgCount: 9n,
         timestamp: t3,
-        isOpen: true,
       });
-      expect((await archiver.getInboxBucket(1n))!.isOpen).toBe(false);
 
       // At-or-before lookups resolve the latest bucket not opened after the given timestamp.
       expect((await archiver.getLatestInboxBucketAtOrBefore(t3))!.seq).toEqual(3n);

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/message_store.test.ts

@@ -23,14 +23,17 @@ import {
 } from '../structs/inbox_message.js';
 
 /**
- * Persisted snapshot of an Inbox rolling-hash bucket. Mirrors the fields the on-chain Inbox tracks per bucket,
- * plus the last absorbed message index so the between-buckets query can range-scan messages directly.
+ * Persisted snapshot of an Inbox rolling-hash bucket. Mirrors the fields the on-chain Inbox tracks per bucket, plus
+ * the L1 block the bucket was opened in and the index span of its messages, so rollbacks and range queries can work
+ * off bucket records alone without scanning messages.
  */
 type BucketSnapshot = {
   inboxRollingHash: Fr;
   totalMsgCount: bigint;
   timestamp: bigint;
+  l1BlockNumber: bigint;
   msgCount: number;
+  firstMessageIndex: bigint;
   lastMessageIndex: bigint;
 };
 
@@ -39,7 +42,9 @@ function serializeBucketSnapshot(snapshot: BucketSnapshot): Buffer {
     snapshot.inboxRollingHash,
     bigintToUInt64BE(snapshot.totalMsgCount),
     bigintToUInt64BE(snapshot.timestamp),
+    bigintToUInt64BE(snapshot.l1BlockNumber),
     numToUInt32BE(snapshot.msgCount),
+    bigintToUInt64BE(snapshot.firstMessageIndex),
     bigintToUInt64BE(snapshot.lastMessageIndex),
   ]);
 }
@@ -49,9 +54,34 @@ function deserializeBucketSnapshot(buffer: Buffer): BucketSnapshot {
   const inboxRollingHash = reader.readObject(Fr);
   const totalMsgCount = reader.readUInt64();
   const timestamp = reader.readUInt64();
+  const l1BlockNumber = reader.readUInt64();
   const msgCount = reader.readNumber();
+  const firstMessageIndex = reader.readUInt64();
   const lastMessageIndex = reader.readUInt64();
-  return { inboxRollingHash, totalMsgCount, timestamp, msgCount, lastMessageIndex };
+  return { inboxRollingHash, totalMsgCount, timestamp, l1BlockNumber, msgCount, firstMessageIndex, lastMessageIndex };
+}
+
+/** The messages of a single Inbox bucket within an incoming batch, in insertion order. */
+type IncomingBucket = {
+  seq: bigint;
+  messages: InboxMessage[];
+};
+
+/**
+ * Splits an incoming batch of messages into per-bucket groups, in delivery order. Messages arrive ordered by index
+ * and a bucket's messages are contiguous within that order, so a group ends as soon as the bucket sequence changes.
+ */
+function groupMessagesByBucket(messages: InboxMessage[]): IncomingBucket[] {
+  const buckets: IncomingBucket[] = [];
+  for (const message of messages) {
+    const current = buckets.at(-1);
+    if (current !== undefined && current.seq === message.bucketSeq) {
+      current.messages.push(message);
+    } else {
+      buckets.push({ seq: message.bucketSeq, messages: [message] });
+    }
+  }
+  return buckets;
 }
 
 export class MessageStoreError extends Error {
@@ -137,11 +167,20 @@ export class MessageStore {
   }
 
   /**
-   * Append L1 to L2 messages to the store.
-   * Requires new messages to be in order and strictly after the last message added.
-   * Throws if out of order messages are added or if the rolling hash is invalid.
+   * Appends L1 to L2 messages to the store, one whole Inbox bucket at a time.
+   *
+   * A bucket is opened and closed within a single L1 block, and callers retrieve Inbox logs in whole-L1-block ranges,
+   * so every message of a bucket reaches this method in the same call — including the rollover buckets a full block
+   * spills into. The bucket snapshots are derived from that: the batch is split per bucket sequence and each bucket
+   * gets a single snapshot built from its complete message set. Delivering only part of a bucket already held in the
+   * store is rejected, since the snapshot would then undercount the bucket. Delivering a stored bucket again from its
+   * first message is allowed: an L1 reorg can replace a bucket's tail, and the re-sync that follows replays the whole
+   * L1 block it lives in.
+   *
+   * Requires messages to be ordered by index and to continue the stored chain. Throws a `MessageStoreError` if
+   * messages arrive out of order, if the rolling hash chain breaks, or if a bucket arrives incomplete.
    */
-  public addL1ToL2Messages(messages: InboxMessage[]): Promise<void> {
+  public addL1ToL2MessageBuckets(messages: InboxMessage[]): Promise<void> {
     if (messages.length === 0) {
       return Promise.resolve();
     }
@@ -150,13 +189,8 @@ export class MessageStore {
       let lastMessage = await this.getLastMessage();
       let messageCount = 0;
 
-      // Running cumulative message count and in-progress bucket state, threaded across the batch so we can snapshot
-      // each Inbox bucket as its messages are inserted. Seeded from the last stored message so a bucket that spans
-      // two batches keeps accumulating.
-      let cumulativeTotal = await this.getTotalL1ToL2MessageCount();
-      let currentBucketSeq: bigint | undefined = lastMessage?.bucketSeq;
-      let currentBucketMsgCount =
-        currentBucketSeq !== undefined ? ((await this.getBucketSnapshotBySeq(currentBucketSeq))?.msgCount ?? 0) : 0;
+      const incomingBuckets = groupMessagesByBucket(messages);
+      await this.assertIncomingBucketsAreComplete(incomingBuckets);
 
       for (const message of messages) {
         // Check messages are inserted in increasing order, but allow reinserting messages.
@@ -241,31 +275,74 @@ export class MessageStore {
         await this.#l1ToL2MessageIndices.set(this.leafToIndexKey(message.leaf), message.index);
         messageCount++;
 
-        // Snapshot the bucket this message was absorbed into. A message opens a new bucket whenever its bucket
-        // sequence differs from the one currently being accumulated; otherwise it extends the current bucket.
-        cumulativeTotal += 1n;
-        if (currentBucketSeq === undefined || message.bucketSeq !== currentBucketSeq) {
-          currentBucketSeq = message.bucketSeq;
-          currentBucketMsgCount = 0;
-        }
-        currentBucketMsgCount += 1;
-        await this.writeBucketSnapshot(message.bucketSeq, {
-          inboxRollingHash: message.inboxRollingHash,
-          totalMsgCount: cumulativeTotal,
-          timestamp: message.bucketTimestamp,
-          msgCount: currentBucketMsgCount,
-          lastMessageIndex: message.index,
-        });
-
         this.#log.trace(`Inserted L1 to L2 message ${message.leaf} with index ${message.index} into the store`);
         lastMessage = message;
       }
 
+      await this.writeIncomingBucketSnapshots(incomingBuckets);
+
       // Update total message count with the number of inserted messages.
       await this.increaseTotalMessageCount(messageCount);
     });
   }
 
+  /**
+   * Rejects a batch that delivers an Inbox bucket the store already holds without replaying it from its first message,
+   * or that opens a bucket older than the newest one stored. Either would produce a snapshot that disagrees with the
+   * messages it covers, since each snapshot is derived from the batch's messages for that bucket alone.
+   */
+  private async assertIncomingBucketsAreComplete(incomingBuckets: IncomingBucket[]): Promise<void> {
+    const newestStoredSeq = await this.getNewestBucketSeq();
+    let previousSeq: bigint | undefined;
+    for (const bucket of incomingBuckets) {
+      if (previousSeq !== undefined && bucket.seq <= previousSeq) {
+        throw new MessageStoreError(
+          `Inbox bucket ${bucket.seq} arrives after bucket ${previousSeq} in the same batch`,
+          bucket.messages[0],
+        );
+      }
+      previousSeq = bucket.seq;
+
+      const stored = await this.getBucketSnapshotBySeq(bucket.seq);
+      if (stored === undefined) {
+        if (newestStoredSeq !== undefined && bucket.seq <= newestStoredSeq) {
+          throw new MessageStoreError(
+            `Cannot open Inbox bucket ${bucket.seq} after bucket ${newestStoredSeq} has been stored`,
+            bucket.messages[0],
+          );
+        }
+      } else if (stored.firstMessageIndex !== bucket.messages[0].index) {
+        throw new MessageStoreError(
+          `Incomplete Inbox bucket ${bucket.seq}: stored messages start at index ${stored.firstMessageIndex} ` +
+            `but the batch starts at index ${bucket.messages[0].index}`,
+          bucket.messages[0],
+        );
+      }
+    }
+  }
+
+  /**
+   * Writes one snapshot per bucket in the batch, each derived from the bucket's complete message set. Cumulative
+   * totals thread forward from the bucket preceding the batch, so a bucket re-delivered with extra messages shifts
+   * the totals of the buckets after it within the same batch.
+   */
+  private async writeIncomingBucketSnapshots(incomingBuckets: IncomingBucket[]): Promise<void> {
+    let cumulativeTotal = await this.getTotalMsgCountBeforeBucket(incomingBuckets[0].seq);
+    for (const { seq, messages } of incomingBuckets) {
+      const lastInBucket = messages.at(-1)!;
+      cumulativeTotal += BigInt(messages.length);
+      await this.writeBucketSnapshot(seq, {
+        inboxRollingHash: lastInBucket.inboxRollingHash,
+        totalMsgCount: cumulativeTotal,
+        timestamp: lastInBucket.bucketTimestamp,
+        l1BlockNumber: lastInBucket.l1BlockNumber,
+        msgCount: messages.length,
+        firstMessageIndex: messages[0].index,
+        lastMessageIndex: lastInBucket.index,
+      });
+    }
+  }
+
   /**
    * Gets the L1 to L2 message index in the L1 to L2 message tree.
    * @param l1ToL2Message - The L1 to L2 message.
@@ -391,11 +468,14 @@ export class MessageStore {
         }
         msgCount += 1;
       }
+      const stored = await this.getBucketSnapshotBySeq(boundarySeq);
       await this.writeBucketSnapshot(boundarySeq, {
         inboxRollingHash: lastRemaining.inboxRollingHash,
         totalMsgCount: await this.getTotalL1ToL2MessageCount(),
         timestamp: lastRemaining.bucketTimestamp,
+        l1BlockNumber: stored?.l1BlockNumber ?? lastRemaining.l1BlockNumber,
         msgCount,
+        firstMessageIndex: stored?.firstMessageIndex ?? lastRemaining.index - BigInt(msgCount) + 1n,
         lastMessageIndex: lastRemaining.index,
       });
     }
@@ -466,16 +546,28 @@ export class MessageStore {
     await this.#bucketTimestampToSeq.set(this.timestampToKey(snapshot.timestamp), this.bucketSeqToKey(seq));
   }
 
-  private async toInboxBucket(seq: bigint, snapshot: BucketSnapshot): Promise<InboxBucket> {
-    const lastMessage = await this.getLastMessage();
+  /** Returns the sequence number of the newest stored bucket, or undefined if none has been stored yet. */
+  private async getNewestBucketSeq(): Promise<bigint | undefined> {
+    const [seqKey] = await toArray(this.#inboxBuckets.keysAsync({ reverse: true, limit: 1 }));
+    return seqKey === undefined ? undefined : BigInt(seqKey);
+  }
+
+  /** Returns the cumulative Inbox message count through the newest stored bucket before the given sequence number. */
+  private async getTotalMsgCountBeforeBucket(seq: bigint): Promise<bigint> {
+    const [snapBuffer] = await toArray(
+      this.#inboxBuckets.valuesAsync({ end: this.bucketSeqToKey(seq) - 1, reverse: true, limit: 1 }),
+    );
+    return snapBuffer === undefined ? 0n : deserializeBucketSnapshot(snapBuffer).totalMsgCount;
+  }
+
+  private toInboxBucket(seq: bigint, snapshot: BucketSnapshot): InboxBucket {
     return {
       seq,
       inboxRollingHash: snapshot.inboxRollingHash,
       totalMsgCount: snapshot.totalMsgCount,
       timestamp: snapshot.timestamp,
       msgCount: snapshot.msgCount,
       lastMessageIndex: snapshot.lastMessageIndex,
-      isOpen: lastMessage?.bucketSeq === seq,
     };
   }
 

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

@@ -216,12 +216,12 @@ describe('ArchiverApiSchema', () => {
 
   it('getLatestInboxBucketAtOrBefore', async () => {
     const result = await context.client.getLatestInboxBucketAtOrBefore(123n);
-    expect(result).toMatchObject({ seq: 1n, msgCount: 3, totalMsgCount: 3n, isOpen: false });
+    expect(result).toMatchObject({ seq: 1n, msgCount: 3, totalMsgCount: 3n });
   });
 
   it('getInboxBucket', async () => {
     const result = await context.client.getInboxBucket(2n);
-    expect(result).toMatchObject({ seq: 2n, msgCount: 3, isOpen: true });
+    expect(result).toMatchObject({ seq: 2n, msgCount: 3 });
   });
 
   it('getL1ToL2MessagesBetweenBuckets', async () => {
@@ -576,7 +576,6 @@ class MockArchiver implements ArchiverApi {
       timestamp: 100n,
       msgCount: 3,
       lastMessageIndex: 2n,
-      isOpen: false,
     });
   }
   getInboxBucket(seq: bigint): Promise<InboxBucket | undefined> {
@@ -588,7 +587,6 @@ class MockArchiver implements ArchiverApi {
       timestamp: 100n,
       msgCount: 3,
       lastMessageIndex: 2n,
-      isOpen: true,
     });
   }
   getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {

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

@@ -25,11 +25,6 @@ export type InboxBucket = {
   msgCount: number;
   /** Global leaf index of the last message absorbed into this bucket. */
   lastMessageIndex: bigint;
-  /**
-   * Whether this is the latest bucket the archiver has synced. A latest bucket may still grow as more messages
-   * arrive on L1; earlier buckets are complete. Consumers that need a settled bucket apply a lag on the timestamp.
-   */
-  isOpen: boolean;
 };
 
 export const InboxBucketSchema = z.object({
@@ -39,5 +34,4 @@ export const InboxBucketSchema = z.object({
   timestamp: schemas.BigInt,
   msgCount: schemas.Integer,
   lastMessageIndex: schemas.BigInt,
-  isOpen: z.boolean(),
 }) satisfies z.ZodType<InboxBucket>;

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

@@ -460,8 +460,8 @@ describe('ValidatorClient Integration', () => {
     it('validates and attests with txs anchored to proposed blocks and non-empty l1-to-l2 messages', async () => {
       // Create l1 to l2 messages and seed them into the archivers
       const l1ToL2Messages = makeInboxMessages(4, { messagesPerCheckpoint: 4 });
-      await proposer.archiver.dataStores.messages.addL1ToL2Messages(l1ToL2Messages);
-      await attestor.archiver.dataStores.messages.addL1ToL2Messages(l1ToL2Messages);
+      await proposer.archiver.dataStores.messages.addL1ToL2MessageBuckets(l1ToL2Messages);
+      await attestor.archiver.dataStores.messages.addL1ToL2MessageBuckets(l1ToL2Messages);
 
       // Build txs anchored to the previously proposed block
       const { blocks, proposal } = await buildCheckpoint(
@@ -660,10 +660,10 @@ describe('ValidatorClient Integration', () => {
 
     it('refuses block proposal with mismatching l1 to l2 messages', async () => {
       const l1ToL2Messages = makeInboxMessages(4, { messagesPerCheckpoint: 4 });
-      await proposer.archiver.dataStores.messages.addL1ToL2Messages(l1ToL2Messages);
+      await proposer.archiver.dataStores.messages.addL1ToL2MessageBuckets(l1ToL2Messages);
 
       const otherL1ToL2Messages = makeInboxMessages(4, { messagesPerCheckpoint: 4 });
-      await attestor.archiver.dataStores.messages.addL1ToL2Messages(otherL1ToL2Messages);
+      await attestor.archiver.dataStores.messages.addL1ToL2MessageBuckets(otherL1ToL2Messages);
 
       const { blocks } = await buildCheckpoint(
         CheckpointNumber(1),