diff --git a/yarn-project/archiver/src/archiver-sync.test.ts b/yarn-project/archiver/src/archiver-sync.test.ts index f30b454aeb..290f59e64e 100644 --- a/yarn-project/archiver/src/archiver-sync.test.ts +++ b/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]); diff --git a/yarn-project/archiver/src/modules/inbox_message_synchronizer.ts b/yarn-project/archiver/src/modules/inbox_message_synchronizer.ts index 5ce17f739b..048a428ef7 100644 --- a/yarn-project/archiver/src/modules/inbox_message_synchronizer.ts +++ b/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) {