feat(fast-inbox): prune the proposed chain when an L1 reorg orphans consumed inbox messages

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

@@ -27,6 +27,7 @@ import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers';
 import { CheckpointHeader } from '@aztec/stdlib/rollup';
 import { mockCheckpointAndMessages } from '@aztec/stdlib/testing';
 import { ConsensusTimetable } from '@aztec/stdlib/timetable';
+import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees';
 import { BlockHeader } from '@aztec/stdlib/tx';
 import { getTelemetryClient } from '@aztec/telemetry-client';
 
@@ -2165,6 +2166,193 @@ describe('Archiver Sync', () => {
     });
   });
 
+  describe('pruning proposed blocks that consumed rolled back messages', () => {
+    let pruneSpy: jest.Mock;
+
+    // L1 is held here for the whole test: a reorg replaces the head at the same height, so the sync still runs,
+    // while the slot of the proposed blocks is never in the past and neither slot-based prune fires.
+    const l1BlockNumber = 110n;
+
+    beforeEach(() => {
+      pruneSpy = jest.fn();
+      archiver.events.on(L2BlockSourceEvents.L2PruneUncheckpointed, pruneSpy);
+    });
+
+    afterEach(() => {
+      archiver.events.off(L2BlockSourceEvents.L2PruneUncheckpointed, pruneSpy);
+    });
+
+    // Two messages on L1 block 100 (indices 0 and 1) and two on block 102 (indices 2 and 3), so a reorg of block
+    // 102 removes everything from index 2 on.
+    const addMessages = () => {
+      const early = [Fr.random(), Fr.random()];
+      const late = [Fr.random(), Fr.random()];
+      fake.addMessages(CheckpointNumber(1), 100n, early);
+      fake.addMessages(CheckpointNumber(1), 102n, late);
+      return { early, late };
+    };
+
+    // Builds a chain of blocks whose L1-to-L2 tree leaf counts are the given cumulative totals of consumed messages.
+    const makeBlocksConsumingThrough = async (leafCounts: number[], builtAtL1Block = l1BlockNumber) => {
+      const blocks = await fake.makeBlocks(CheckpointNumber(1), {
+        numBlocks: leafCounts.length,
+        txsPerBlock: 1,
+        l1BlockNumber: builtAtL1Block,
+      });
+      blocks.forEach((block, i) => {
+        block.header.state.l1ToL2MessageTree = new AppendOnlyTreeSnapshot(Fr.random(), leafCounts[i]);
+      });
+      return blocks;
+    };
+
+    it('prunes from the first proposed block that consumed a removed message', async () => {
+      const { early } = addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      // Blocks 1 and 2 consumed only the two surviving messages (block 2 consumed none of its own), block 3
+      // consumed the two that the reorg removes, and block 4 consumed nothing but builds on block 3.
+      const blocks = await makeBlocksConsumingThrough([2, 2, 4, 4]);
+      for (const block of blocks) {
+        await archiver.addBlock(block);
+      }
+      await archiver.addProposedCheckpoint({
+        checkpointNumber: CheckpointNumber(1),
+        header: CheckpointHeader.empty({ slotNumber: fake.getL2SlotAtL1Block(l1BlockNumber) }),
+        startBlock: BlockNumber(1),
+        blockCount: blocks.length,
+        totalManaUsed: 0n,
+        feeAssetPriceModifier: 0n,
+      });
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(4));
+
+      // L1 block 102 is replaced by one that does not carry its two messages.
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      // The tips the world state follows move back with the store, which is what unwinds it.
+      expect((await archiver.getL2Tips()).proposed.number).toEqual(BlockNumber(2));
+      // The proposed checkpoint covering the pruned blocks is evicted along with them.
+      expect(await archiverStore.blocks.getLastProposedCheckpoint()).toBeUndefined();
+      expect(pruneSpy).toHaveBeenCalledWith(
+        expect.objectContaining({
+          type: L2BlockSourceEvents.L2PruneUncheckpointed,
+          slotNumber: fake.getL2SlotAtL1Block(l1BlockNumber),
+          blocks: [blocks[2], blocks[3]],
+        }),
+      );
+    });
+
+    it('prunes when the reorg replaces the removed messages at the same indices', async () => {
+      const { early, late } = addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      const blocks = await makeBlocksConsumingThrough([2, 4]);
+      for (const block of blocks) {
+        await archiver.addBlock(block);
+      }
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+
+      // The replacement L1 block carries the same two messages in the opposite order: the message count is
+      // unchanged and indices 2 and 3 are filled again, but block 2 consumed the leaves that were there before.
+      fake.reorderMessagesAtL1Block(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex([...early, late[1], late[0]]));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1));
+      expect(pruneSpy).toHaveBeenCalledWith(
+        expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks: [blocks[1]] }),
+      );
+    });
+
+    // The rollback and the prune have to commit together: if the messages were dropped on their own, the next
+    // sync pass would re-download the canonical ones, find the local state consistent with L1, and never come
+    // back to the proposed blocks built on the messages that are gone.
+    it('keeps the messages when pruning the proposed chain fails', async () => {
+      const { early, late } = addMessages();
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      const blocks = await makeBlocksConsumingThrough([2, 4]);
+      for (const block of blocks) {
+        await archiver.addBlock(block);
+      }
+
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      const pruneFailure = new Error('cannot remove blocks');
+      jest.spyOn(archiverStore.blocks, 'removeBlocksAfter').mockImplementationOnce(() => {
+        throw pruneFailure;
+      });
+
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await expect(archiver.syncImmediate()).rejects.toThrow(pruneFailure);
+
+      expect(await getStoredLeaves()).toEqual(asHex([...early, ...late]));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+
+      // With the store rolled back, the next pass sees the same divergence again and completes both halves.
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1));
+    });
+
+    // A rollback below what the published chain already consumed means the message store and a checkpoint L1
+    // accepted disagree about L1, and the message store is not necessarily the right view. Every proposed block
+    // then satisfies the predicate, so acting on it would drop blocks that consumed nothing removed while their
+    // published parent, which did, stays.
+    it('leaves proposed blocks alone when the rollback reaches below the checkpointed tip', async () => {
+      const { early } = addMessages();
+      const checkpointedBlocks = await makeBlocksConsumingThrough([2, 4], 105n);
+      await fake.addCheckpoint(CheckpointNumber(1), {
+        blocks: checkpointedBlocks,
+        l1BlockNumber: 105n,
+        numL1ToL2Messages: 0,
+      });
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+
+      // A proposed block that consumed nothing of its own, sitting on the checkpointed tip.
+      const [proposed] = await fake.makeBlocks(CheckpointNumber(2), { numBlocks: 1, txsPerBlock: 1, l1BlockNumber });
+      proposed.header.state.l1ToL2MessageTree = new AppendOnlyTreeSnapshot(Fr.random(), 4);
+      await archiver.addBlock(proposed);
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3));
+
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3));
+      expect(pruneSpy).not.toHaveBeenCalled();
+    });
+
+    it('leaves checkpointed blocks alone when the messages they consumed are rolled back', async () => {
+      const { early } = addMessages();
+      const blocks = await makeBlocksConsumingThrough([2, 4], 105n);
+      await fake.addCheckpoint(CheckpointNumber(1), { blocks, l1BlockNumber: 105n, numL1ToL2Messages: 0 });
+      fake.setL1BlockNumber(l1BlockNumber);
+      await archiver.syncImmediate();
+      expect(await archiver.getCheckpointNumber()).toEqual(CheckpointNumber(1));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+
+      // Only the L1 checkpoint comparison may unwind published state, so the message rollback leaves it in place
+      // even though block 2 consumed a message that is now gone.
+      fake.removeMessagesAfter(2);
+      fake.reorgL1BlocksFrom(102n);
+      await archiver.syncImmediate();
+
+      expect(await getStoredLeaves()).toEqual(asHex(early));
+      expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2));
+      expect(pruneSpy).not.toHaveBeenCalled();
+    });
+  });
+
   describe('finalized checkpoint', () => {
     it('reports no finalized blocks before any checkpoint is proven', async () => {
       fake.setL1BlockNumber(100n);

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

@@ -1,3 +1,4 @@
+import type { L1BlockId } from '@aztec/ethereum/l1-types';
 import { BlockNumber, CheckpointNumber } from '@aztec/foundation/branded-types';
 import { filterAsync } from '@aztec/foundation/collection';
 import { createLogger } from '@aztec/foundation/log';
@@ -57,6 +58,10 @@ export class ArchiverDataStoreUpdater {
    * @param pendingChainValidationStatus - Optional validation status to set.
    * @returns True if the operation is successful.
    */
+  // TODO: a block built before an L1 reorg rolled back the messages it consumed can be pushed here after
+  // `removeProposedBlocksConsumingMessagesFrom` already pruned the chain it belongs to, and lands on the parent that
+  // survived the prune with nothing left to remove it. Closing it needs the builder to carry the message-store state
+  // it built against so this insert can reject a stale one.
   public async addProposedBlock(
     block: L2Block,
     pendingChainValidationStatus?: ValidateCheckpointResult,
@@ -255,22 +260,105 @@ export class ArchiverDataStoreUpdater {
    * @throws Error if any block to be removed is checkpointed.
    */
   public async removeUncheckpointedBlocksAfter(blockNumber: BlockNumber): Promise<L2Block[]> {
-    const result = await this.stores.db.transactionAsync(async () => {
-      // Verify we're only removing uncheckpointed blocks
-      const lastCheckpointedBlockNumber = await this.stores.blocks.getCheckpointedL2BlockNumber();
-      if (blockNumber < lastCheckpointedBlockNumber) {
-        throw new Error(
-          `Cannot remove blocks after ${blockNumber} because checkpointed blocks exist up to ${lastCheckpointedBlockNumber}`,
-        );
-      }
+    const result = await this.stores.db.transactionAsync(() => this.removeUncheckpointedBlocksAfterInTx(blockNumber));
+    await this.l2TipsCache?.refresh();
+    return result;
+  }
 
-      const prunedBlocks = await this.removeBlocksAfter(blockNumber);
-      await this.evictProposedCheckpointsForPrunedBlocks(prunedBlocks);
+  /** Body of {@link removeUncheckpointedBlocksAfter}, for callers that supply the transaction and the tips refresh. */
+  private async removeUncheckpointedBlocksAfterInTx(blockNumber: BlockNumber): Promise<L2Block[]> {
+    // Verify we're only removing uncheckpointed blocks
+    const lastCheckpointedBlockNumber = await this.stores.blocks.getCheckpointedL2BlockNumber();
+    if (blockNumber < lastCheckpointedBlockNumber) {
+      throw new Error(
+        `Cannot remove blocks after ${blockNumber} because checkpointed blocks exist up to ${lastCheckpointedBlockNumber}`,
+      );
+    }
+
+    const prunedBlocks = await this.removeBlocksAfter(blockNumber);
+    await this.evictProposedCheckpointsForPrunedBlocks(prunedBlocks);
 
-      return prunedBlocks;
+    return prunedBlocks;
+  }
+
+  /**
+   * Rewinds the message sync point, dropping every L1-to-L2 message from `firstRemovedIndex` on, and prunes the
+   * proposed blocks that consumed one of them, in a single transaction.
+   *
+   * Splitting the two would make the prune unrecoverable: a crash after the messages were dropped leaves the
+   * proposed chain built on messages the store no longer holds, and the next sync pass re-downloads the canonical
+   * messages, finds the local state consistent with L1, and never comes back to those blocks.
+   *
+   * @returns The pruned blocks.
+   */
+  public async rewindMessagesAndPruneProposedBlocks(
+    messagesSyncPoint: L1BlockId,
+    firstRemovedIndex: bigint,
+  ): Promise<L2Block[]> {
+    const prunedBlocks = await this.stores.db.transactionAsync(async () => {
+      await this.stores.messages.rewindMessagesTo(messagesSyncPoint, firstRemovedIndex);
+      return await this.removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex);
     });
     await this.l2TipsCache?.refresh();
-    return result;
+    return prunedBlocks;
+  }
+
+  /**
+   * Removes the proposed (not yet L1-checkpointed) blocks that consumed an L1-to-L2 message at or after
+   * `firstRemovedIndex`, along with every block after them. A block's L1-to-L2 tree leaf count is the cumulative
+   * count of messages consumed through it, so a leaf count above the index means the block consumed a message the
+   * local chain no longer has, and one equal to it means the block stopped below it.
+   *
+   * The index is where the rollback rewound to, not the first leaf whose value actually changed, so a reorg that
+   * re-mines the same messages in a different L1 block also prunes blocks whose trees are unchanged and which L1
+   * would still accept. That over-pruning is accepted for now: the rollback rewinds before it knows what the
+   * canonical chain re-delivers, so it cannot tell a re-mine from a content change, and a prune from the first
+   * differing leaf has to wait until the rollback becomes content-aware.
+   *
+   * Checkpointed blocks are never touched: a message store that disagrees with a checkpoint L1 accepted means one
+   * of the two views of L1 is mid-reorg and this one is not necessarily the right one, so only the archive
+   * comparison in the checkpoint step may unwind published state. For the same reason nothing is pruned when the
+   * rollback reaches below the checkpointed tip's leaf count, where every proposed descendant would qualify.
+   *
+   * @param firstRemovedIndex - Index of the first L1-to-L2 message that was removed from the store.
+   * @returns The removed blocks.
+   */
+  private async removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex: bigint): Promise<L2Block[]> {
+    const [checkpointedBlockNumber, latestBlockNumber] = await Promise.all([
+      this.stores.blocks.getCheckpointedL2BlockNumber(),
+      this.stores.blocks.getLatestL2BlockNumber(),
+    ]);
+    if (latestBlockNumber <= checkpointedBlockNumber) {
+      return [];
+    }
+
+    const checkpointedTip =
+      checkpointedBlockNumber > 0
+        ? await this.stores.blocks.getBlockData({ number: BlockNumber(checkpointedBlockNumber) })
+        : undefined;
+    const checkpointedTipLeafCount = BigInt(
+      checkpointedTip?.header.state.l1ToL2MessageTree.nextAvailableLeafIndex ?? 0,
+    );
+    if (firstRemovedIndex < checkpointedTipLeafCount) {
+      this.log.warn(
+        `Not pruning proposed blocks: rollback index ${firstRemovedIndex} is below the checkpointed tip's leaf count`,
+        { firstRemovedIndex, checkpointedTipLeafCount, checkpointedBlockNumber },
+      );
+      return [];
+    }
+
+    const proposedBlocks = await this.stores.blocks.getBlocksData({
+      from: BlockNumber(checkpointedBlockNumber + 1),
+      limit: latestBlockNumber - checkpointedBlockNumber,
+    });
+    const firstAffected = proposedBlocks.find(
+      block => BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex) > firstRemovedIndex,
+    );
+    if (firstAffected === undefined) {
+      return [];
+    }
+
+    return await this.removeUncheckpointedBlocksAfterInTx(BlockNumber(firstAffected.header.getBlockNumber() - 1));
   }
 
   /**

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

@@ -85,7 +85,7 @@ export class ArchiverInstrumentation {
     this.pruneDuration = meter.createHistogram(Metrics.ARCHIVER_PRUNE_DURATION);
 
     this.pruneCount = createUpDownCounterWithDefault(meter, Metrics.ARCHIVER_PRUNE_COUNT, {
-      [Attributes.PRUNE_TYPE]: ['unproven', 'uncheckpointed', 'l1_conflict', 'orphan', 'l1_mismatch'],
+      [Attributes.PRUNE_TYPE]: ['unproven', 'uncheckpointed', 'l1_conflict', 'orphan', 'l1_mismatch', 'inbox_rollback'],
     });
 
     this.blockProposalTxTargetCount = createUpDownCounterWithDefault(
@@ -161,9 +161,10 @@ export class ArchiverInstrumentation {
    * type distinguishes the cause: 'uncheckpointed' (slot ended without a checkpoint), 'l1_conflict' (proposed blocks
    * conflicting with an L1 checkpoint), 'orphan' (no matching proposed checkpoint arrived before the deadline), or
    * 'l1_mismatch' (the local checkpointed tip diverged from L1 — an L1 reorg or a pruned/missed-proof checkpoint — so
-   * already-checkpointed blocks were rewound).
+   * already-checkpointed blocks were rewound), or 'inbox_rollback' (an L1 reorg orphaned Inbox messages the blocks
+   * had consumed).
    */
-  public recordPrune(pruneType: 'uncheckpointed' | 'l1_conflict' | 'orphan' | 'l1_mismatch') {
+  public recordPrune(pruneType: 'uncheckpointed' | 'l1_conflict' | 'orphan' | 'l1_mismatch' | 'inbox_rollback') {
     this.pruneCount.add(1, { [Attributes.PRUNE_TYPE]: pruneType });
   }