fix(fast-inbox): fail loudly when an inbox bucket range names an unsynced bucket (A-1379)

`getL1ToL2MessagesBetweenBuckets` returned an empty array when either bound was
missing from the store, which callers could not tell apart from a range that
genuinely holds no messages — and every caller asks because it wants the range's
messages. Throw `InboxBucketNotSyncedError` instead, so a node that is behind on
L1 sync surfaces that rather than deriving an empty message bundle, and reject an
inverted range outright.

Author: spalladino

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

@@ -8,7 +8,7 @@ import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
 import { updateInboxRollingHash } from '@aztec/stdlib/messaging';
 import '@aztec/stdlib/testing/jest';
 
-import { L1ToL2MessagesNotReadyError } from '../errors.js';
+import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js';
 import { type InboxMessage, updateRollingHash } from '../structs/inbox_message.js';
 import {
   makeInboxMessage,
@@ -478,10 +478,18 @@ describe('MessageStore', () => {
       expect(await messageStore.getL1ToL2MessagesBetweenBuckets(2n, 3n)).toEqual(leaves.slice(5));
       // An empty (fromExclusive, toInclusive] range yields no messages.
       expect(await messageStore.getL1ToL2MessagesBetweenBuckets(3n, 3n)).toEqual([]);
-      // Unknown upper bucket yields no messages.
-      expect(await messageStore.getL1ToL2MessagesBetweenBuckets(0n, 9n)).toEqual([]);
-      // An unknown nonzero lower bucket is treated as unavailable, not as genesis.
-      expect(await messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).toEqual([]);
+    });
+
+    it('throws when a bucket bound has not been synced', async () => {
+      const msgs = makeBucketedMessages(threeBucketSpec);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // An unsynced bound is reported rather than collapsing into an empty range.
+      await expect(messageStore.getL1ToL2MessagesBetweenBuckets(0n, 9n)).rejects.toThrow(InboxBucketNotSyncedError);
+      await expect(messageStore.getL1ToL2MessagesBetweenBuckets(9n, 12n)).rejects.toThrow(InboxBucketNotSyncedError);
+      // A nonzero lower bound is never treated as genesis.
+      await expect(messageStore.getL1ToL2MessagesBetweenBuckets(4n, 5n)).rejects.toThrow(InboxBucketNotSyncedError);
+      await expect(messageStore.getL1ToL2MessagesBetweenBuckets(4n, 3n)).rejects.toThrow(/Invalid Inbox bucket range/);
     });
 
     it('rewinds buckets when messages are removed', async () => {

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

@@ -14,7 +14,7 @@ import {
 } from '@aztec/kv-store';
 import { type InboxBucket, InboxLeaf, updateInboxRollingHash } from '@aztec/stdlib/messaging';
 
-import { L1ToL2MessagesNotReadyError } from '../errors.js';
+import { InboxBucketNotSyncedError, L1ToL2MessagesNotReadyError } from '../errors.js';
 import {
   type InboxMessage,
   deserializeInboxMessage,
@@ -505,27 +505,27 @@ export class MessageStore {
   }
 
   /**
-   * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion
-   * order (AZIP-22 Fast Inbox). Returns an empty array if the upper bucket has not been synced.
+   * Returns the message leaves absorbed into buckets in the range `(fromExclusive, toInclusive]`, in insertion order
+   * (AZIP-22 Fast Inbox). Both bounds must name buckets this archiver has synced, so that an empty result means the
+   * range holds no messages rather than hiding an unsynced bound; callers route the
+   * `InboxBucketNotSyncedError` to their own catch-up handling. Sequence 0 is the genesis base case and always
+   * resolves: the range then starts at the first message of the Inbox.
    */
   public async getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
-    const toBucket = await this.getBucketSnapshotBySeq(toInclusive);
-    if (toBucket === undefined) {
-      return [];
+    if (fromExclusive > toInclusive) {
+      throw new Error(`Invalid Inbox bucket range (${fromExclusive}, ${toInclusive}]`);
     }
-    // A nonzero lower bound must reference a synced bucket; otherwise the range is unavailable (rather than
-    // defaulting to genesis, which would silently over-return). Sequence 0 is the genesis base case: start from
-    // the beginning of the Inbox.
-    let startIndex = 0n;
-    if (fromExclusive > 0n) {
-      const fromBucket = await this.getBucketSnapshotBySeq(fromExclusive);
-      if (fromBucket === undefined) {
-        return [];
-      }
-      startIndex = fromBucket.lastMessageIndex + 1n;
+    if (toInclusive === 0n) {
+      return [];
     }
-    const endIndexExclusive = toBucket.lastMessageIndex + 1n;
+    const endIndexExclusive = (await this.getSyncedBucketSnapshot(toInclusive)).lastMessageIndex + 1n;
+    const startIndex =
+      fromExclusive === 0n ? 0n : (await this.getSyncedBucketSnapshot(fromExclusive)).lastMessageIndex + 1n;
+    return this.getMessageLeavesInIndexRange(startIndex, endIndexExclusive);
+  }
 
+  /** Collects the message leaves in the global index range `[startIndex, endIndexExclusive)`, in insertion order. */
+  private async getMessageLeavesInIndexRange(startIndex: bigint, endIndexExclusive: bigint): Promise<Fr[]> {
     const leaves: Fr[] = [];
     for await (const msgBuffer of this.#l1ToL2Messages.valuesAsync({
       start: this.indexToKey(startIndex),
@@ -541,6 +541,15 @@ export class MessageStore {
     return buffer && deserializeBucketSnapshot(buffer);
   }
 
+  /** Reads a bucket snapshot, failing loudly if the archiver has not synced that bucket. */
+  private async getSyncedBucketSnapshot(seq: bigint): Promise<BucketSnapshot> {
+    const snapshot = await this.getBucketSnapshotBySeq(seq);
+    if (snapshot === undefined) {
+      throw new InboxBucketNotSyncedError(seq);
+    }
+    return snapshot;
+  }
+
   private async writeBucketSnapshot(seq: bigint, snapshot: BucketSnapshot): Promise<void> {
     await this.#inboxBuckets.set(this.bucketSeqToKey(seq), serializeBucketSnapshot(snapshot));
     await this.#bucketTimestampToSeq.set(this.timestampToKey(snapshot.timestamp), this.bucketSeqToKey(seq));

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

@@ -41,8 +41,10 @@ export interface L1ToL2MessageSource {
 
   /**
    * 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
-   * bucket has not been synced.
+   * order, for streaming message-bundle derivation (AZIP-22 Fast Inbox). Both bounds must name buckets the source
+   * has synced; it throws otherwise, so that an empty result means the range holds no messages instead of hiding an
+   * unsynced bound. Callers that can tolerate an unsynced source resolve both bounds first, or map the failure to
+   * their own catch-up handling.
    * @param fromExclusive - The lower bucket sequence bound, exclusive (0 means from the start of the Inbox).
    * @param toInclusive - The upper bucket sequence bound, inclusive.
    */