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

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.