fix(fast-inbox): preserve proposals across bucket repartitioning

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

@@ -24,6 +24,7 @@ import { openTmpStore } from '@aztec/kv-store/lmdb-v2';
 import { GENESIS_BLOCK_HEADER_HASH, L2BlockSourceEvents, type L2BlockSourceUpdatedEvent } from '@aztec/stdlib/block';
 import type { ProposedCheckpointInput } from '@aztec/stdlib/checkpoint';
 import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers';
+import { InboxBucketRef } from '@aztec/stdlib/messaging';
 import { CheckpointHeader } from '@aztec/stdlib/rollup';
 import { mockCheckpointAndMessages } from '@aztec/stdlib/testing';
 import { ConsensusTimetable } from '@aztec/stdlib/timetable';
@@ -38,7 +39,7 @@ import { type MockProxy, mock } from 'jest-mock-extended';
 import type { GetBlockReturnType } from 'viem';
 
 import { Archiver, type ArchiverEmitter } from './archiver.js';
-import { BlockOrCheckpointSlotExpiredError } from './errors.js';
+import { BlockOrCheckpointSlotExpiredError, InboxPrefixMismatchError } from './errors.js';
 import type { ArchiverInstrumentation } from './modules/instrumentation.js';
 import { ArchiverL1Synchronizer } from './modules/l1_synchronizer.js';
 import { type ArchiverDataStores, createArchiverDataStores } from './store/data_stores.js';
@@ -2324,6 +2325,37 @@ describe('Archiver Sync', () => {
       return blocks;
     };
 
+    it('rejects the caller and emits no update event when a proposed block fails the prefix guard', async () => {
+      // The facade's promise is tied to the atomic updater call, so a prefix rejection has to surface to the
+      // producer and leave the pass looking like a no-op: no block stored, no tip moved, no aggregate event.
+      addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      const updateSpy = jest.fn();
+      archiver.events.on(L2BlockSourceEvents.L2BlockSourceUpdated, updateSpy);
+      try {
+        const [block] = await makeBlocksConsumingThrough([2]);
+
+        // The reference names a prefix this archiver's messages do not hash to at the block's end count.
+        await expect(archiver.addBlock(block, new InboxBucketRef(Fr.random()))).rejects.toThrow(
+          InboxPrefixMismatchError,
+        );
+
+        expect(await archiver.getBlockNumber()).toEqual(BlockNumber(0));
+        expect(await archiver.getBlock({ number: BlockNumber(1) })).toBeUndefined();
+        expect(updateSpy).not.toHaveBeenCalled();
+        expect(pruneSpy).not.toHaveBeenCalled();
+
+        // The same block with the right reference is accepted, so the rejection was the guard and not the fixture.
+        const inboxRollingHash = (await archiverStore.messages.getInboxRollingHashAt(2n))!;
+        await archiver.addBlock(block, new InboxBucketRef(inboxRollingHash));
+        expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1));
+      } finally {
+        archiver.events.off(L2BlockSourceEvents.L2BlockSourceUpdated, updateSpy);
+      }
+    });
+
     it('prunes from the first proposed block that consumed a removed message', async () => {
       const { early } = addMessages();
       fake.setL1BlockNumber(l1BlockNumber);
@@ -2417,7 +2449,7 @@ describe('Archiver Sync', () => {
       });
     });
 
