fix(archiver): scan the Inbox deployment block

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

@@ -1956,7 +1956,7 @@ describe('Archiver Sync', () => {
       // resumes there. No comparison with the Inbox covered them, so they leave no syncpoint and nothing was
       // announced.
       expect(await getStoredLeaves()).toEqual(asHex(msgs.slice(0, 4)));
-      expect((await archiverStore.messages.getScannedL1Block())?.l1BlockNumber).toEqual(106n);
+      expect((await archiverStore.messages.getScannedL1Block())?.l1BlockNumber).toEqual(107n);
       expect(await archiverStore.messages.getSynchedL1Block()).toBeUndefined();
       expect(archiver.getL1BlockNumber()).toBeUndefined();
 
@@ -2215,12 +2215,12 @@ describe('Archiver Sync', () => {
       const [a] = randomLeaves(1);
       fake.addMessages(CheckpointNumber(1), 2n, [a]);
       fake.setL1BlockNumber(4n);
-      // While the first batch (blocks 1-2) is being fetched, L1 replaces block 2 with a block carrying no message and
-      // shortens to it: the batch's logs belong to the old chain, its syncpoint block to the new one.
+      // While the batch covering blocks 2-3 is being fetched, L1 replaces block 2 with a block carrying no message
+      // and shortens to it: the batch's logs belong to the old chain, its syncpoint block to the new one.
       const readLogs = inboxContract.getMessageSentEvents.getMockImplementation()!;
       inboxContract.getMessageSentEvents.mockImplementation(async (from, to) => {
         const logs = await readLogs(from, to);
-        if (to === 2n) {
+        if (to === 3n) {
           inboxContract.getMessageSentEvents.mockImplementation(readLogs);
           fake.removeMessagesAfter(0);
           fake.reorgL1BlocksFrom(2n);
@@ -2399,28 +2399,28 @@ describe('Archiver Sync', () => {
       fake.addMessages(CheckpointNumber(1), 10n, [c]);
       fake.setL1BlockNumber(10n);
 
-      // The provider answers the blocks 1-2 range without B and then stops serving logs, so the empty blocks 3-4
+      // The provider answers the blocks 2-3 range without B and then stops serving logs, so the empty blocks 4-5
       // batch is the last one to be scanned. Every block involved stays canonical: only the response was incomplete.
       const readLogs = inboxContract.getMessageSentEvents.getMockImplementation()!;
       inboxContract.getMessageSentEvents.mockImplementation(async (from, to) => {
         const logs = await readLogs(from, to);
-        return to === 2n ? logs.slice(0, 1) : logs;
+        return to === 3n ? logs.slice(0, 1) : logs;
       });
-      fake.setMessageSentEventsFailure(from => from >= 5n);
+      fake.setMessageSentEventsFailure(from => from >= 6n);
       await expect(archiver.syncImmediate()).rejects.toThrow(/Cannot serve MessageSent logs/);
 
       expect(await getStoredLeaves()).toEqual(asHex([a]));
       expect(archiver.getL1BlockNumber()).toBeUndefined();
 
-      // A later view of L1 ends exactly at block 4, the last block scanned. Nothing has compared the log with the
+      // A later view of L1 ends exactly at block 5, the last block scanned. Nothing has compared the log with the
       // Inbox there, so the head must not be answered from the scanned cursor: B is still missing.
       inboxContract.getMessageSentEvents.mockImplementation(readLogs);
       fake.setMessageSentEventsFailure(undefined);
-      fake.setL1BlockNumber(4n);
+      fake.setL1BlockNumber(5n);
       await archiver.syncImmediate();
 
       expect(await getStoredLeaves()).toEqual(asHex([a, b]));
-      expect(archiver.getL1BlockNumber()).toEqual(4n);
+      expect(archiver.getL1BlockNumber()).toEqual(5n);
 
       fake.setL1BlockNumber(10n);
       await archiver.syncImmediate();
@@ -2464,7 +2464,7 @@ describe('Archiver Sync', () => {
       const [a] = randomLeaves(1);
       fake.addMessages(CheckpointNumber(1), 2n, [a]);
       fake.setL1BlockNumber(4n);
-      fake.setMessageSentEventsFailure((_from, to) => to >= 3n);
+      fake.setMessageSentEventsFailure((_from, to) => to >= 4n);
       await expect(archiver.syncImmediate()).rejects.toThrow(/Cannot serve MessageSent logs/);
       expect(await getStoredLeaves()).toEqual(asHex([a]));
       expect(await archiverStore.messages.getSynchedL1Block()).toBeUndefined();
@@ -2567,6 +2567,124 @@ describe('Archiver Sync', () => {
       });
     });
 
+    describe('deployment block ingestion', () => {
+      // The Inbox's first message can be sent by a later transaction inside the block the contracts were deployed
+      // in. An exclusive scanned cursor defaulting to that block would resume one block later and never read it.
+      it('fetches a message emitted in the deployment block when the head is still there', async () => {
+        const [a] = randomLeaves(1);
+        fake.addMessages(CheckpointNumber(1), 0n, [a]);
+        fake.setL1BlockNumber(0n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual(asHex([a]));
+        expect(synchronizer.isRecoveringMessages()).toBe(false);
+      });
+
+      it('fetches a deployment-block message once L1 has advanced past it', async () => {
+        const [a] = randomLeaves(1);
+        fake.addMessages(CheckpointNumber(1), 0n, [a]);
+        fake.setL1BlockNumber(6n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual(asHex([a]));
+        expect(archiver.getL1BlockNumber()).toEqual(6n);
+      });
+
+      it('does not lose index 0 when a later message arrives after it', async () => {
+        const [a, b] = randomLeaves(2);
+        fake.addMessages(CheckpointNumber(1), 0n, [a]);
+        fake.setL1BlockNumber(2n);
+        await archiver.syncImmediate();
+        fake.addMessages(CheckpointNumber(1), 4n, [b]);
+        fake.setL1BlockNumber(6n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual(asHex([a, b]));
+        expect(synchronizer.isRecoveringMessages()).toBe(false);
+      });
+
+      it('re-reads the deployment block after a restart, one L1 block at a time', async () => {
+        await useArchiver({ batchSize: 1 });
+        const [a] = randomLeaves(1);
+        fake.addMessages(CheckpointNumber(1), 0n, [a]);
+        fake.setL1BlockNumber(4n);
+        await archiver.syncImmediate();
+        expect(await getStoredLeaves()).toEqual(asHex([a]));
+
+        // A fresh archiver over the same store resumes from the persisted cursor and must not re-read or duplicate.
+        const restarted = await buildArchiver('archiver_message_recovery', { batchSize: 1, store: archiverStore });
+        try {
+          fake.setL1BlockNumber(6n);
+          await restarted.archiver.syncImmediate();
+          expect(await getStoredLeaves()).toEqual(asHex([a]));
+          expect(restarted.synchronizer.isRecoveringMessages()).toBe(false);
+        } finally {
+          await restarted.archiver.stop();
+        }
+      });
+
+      it('unsticks a store whose cursor was already rewound onto the deployment block', async () => {
+        const [a] = randomLeaves(1);
+        fake.addMessages(CheckpointNumber(1), 0n, [a]);
+        // Reproduce what a zero-anchor recovery persists: cursor pinned at the deployment block, empty log.
+        await archiverStore.messages.setMessageSyncState({
+          l1Block: { l1BlockNumber: 0n, l1BlockHash: fake.getL1BlockHash(0n) },
+          authenticated: false,
+        });
+        fake.setL1BlockNumber(6n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual(asHex([a]));
+        expect(synchronizer.isRecoveringMessages()).toBe(false);
+      });
+
+      it('refills the deployment block after a zero-anchor rollback rewinds onto it', async () => {
+        const [a, b] = randomLeaves(2);
+        // No finality marker, so the search has to look every candidate up rather than trusting a finalized height.
+        fake.setFinalizedL1BlockNumber(undefined);
+        fake.addMessages(CheckpointNumber(1), 0n, [a]);
+        fake.addMessages(CheckpointNumber(1), 100n, [b]);
+        fake.setL1BlockNumber(110n);
+        await archiver.syncImmediate();
+        await addLocalBlocksConsuming([2]);
+
+        // L1 re-mines from the deployment block itself and emits a different first message there, so no lookup can
+        // place anything and recovery keeps nothing. The refill then has to read the deployment block again.
+        const [replacement] = randomLeaves(1);
+        fake.removeMessagesAfter(0);
+        fake.addMessages(CheckpointNumber(1), 0n, [replacement]);
+        fake.reorgL1BlocksFrom(0n);
+        fake.setL1BlockNumber(111n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual(asHex([replacement]));
+        expect(await localBlockNumbers()).toEqual([]);
+        expect(archiver.getL1BlockNumber()).toEqual(111n);
+      });
+
+      it('stays synced when the deployment block holds no message', async () => {
+        fake.setL1BlockNumber(4n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual([]);
+        expect(archiver.getL1BlockNumber()).toEqual(4n);
+        expect(synchronizer.isRecoveringMessages()).toBe(false);
+      });
+
+      it('keeps exclusive semantics once the cursor is past the deployment block', async () => {
+        const [a, b] = randomLeaves(2);
+        fake.addMessages(CheckpointNumber(1), 2n, [a]);
+        fake.setL1BlockNumber(4n);
+        await archiver.syncImmediate();
+        fake.addMessages(CheckpointNumber(1), 6n, [b]);
+        fake.setL1BlockNumber(8n);
+        await archiver.syncImmediate();
+
+        expect(await getStoredLeaves()).toEqual(asHex([a, b]));
+        expect(synchronizer.isRecoveringMessages()).toBe(false);
+      });
+    });
+
     it('leaves the batch that refills the log after a rollback uncommitted when it disagrees with the Inbox', async () => {
       const [a, b, c, d] = randomLeaves(4);
       fake.addMessages(CheckpointNumber(1), 100n, [a]);

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

@@ -86,7 +86,9 @@ export type InboxMessageRecoveryProgress = {
  * needs no event lookups), or a stored message whose event L1 still emits at the same index and hash within five L1
  * blocks of the height it was observed at, found by walking the log backwards with a bounded number of event lookups
  * per pass. A lookup that misses moves the search to an older candidate, and running out of candidates falls back to
- * the deployment block. Once an anchor is chosen the log is rolled back to it in one store transaction: the suffix
+ * the deployment block, which is itself re-read: the Inbox's first message can be emitted by a later transaction in
+ * the block the contracts were deployed in, so the deployment block is the one block an exclusive cursor may not
+ * skip. Once an anchor is chosen the log is rolled back to it in one store transaction: the suffix
  * rows are deleted, the proposed blocks that consumed more messages than the retained count are pruned with their
  * descendants, the scanned cursor rewinds to the block before the anchor's and the syncpoint is cleared. Nothing is
  * fetched in that pass; ordinary forward ingestion refills the log from the rewound cursor, rewriting the retained
@@ -256,8 +258,9 @@ export class InboxMessageSynchronizer {
       return this.startRecovery(head, remote, finalizedL1Block);
     }
 
-    if (head.l1BlockNumber <= cursor.l1BlockNumber) {
-      // A head at or below what has already been scanned, and the log does not agree with it: there is no forward
+    const ingestFrom = this.ingestionStartFor(cursor);
+    if (head.l1BlockNumber < ingestFrom) {
+      // A head below the first block still to be scanned, and the log does not agree with it: there is no forward
       // range to fetch, so find where the local log and the canonical one part ways.
       return this.startRecovery(head, remote, finalizedL1Block);
     }
@@ -290,7 +293,7 @@ export class InboxMessageSynchronizer {
 
     let headBatch: InboxMessage[];
     try {
-      headBatch = await this.ingestForward(cursor.l1BlockNumber + 1n, head);
+      headBatch = await this.ingestForward(ingestFrom, head);
     } catch (err) {
       if (err instanceof CapturedHeadReplacedError) {
         this.log.verbose(`L1 head ${head.l1BlockNumber} was replaced while fetching L1 to L2 messages`);
@@ -376,6 +379,21 @@ export class InboxMessageSynchronizer {
     return 'lagged';
   }
 
+  /**
+   * First L1 block ordinary ingestion must read, given the scanned cursor.
+   *
+   * The cursor is exclusive, so fetching normally resumes at the block after it. The deployment block is the
+   * exception: message index 0 can be emitted by a later transaction in that very block, and a cursor sitting at it
+   * means nothing has read it yet — that is where the archiver starts with no persisted cursor, and where the
+   * zero-anchor rollback rewinds to. Resuming one block later would skip index 0 permanently, since no later message
+   * can fill the gap and every pass would rediscover the same disagreement. Only the deployment block is re-read;
+   * genuine completed cursors keep exclusive semantics, and re-reading it is harmless because the store rewrites an
+   * unchanged message in place.
+   */
+  private ingestionStartFor(cursor: L1BlockId): bigint {
+    return cursor.l1BlockNumber <= this.l1Start.l1BlockNumber ? this.l1Start.l1BlockNumber : cursor.l1BlockNumber + 1n;
+  }
+
   /**
    * Fetches messages forward in bounded L1 block ranges and commits each batch with the scanned cursor that covers
    * it, except for the batch reaching the head, which is returned staged instead of stored. No intermediate batch is
@@ -489,7 +507,7 @@ export class InboxMessageSynchronizer {
           headL1BlockNumber: recovery.head.l1BlockNumber,
           lookups: recovery.lookups,
         });
-        return { keep: zeroMessagePosition(), anchorL1Block: this.l1Start.l1BlockNumber + 1n };
+        return { keep: zeroMessagePosition(), anchorL1Block: this.l1Start.l1BlockNumber };
       }
       const candidate = await this.stores.messages.getL1ToL2Message(candidateIndex);
       if (candidate === undefined) {