fix(fast-inbox): replay published blocks by message count

yarn-project/archiver/src/errors.ts

@@ -114,13 +114,18 @@ export class InboxBucketNotSyncedError extends Error {
 }
 
 /**
- * Thrown when a cumulative Inbox message count does not resolve to a bucket boundary this archiver has synced, either
- * because the count sits inside a bucket or because the bucket is not synced yet.
+ * Thrown when a cumulative Inbox message-count range is not fully backed by the messages this archiver has synced,
+ * either because it reaches past the synced tip or because the store is missing a message the range needs.
+ * Distinguishes "not available locally, retry once L1 sync catches up" from a genuinely empty range.
  */
-export class InboxBucketBoundaryNotSyncedError extends Error {
-  constructor(public readonly totalMsgCount: bigint) {
-    super(`No synced Inbox bucket ends at cumulative message count ${totalMsgCount}`);
-    this.name = 'InboxBucketBoundaryNotSyncedError';
+export class InboxMessageRangeNotSyncedError extends Error {
+  constructor(
+    public readonly startLeafCount: bigint,
+    public readonly endLeafCount: bigint,
+    detail: string,
+  ) {
+    super(`Inbox message range [${startLeafCount}, ${endLeafCount}) is not fully synced: ${detail}`);
+    this.name = 'InboxMessageRangeNotSyncedError';
   }
 }
 

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

@@ -40,7 +40,13 @@ 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 { InboxBucket, L1ToL2MessageSource, L2ToL1MembershipWitness } from '@aztec/stdlib/messaging';
+import type {
+  InboxBucket,
+  InboxMessagePosition,
+  InboxMessageRange,
+  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';
@@ -340,6 +346,18 @@ export abstract class ArchiverDataSourceBase
     return this.stores.messages.getL1ToL2MessagesBetweenLeafCounts(startLeafCount, endLeafCount);
   }
 
+  public getMessagePosition(totalMessageCount: bigint): Promise<InboxMessagePosition | undefined> {
+    return this.stores.messages.getMessagePosition(totalMessageCount);
+  }
+
+  public getSyncedMessagePosition(): Promise<InboxMessagePosition> {
+    return this.stores.messages.getSyncedMessagePosition();
+  }
+
+  public getL1ToL2MessageRange(startLeafCount: bigint, endLeafCount: bigint): Promise<InboxMessageRange> {
+    return this.stores.messages.getL1ToL2MessageRange(startLeafCount, endLeafCount);
+  }
+
   private async getPublishedCheckpointFromCheckpointData(checkpoint: CheckpointData): Promise<PublishedCheckpoint> {
     const blocksForCheckpoint = await this.stores.blocks.getBlocksForCheckpoint(checkpoint.checkpointNumber);
     if (!blocksForCheckpoint) {

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

@@ -2,12 +2,13 @@ import { CheckpointNumber } from '@aztec/foundation/branded-types';
 import { Buffer32 } from '@aztec/foundation/buffer';
 import { Fr } from '@aztec/foundation/curves/bn254';
 import { toArray } from '@aztec/foundation/iterable';
+import type { AztecAsyncKVStore } from '@aztec/kv-store';
 import { openTmpStore } from '@aztec/kv-store/lmdb-v2';
 import { Checkpoint, type PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
 import { updateInboxRollingHash } from '@aztec/stdlib/messaging';
 import '@aztec/stdlib/testing/jest';
 
-import { InboxBucketBoundaryNotSyncedError, InboxBucketNotSyncedError } from '../errors.js';
+import { InboxBucketNotSyncedError, InboxMessageRangeNotSyncedError } from '../errors.js';
 import type { InboxMessage } from '../structs/inbox_message.js';
 import {
   makeInboxMessage,
@@ -23,6 +24,7 @@ import { type ArchiverL1SynchPoint, getArchiverSynchPoint } from './data_stores.
 import { MessageStore, MessageStoreError } from './message_store.js';
 
 describe('MessageStore', () => {
+  let db: AztecAsyncKVStore;
   let blockStore: BlockStore;
   let messageStore: MessageStore;
   let publishedCheckpoints: PublishedCheckpoint[];
@@ -40,7 +42,7 @@ describe('MessageStore', () => {
     });
 
   beforeEach(async () => {
-    const db = await openTmpStore('message_store_test');
+    db = await openTmpStore('message_store_test');
     blockStore = new BlockStore(db);
     messageStore = new MessageStore(db);
     // Create checkpoints sequentially to ensure archive roots are chained properly.
@@ -224,6 +226,147 @@ describe('MessageStore', () => {
     });
   });
 
+  describe('iterateL1ToL2Messages', () => {
+    it('honours zero range bounds', async () => {
+      const msgs = makeInboxMessages(3);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Zero is a valid compact index, so an exclusive end of zero selects nothing rather than everything.
+      expect(await toArray(messageStore.iterateL1ToL2Messages({ end: 0n }))).toEqual([]);
+      expect(await toArray(messageStore.iterateL1ToL2Messages({ start: 0n, end: 2n }))).toEqual(msgs.slice(0, 2));
+      expect(await toArray(messageStore.iterateL1ToL2Messages({ start: 0n }))).toEqual(msgs);
+    });
+  });
+
+  describe('message positions', () => {
+    const zeroPosition = { totalMessageCount: 0n, rollingHash: Fr.ZERO };
+    const positionAfter = (msg: InboxMessage) => ({
+      totalMessageCount: msg.index + 1n,
+      rollingHash: msg.inboxRollingHash,
+    });
+
+    it('resolves the position at a synced message count', async () => {
+      const msgs = makeInboxMessages(6);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Position zero always resolves; every other count resolves to the hash stored with the message before it.
+      expect(await messageStore.getMessagePosition(0n)).toEqual(zeroPosition);
+      expect(await messageStore.getMessagePosition(1n)).toEqual(positionAfter(msgs[0]));
+      expect(await messageStore.getMessagePosition(4n)).toEqual(positionAfter(msgs[3]));
+      expect(await messageStore.getMessagePosition(6n)).toEqual(positionAfter(msgs[5]));
+      // Past the synced tip there is no position yet.
+      expect(await messageStore.getMessagePosition(7n)).toBeUndefined();
+      await expect(messageStore.getMessagePosition(-1n)).rejects.toThrow(/Invalid Inbox message count/);
+    });
+
+    it('resolves position zero on an empty store', async () => {
+      expect(await messageStore.getMessagePosition(0n)).toEqual(zeroPosition);
+      expect(await messageStore.getMessagePosition(1n)).toBeUndefined();
+      expect(await messageStore.getSyncedMessagePosition()).toEqual(zeroPosition);
+    });
+
+    it('hands out positions a caller can mutate without affecting later reads', async () => {
+      const position = (await messageStore.getMessagePosition(0n))!;
+      position.rollingHash = new Fr(99);
+      position.totalMessageCount = 99n;
+
+      expect(await messageStore.getMessagePosition(0n)).toEqual(zeroPosition);
+      expect((await messageStore.getL1ToL2MessageRange(0n, 0n)).start).toEqual(zeroPosition);
+    });
+
+    it('tracks the synced position through appends and removals', async () => {
+      const msgs = makeInboxMessages(6);
+      await messageStore.addL1ToL2MessageBuckets(msgs.slice(0, 4));
+      expect(await messageStore.getSyncedMessagePosition()).toEqual(positionAfter(msgs[3]));
+
+      await messageStore.addL1ToL2MessageBuckets(msgs.slice(4));
+      expect(await messageStore.getSyncedMessagePosition()).toEqual(positionAfter(msgs[5]));
+
+      await messageStore.removeL1ToL2Messages(2n);
+      expect(await messageStore.getSyncedMessagePosition()).toEqual(positionAfter(msgs[1]));
+      // The removed suffix no longer has positions.
+      expect(await messageStore.getMessagePosition(3n)).toBeUndefined();
+    });
+
+    it('reads a message range together with the positions at both bounds', async () => {
+      const msgs = makeInboxMessages(6);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+      const leaves = msgs.map(m => m.leaf);
+
+      expect(await messageStore.getL1ToL2MessageRange(0n, 6n)).toEqual({
+        messages: leaves,
+        start: zeroPosition,
+        end: positionAfter(msgs[5]),
+      });
+      expect(await messageStore.getL1ToL2MessageRange(1n, 4n)).toEqual({
+        messages: leaves.slice(1, 4),
+        start: positionAfter(msgs[0]),
+        end: positionAfter(msgs[3]),
+      });
+      // An empty range is valid at any synced count and returns equal positions.
+      expect(await messageStore.getL1ToL2MessageRange(3n, 3n)).toEqual({
+        messages: [],
+        start: positionAfter(msgs[2]),
+        end: positionAfter(msgs[2]),
+      });
+      expect(await messageStore.getL1ToL2MessageRange(6n, 6n)).toEqual({
+        messages: [],
+        start: positionAfter(msgs[5]),
+        end: positionAfter(msgs[5]),
+      });
+    });
+
+    it('reads the messages and the ending position from one snapshot under a concurrent removal', async () => {
+      const msgs = makeInboxMessages(6);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Both operations are queued without awaiting: the read runs as one store transaction, so it sees either the
+      // full sequence or the truncated one, never the leaves of one with the ending hash of the other.
+      const rangePromise = messageStore.getL1ToL2MessageRange(0n, 6n);
+      const removalPromise = messageStore.removeL1ToL2Messages(3n);
+      const [range] = await Promise.all([rangePromise, removalPromise]);
+
+      expect(range.messages).toEqual(msgs.map(m => m.leaf));
+      expect(range.end).toEqual(positionAfter(msgs[5]));
+      expect(await messageStore.getSyncedMessagePosition()).toEqual(positionAfter(msgs[2]));
+      await expect(messageStore.getL1ToL2MessageRange(0n, 6n)).rejects.toThrow(InboxMessageRangeNotSyncedError);
+    });
+
+    it('reads the empty range at position zero on an empty store', async () => {
+      expect(await messageStore.getL1ToL2MessageRange(0n, 0n)).toEqual({
+        messages: [],
+        start: zeroPosition,
+        end: zeroPosition,
+      });
+    });
+
+    it('throws on an invalid or unsynced message range', async () => {
+      const msgs = makeInboxMessages(6);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      await expect(messageStore.getL1ToL2MessageRange(3n, 9n)).rejects.toThrow(InboxMessageRangeNotSyncedError);
+      await expect(messageStore.getL1ToL2MessageRange(7n, 7n)).rejects.toThrow(InboxMessageRangeNotSyncedError);
+      await expect(messageStore.getL1ToL2MessageRange(5n, 3n)).rejects.toThrow(/Invalid Inbox leaf count range/);
+      await expect(messageStore.getL1ToL2MessageRange(-1n, 3n)).rejects.toThrow(/Invalid Inbox leaf count range/);
+    });
+
+    it('throws when the range or its starting position has a hole', async () => {
+      const msgs = makeInboxMessages(6);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+      await db.openMap<number, Buffer>('archiver_l1_to_l2_messages').delete(2);
+
+      // Index 2 is missing: ranges over it are short, and a range starting at count 3 has no starting position.
+      await expect(messageStore.getL1ToL2MessageRange(0n, 6n)).rejects.toThrow(InboxMessageRangeNotSyncedError);
+      await expect(messageStore.getL1ToL2MessageRange(3n, 6n)).rejects.toThrow(/missing the message at index 2/);
+      // Ranges that need neither the hole nor a position at it are unaffected.
+      expect(await messageStore.getL1ToL2MessageRange(4n, 6n)).toEqual({
+        messages: msgs.slice(4).map(m => m.leaf),
+        start: positionAfter(msgs[3]),
+        end: positionAfter(msgs[5]),
+      });
+    });
+  });
+
   describe('Inbox buckets', () => {
     // Builds `count` consecutive valid messages, then reassigns their bucket sequence and timestamp per the given
     // per-message spec so we can exercise multi-message and rollover buckets.
@@ -484,23 +627,61 @@ describe('MessageStore', () => {
       expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 0n)).toEqual([]);
     });
 
-    it('throws when a leaf count does not land on a synced bucket boundary', async () => {
+    it('returns messages between leaf counts interior to the bucket partition', async () => {
       const msgs = makeBucketedMessages(threeBucketSpec);
       await messageStore.addL1ToL2MessageBuckets(msgs);
+      const leaves = msgs.map(m => m.leaf);
 
-      // Counts inside a bucket and past the last synced bucket both fail rather than returning a partial range.
-      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 4n)).rejects.toThrow(
-        InboxBucketBoundaryNotSyncedError,
-      );
-      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(4n, 6n)).rejects.toThrow(
-        InboxBucketBoundaryNotSyncedError,
-      );
+      // Counts 1, 2 and 4 sit inside a bucket. A published block commits to a leaf count, and a reorg can merge away
+      // the bucket that ended there, so the range is addressed by message index alone.
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(1n, 6n)).toEqual(leaves.slice(1));
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 4n)).toEqual(leaves.slice(0, 4));
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(1n, 4n)).toEqual(leaves.slice(1, 4));
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(4n, 6n)).toEqual(leaves.slice(4));
+      // An empty range at an interior count consumes nothing rather than failing.
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(4n, 4n)).toEqual([]);
+    });
+
+    it('throws on an invalid or unsynced leaf count range', async () => {
+      const msgs = makeBucketedMessages(threeBucketSpec);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Ranges reaching past the synced tip fail rather than returning a partial range, empty ones included.
       await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(3n, 9n)).rejects.toThrow(
-        InboxBucketBoundaryNotSyncedError,
+        InboxMessageRangeNotSyncedError,
+      );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(7n, 7n)).rejects.toThrow(
+        InboxMessageRangeNotSyncedError,
       );
+      // Reversed and negative bounds are caller errors, reported like every other failure of this API: as a
+      // rejection, so remote callers behind the archiver RPC see them the same way.
       await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(5n, 3n)).rejects.toThrow(
         /Invalid Inbox leaf count range/,
       );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(-1n, 3n)).rejects.toThrow(
+        /Invalid Inbox leaf count range/,
+      );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, -1n)).rejects.toThrow(
+        /Invalid Inbox leaf count range/,
+      );
+    });
+
+    it('throws when the leaf count range has a hole', async () => {
+      const msgs = makeBucketedMessages(threeBucketSpec);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      // Defense in depth: insertion is contiguity-checked and removal only ever drops a suffix, so no caller can put
+      // a hole in the middle of the log. Punch one straight into the map to prove a short read is never returned.
+      await db.openMap<number, Buffer>('archiver_l1_to_l2_messages').delete(2);
+
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(0n, 6n)).rejects.toThrow(
+        InboxMessageRangeNotSyncedError,
+      );
+      await expect(messageStore.getL1ToL2MessagesBetweenLeafCounts(2n, 3n)).rejects.toThrow(
+        InboxMessageRangeNotSyncedError,
+      );
+      // Ranges that do not cover the hole are unaffected.
+      expect(await messageStore.getL1ToL2MessagesBetweenLeafCounts(3n, 6n)).toEqual(msgs.slice(3).map(m => m.leaf));
     });
 
     it('rewinds buckets when messages are removed', async () => {