-    it('prunes when the reorg merges away the bucket boundary a block ended on', async () => {
+    it('keeps the proposed chain when the reorg merges away the bucket boundary a block ended on', async () => {
       const { early, late } = addMessages();
       fake.setL1BlockNumber(l1BlockNumber);
       await archiver.syncImmediate();
@@ -2427,25 +2459,27 @@ describe('Archiver Sync', () => {
         await archiver.addBlock(block);
       }
       expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
-      // The rolling hash the checkpoint carrying these blocks committed to.
-      const consumedRollingHash = (await archiverStore.messages.getInboxBucket(2n))!.inboxRollingHash;
+      // The prefix hash block 1 signed, which is also the boundary hash of the bucket it ended on.
+      const block1PrefixHash = (await archiverStore.messages.getInboxRollingHashAt(2n))!;
 
       // The messages of block 100 are re-mined into block 102, so all four end up in a single bucket. Every leaf
-      // survives under its own index, but the boundary at leaf count 2 does not, and block 1 ended on it: nothing
-      // can derive the messages that block inserted any more, so the chain built on it goes.
+      // survives under its own index and so does every prefix hash; only the boundary at leaf count 2 is gone. A
+      // block is authenticated by the prefix it consumed, not by the boundary it stopped on, so nothing is pruned.
       fake.retimeMessages(100n, 102n);
       await archiver.syncImmediate();
 
       expect(await getStoredLeaves()).toEqual(asHex([...early, ...late]));
-      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(0));
-      expect(pruneSpy).toHaveBeenCalledWith(
-        expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks }),
-      );
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      expect(pruneSpy).not.toHaveBeenCalled();
+      // The bucket metadata is rewritten even though not a single message moved.
       expect(await archiverStore.messages.getInboxBucket(1n)).toMatchObject({ msgCount: 4, l1BlockNumber: 102n });
       expect(await archiverStore.messages.getInboxBucket(2n)).toBeUndefined();
-      // The merged bucket still carries the hash the sealed header committed to, so a checkpoint whose blocks the
-      // merge left alone stays publishable; here they did not survive it.
-      expect(await archiverStore.messages.getInboxBucketByRollingHash(consumedRollingHash)).toMatchObject({ seq: 1n });
+      // Block 1's signed prefix still resolves by count, so its bundle is still derivable — even though no bucket
+      // ends there any more.
+      expect(await archiverStore.messages.getInboxRollingHashAt(2n)).toEqual(block1PrefixHash);
+      expect(await archiverStore.messages.getInboxBucketByRollingHash(block1PrefixHash)).toBeUndefined();
+      expect(await archiverStore.messages.getL1ToL2MessagesBetweenLeafCounts(0n, 2n)).toEqual(early);
+      expect(await archiverStore.messages.getL1ToL2MessagesBetweenLeafCounts(2n, 4n)).toEqual(late);
     });
 
     it('keeps the proposed chain when the reorg splits the bucket a block consumed', async () => {
@@ -2476,7 +2510,7 @@ describe('Archiver Sync', () => {
       expect(pruneSpy).not.toHaveBeenCalled();
     });
 
-    it('prunes when the reorg extends the last canonical bucket past the boundary a block ended on', async () => {
+    it('keeps the proposed chain when the reorg extends the last canonical bucket past a block boundary', async () => {
       const { early, late } = addMessages();
       // L1 block 101 carries block 100's timestamp, so a message mined there joins the bucket block 100 opened.
       fake.shareTimestampWithL1Block(101n, 100n);
@@ -2490,18 +2524,18 @@ describe('Archiver Sync', () => {
       expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
 
       // The two messages of L1 block 102 are re-mined into 101, which leaves block 100 untouched: the bucket the
-      // rollback walks back to is canonical on both of its own L1 blocks, and still absorbs the re-mined messages,
-      // so the boundary block 1 ended on is gone even though nothing below it moved.
+      // rollback walks back to is canonical on both of its own L1 blocks and now absorbs the re-mined messages too,
+      // so the boundary block 1 ended on is gone even though nothing below it moved. Content-wise the chain is
+      // unchanged, so both blocks stay.
       fake.retimeMessages(102n, 101n);
       await archiver.syncImmediate();
 
       expect(await getStoredLeaves()).toEqual(asHex([...early, ...late]));
       expect(await archiverStore.messages.getInboxBucket(1n)).toMatchObject({ msgCount: 4, totalMsgCount: 4n });
       expect(await archiverStore.messages.getInboxBucket(2n)).toBeUndefined();
-      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(0));
-      expect(pruneSpy).toHaveBeenCalledWith(
-        expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks }),
-      );
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      expect(pruneSpy).not.toHaveBeenCalled();
+      expect(await archiverStore.messages.getL1ToL2MessagesBetweenLeafCounts(0n, 2n)).toEqual(early);
     });
 
     it('prunes from the first leaf the reorg changed, not from the start of its bucket', async () => {

yarn-project/archiver/src/archiver.ts

@@ -31,7 +31,7 @@ import {
   getTimestampForSlot,
   getTimestampRangeForEpoch,
 } from '@aztec/stdlib/epoch-helpers';
-import type { L2ToL1MembershipWitness } from '@aztec/stdlib/messaging';
+import type { InboxBucketRef, L2ToL1MembershipWitness } from '@aztec/stdlib/messaging';
 import { ConsensusTimetable } from '@aztec/stdlib/timetable';
 import type { BlockHeader, TxHash } from '@aztec/stdlib/tx';
 import { type TelemetryClient, type Traceable, type Tracer, trackSpan } from '@aztec/telemetry-client';
@@ -55,6 +55,8 @@ export type { ArchiverEmitter };
 type AddBlockRequest = {
   type: 'block';
   block: L2Block;
+  /** The block proposal's signed Inbox prefix reference; absent only on the trusted compatibility path. */
+  inboxPrefixRef: InboxBucketRef | undefined;
   resolve: () => void;
   reject: (err: Error) => void;
 };
@@ -313,11 +315,14 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
    * The block will be processed by the sync loop.
    * Implements the L2BlockSink interface.
    * @param block - The L2 block to add.
+   * @param inboxPrefixRef - The block proposal's signed Inbox prefix reference, validated against this archiver's
+   *   own messages in the same transaction as the insert. See {@link L2BlockSink.addBlock} for when it may be
+   *   omitted.
    * @returns A promise that resolves when the block has been added to the store, or rejects on error.
    */
-  public addBlock(block: L2Block): Promise<void> {
+  public addBlock(block: L2Block, inboxPrefixRef?: InboxBucketRef): Promise<void> {
     const promise = promiseWithResolvers<void>();
-    this.inboundQueue.push({ block, ...promise, type: 'block' });
+    this.inboundQueue.push({ block, inboxPrefixRef, ...promise, type: 'block' });
     this.log.debug(`Queued block ${block.number} for processing`);
     void this.trySyncImmediate();
     return promise.promise;
@@ -379,7 +384,7 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
 
       try {
         if (type === 'block') {
-          const [durationMs] = await elapsed(() => this.updater.addProposedBlock(item.block));
+          const [durationMs] = await elapsed(() => this.updater.addProposedBlock(item.block, item.inboxPrefixRef));
           this.instrumentation.processNewProposedBlock(durationMs, item.block);
           blocksAdded.push(item.block);
         } else {

yarn-project/archiver/src/errors.ts

@@ -131,6 +131,71 @@ export class InboxMessageRangeNotSyncedError extends Error {
   }
 }
 
+/**
+ * Thrown when a proposed block's signed Inbox prefix reference cannot be checked against the local view, because the
+ * archiver has not synced a message at the block's end count or cannot serve its consumed range whole. Distinguishes
+ * "our view is behind, retry" from {@link InboxPrefixMismatchError}'s "our view disagrees".
+ */
+export class InboxPrefixNotSyncedError extends Error {
+  constructor(
+    public readonly blockNumber: number,
+    public readonly endTotalMsgCount: bigint,
+    cause?: string,
+  ) {
+    super(
+      `Cannot confirm the Inbox prefix at message count ${endTotalMsgCount} for proposed block ${blockNumber}` +
+        (cause ? `: ${cause}` : ''),
+    );
+    this.name = 'InboxPrefixNotSyncedError';
+  }
+}
+
+/**
+ * Thrown when a proposed block's signed Inbox prefix reference does not match the canonical prefix this archiver
+ * holds at the block's end count. The block consumed messages this node's view of L1 does not back, so inserting it
+ * would put a chain nothing can replay into the store.
+ */
+export class InboxPrefixMismatchError extends Error {
+  constructor(
+    public readonly blockNumber: number,
+    public readonly endTotalMsgCount: bigint,
+    public readonly expected: Fr,
+    public readonly actual: Fr,
+  ) {
+    super(
+      `Proposed block ${blockNumber} references Inbox prefix ${expected.toString()} at message count ` +
+        `${endTotalMsgCount}, but the canonical prefix there is ${actual.toString()}`,
+    );
+    this.name = 'InboxPrefixMismatchError';
+  }
+}
+
+/** Thrown when a proposed block's parent is not in the store, so its consumed range has no lower bound. */
+export class ProposedBlockParentNotFoundError extends Error {
+  constructor(
+    public readonly blockNumber: number,
+    public readonly parentBlockNumber: number,
+  ) {
+    super(`Cannot resolve parent block ${parentBlockNumber} of proposed block ${blockNumber}`);
+    this.name = 'ProposedBlockParentNotFoundError';
+  }
+}
+
+/** Thrown when a proposed block's end message count is below its parent's, so consumption would rewind. */
+export class InboxConsumptionRewindsError extends Error {
+  constructor(
+    public readonly blockNumber: number,
+    public readonly endTotalMsgCount: bigint,
+    public readonly parentTotalMsgCount: bigint,
+  ) {
+    super(
+      `Proposed block ${blockNumber} consumes through message count ${endTotalMsgCount}, ` +
+        `behind its parent's ${parentTotalMsgCount}`,
+    );
+    this.name = 'InboxConsumptionRewindsError';
+  }
+}
+
 /** Thrown when a proposed checkpoint number is stale (already processed). */
 export class ProposedCheckpointStaleError extends Error {
   constructor(

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

@@ -336,6 +336,10 @@ export abstract class ArchiverDataSourceBase
     return this.stores.messages.getInboxBucketByRollingHash(inboxRollingHash);
   }
 
+  public getInboxRollingHashAt(totalMsgCount: bigint): Promise<Fr | undefined> {
+    return this.stores.messages.getInboxRollingHashAt(totalMsgCount);
+  }
+
   public getL1ToL2MessagesBetweenBuckets(fromExclusive: bigint, toInclusive: bigint): Promise<Fr[]> {
     return this.stores.messages.getL1ToL2MessagesBetweenBuckets(fromExclusive, toInclusive);
   }