perf(fast-inbox): roll back inbox messages by bucket, not by message (A-1388)

Finding the cut point for an L1-block rollback walked the messages backwards from
the tip, deserializing every message being removed just to learn where the removal
starts. A bucket lives entirely within one L1 block and its snapshot now records
that block plus the bucket's first message index, so scan the bucket snapshots
instead — a few records per L1 block rather than one per message.

Author: spalladino

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

@@ -225,11 +225,14 @@ describe('MessageStore', () => {
   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.
-    const makeBucketedMessages = (spec: { seq: bigint; timestamp: bigint }[]): InboxMessage[] => {
+    const makeBucketedMessages = (
+      spec: { seq: bigint; timestamp: bigint; l1BlockNumber?: bigint }[],
+    ): InboxMessage[] => {
       const msgs = makeInboxMessages(spec.length);
       msgs.forEach((msg, i) => {
         msg.bucketSeq = spec[i].seq;
         msg.bucketTimestamp = spec[i].timestamp;
+        msg.l1BlockNumber = spec[i].l1BlockNumber ?? msg.l1BlockNumber;
       });
       return msgs;
     };
@@ -505,6 +508,27 @@ describe('MessageStore', () => {
       expect((await messageStore.getLatestInboxBucketAtOrBefore(150n))!.seq).toEqual(1n);
     });
 
+    it('rolls back whole buckets past an L1 block', async () => {
+      // Bucket 1 fills L1 block 100, bucket 2 fills L1 block 101 and bucket 3 fills L1 block 102.
+      const msgs = makeBucketedMessages([
+        { seq: 1n, timestamp: 100n, l1BlockNumber: 100n },
+        { seq: 1n, timestamp: 100n, l1BlockNumber: 100n },
+        { seq: 1n, timestamp: 100n, l1BlockNumber: 100n },
+        { seq: 2n, timestamp: 200n, l1BlockNumber: 101n },
+        { seq: 2n, timestamp: 200n, l1BlockNumber: 101n },
+        { seq: 3n, timestamp: 300n, l1BlockNumber: 102n },
+      ]);
+      await messageStore.addL1ToL2MessageBuckets(msgs);
+
+      await messageStore.rollbackL1ToL2MessagesAfterL1Block(100n);
+
+      expect(await toArray(messageStore.iterateL1ToL2Messages())).toEqual(msgs.slice(0, 3));
+      expect(await messageStore.getInboxBucket(2n)).toBeUndefined();
+      expect(await messageStore.getInboxBucket(3n)).toBeUndefined();
+      expect(await messageStore.getInboxBucket(1n)).toMatchObject({ msgCount: 3, totalMsgCount: 3n });
+      expect((await messageStore.getLatestInboxBucketAtOrBefore(300n))!.seq).toEqual(1n);
+    });
+
     it('clears all buckets when every message is removed', async () => {
       const msgs = makeBucketedMessages(threeBucketSpec);
       await messageStore.addL1ToL2MessageBuckets(msgs);

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

@@ -582,15 +582,19 @@ export class MessageStore {
   /**
    * Removes every L1 to L2 message inserted after the given L1 block, so the message store matches the L1 Inbox state
    * as of that block. Used when rolling the archiver back to an earlier checkpoint, whose L1 block is passed here.
+   *
+   * A bucket lives entirely within one L1 block, so the cut always falls on a bucket boundary and can be found from
+   * the bucket snapshots alone, without reading the messages being removed.
    */
   public async rollbackL1ToL2MessagesAfterL1Block(l1BlockNumber: bigint): Promise<void> {
     this.#log.debug(`Deleting L1 to L2 messages inserted after L1 block ${l1BlockNumber}`);
     let removeFromIndex: bigint | undefined;
-    for await (const message of this.iterateL1ToL2Messages({ reverse: true })) {
-      if (message.l1BlockNumber <= l1BlockNumber) {
+    for await (const snapBuffer of this.#inboxBuckets.valuesAsync({ reverse: true })) {
+      const snapshot = deserializeBucketSnapshot(snapBuffer);
+      if (snapshot.l1BlockNumber <= l1BlockNumber) {
         break;
       }
-      removeFromIndex = message.index;
+      removeFromIndex = snapshot.firstMessageIndex;
     }
     if (removeFromIndex !== undefined) {
       await this.removeL1ToL2Messages(removeFromIndex);