From dd2fe44fc925d81594628f18b25ae8ca9beccfd0 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 19:25:47 -0300 Subject: [PATCH 1/9] feat(fast-inbox): let the message store swap the buckets above a canonical one Adds a single-transaction replace that drops leaves only from a given index, deletes the bucket snapshots above the last canonical bucket, re-delivers the canonical messages and moves the syncpoint, so a rollback can commit what the canonical chain holds rather than only what it removes. The snapshot deletion is factored out of the post-removal rewind, since a re-delivery that renumbers or merges buckets is rejected as incomplete while the stale snapshots are still in place. --- .../archiver/src/store/message_store.test.ts | 131 ++++++++++++++++++ .../archiver/src/store/message_store.ts | 62 ++++++++- 2 files changed, 186 insertions(+), 7 deletions(-) diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index d25d75d4ca4e..42a1d833dce8 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -718,5 +718,136 @@ describe('MessageStore', () => { expect(await messageStore.getInboxBucket(1n)).toBeUndefined(); expect(await messageStore.getLatestInboxBucketAtOrBefore(300n)).toBeUndefined(); }); + + describe('replacing the messages above a bucket', () => { + const syncPoint = { l1BlockNumber: 900n, l1BlockHash: makeL1BlockHash(900n) }; + + // Re-times a message onto another L1 block and bucket, the way an L1 reorg re-mines one, leaving its leaf, + // index and rolling hash alone. + const reabsorb = (msg: InboxMessage, bucket: { seq: bigint; timestamp: bigint }): InboxMessage => ({ + ...msg, + bucketSeq: bucket.seq, + bucketTimestamp: bucket.timestamp, + l1BlockNumber: makeL1BlockNumberForBucket(bucket.timestamp), + l1BlockHash: makeL1BlockHash(makeL1BlockNumberForBucket(bucket.timestamp)), + }); + + it('re-times a bucket the reorg re-mined without touching its message', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + const retimed = reabsorb(msgs[5], { seq: 3n, timestamp: 350n }); + + await messageStore.replaceMessagesAboveBucket({ + lastCanonicalBucketSeq: 2n, + firstDifferingIndex: undefined, + messages: [retimed], + syncPoint, + finalizedL1Block: undefined, + }); + + expect(await toArray(messageStore.iterateL1ToL2Messages())).toEqual([...msgs.slice(0, 5), retimed]); + expect(await messageStore.getInboxBucket(3n)).toMatchObject({ + timestamp: 350n, + msgCount: 1, + totalMsgCount: 6n, + l1BlockNumber: retimed.l1BlockNumber, + }); + expect(await messageStore.getLatestInboxBucketAtOrBefore(300n)).toMatchObject({ seq: 2n }); + expect((await messageStore.getLatestInboxBucketAtOrBefore(350n))?.seq).toEqual(3n); + expect((await messageStore.getInboxBucketByRollingHash(msgs[5].inboxRollingHash))?.seq).toEqual(3n); + expect(await messageStore.getSynchedL1Block()).toEqual(syncPoint); + }); + + it('merges the buckets above the last canonical one', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + // The L1 blocks that opened buckets 2 and 3 are re-mined as one, so the six messages fill two buckets. + const merged = msgs.slice(3).map(msg => reabsorb(msg, { seq: 2n, timestamp: 250n })); + + await messageStore.replaceMessagesAboveBucket({ + lastCanonicalBucketSeq: 1n, + firstDifferingIndex: undefined, + messages: merged, + syncPoint, + finalizedL1Block: undefined, + }); + + expect(await toArray(messageStore.iterateL1ToL2Messages())).toEqual([...msgs.slice(0, 3), ...merged]); + expect(await messageStore.getInboxBucket(2n)).toMatchObject({ + timestamp: 250n, + msgCount: 3, + totalMsgCount: 6n, + lastMessageIndex: msgs[5].index, + }); + expect(await messageStore.getInboxBucket(3n)).toBeUndefined(); + expect(await messageStore.getLatestInboxBucketAtOrBefore(200n)).toMatchObject({ seq: 1n }); + expect((await messageStore.getLatestInboxBucketAtOrBefore(300n))?.seq).toEqual(2n); + // The bucket the merge dissolved carries no rolling hash any more, and the one that absorbed it ends on + // the hash a checkpoint consuming through the old bucket 3 committed to. + expect(await messageStore.getInboxBucketByRollingHash(msgs[4].inboxRollingHash)).toBeUndefined(); + expect((await messageStore.getInboxBucketByRollingHash(msgs[5].inboxRollingHash))?.seq).toEqual(2n); + }); + + it('splits a bucket the reorg cut in two', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + // Bucket 2's two messages were absorbed from co-timestamped L1 blocks; re-mining the second with a + // timestamp of its own splits them, and every bucket after it is renumbered. + const split = [ + reabsorb(msgs[3], { seq: 2n, timestamp: 200n }), + reabsorb(msgs[4], { seq: 3n, timestamp: 250n }), + reabsorb(msgs[5], { seq: 4n, timestamp: 300n }), + ]; + + await messageStore.replaceMessagesAboveBucket({ + lastCanonicalBucketSeq: 1n, + firstDifferingIndex: undefined, + messages: split, + syncPoint, + finalizedL1Block: undefined, + }); + + expect(await toArray(messageStore.iterateL1ToL2Messages())).toEqual([...msgs.slice(0, 3), ...split]); + expect(await messageStore.getInboxBucket(2n)).toMatchObject({ msgCount: 1, totalMsgCount: 4n }); + expect(await messageStore.getInboxBucket(3n)).toMatchObject({ msgCount: 1, totalMsgCount: 5n }); + expect(await messageStore.getInboxBucket(4n)).toMatchObject({ msgCount: 1, totalMsgCount: 6n }); + expect((await messageStore.getLatestInboxBucketAtOrBefore(250n))?.seq).toEqual(3n); + expect((await messageStore.getInboxBucketByRollingHash(msgs[4].inboxRollingHash))?.seq).toEqual(3n); + expect((await messageStore.getInboxBucketByRollingHash(msgs[5].inboxRollingHash))?.seq).toEqual(4n); + }); + + it('drops the messages from the first differing leaf and stores the canonical ones', async () => { + const msgs = makeBucketedMessages(threeBucketSpec); + await messageStore.addL1ToL2MessageBuckets(msgs); + // The canonical chain keeps bucket 2's first message, replaces its second and carries nothing after it. + const replacement = makeNextMessage(msgs[3], { seq: 2n, timestamp: 200n }); + + await messageStore.replaceMessagesAboveBucket({ + lastCanonicalBucketSeq: 1n, + firstDifferingIndex: msgs[4].index, + messages: [msgs[3], replacement], + syncPoint, + finalizedL1Block: { l1BlockNumber: 800n, l1BlockHash: makeL1BlockHash(800n) }, + }); + + expect(await toArray(messageStore.iterateL1ToL2Messages())).toEqual([...msgs.slice(0, 4), replacement]); + expect(await messageStore.getTotalL1ToL2MessageCount()).toEqual(5n); + expect(await messageStore.getInboxBucket(2n)).toMatchObject({ + msgCount: 2, + totalMsgCount: 5n, + inboxRollingHash: replacement.inboxRollingHash, + }); + expect(await messageStore.getInboxBucket(3n)).toBeUndefined(); + expect(await messageStore.getLatestInboxBucketAtOrBefore(300n)).toMatchObject({ seq: 2n }); + expect(await messageStore.getInboxBucketByRollingHash(msgs[4].inboxRollingHash)).toBeUndefined(); + expect(await messageStore.getInboxBucketByRollingHash(msgs[5].inboxRollingHash)).toBeUndefined(); + expect((await messageStore.getInboxBucketByRollingHash(replacement.inboxRollingHash))?.seq).toEqual(2n); + // The leaves that are gone no longer resolve to an index, and the replacement does. + expect(await messageStore.getL1ToL2MessageIndex(msgs[4].leaf)).toBeUndefined(); + expect(await messageStore.getL1ToL2MessageIndex(replacement.leaf)).toEqual(replacement.index); + expect(await messageStore.getSynchedL1Block()).toEqual(syncPoint); + expect(await messageStore.getMessagesFinalizedL1Block()).toMatchObject({ l1BlockNumber: 800n }); + }); + }); }); }); diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index fcc462c5d578..50c73bda0bf3 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -453,13 +453,7 @@ export class MessageStore { const lastRemaining = await this.getLastMessage(); const boundarySeq = lastRemaining?.bucketSeq; - const deleteFromKey = boundarySeq === undefined ? 0 : this.bucketSeqToKey(boundarySeq) + 1; - for await (const [seqKey, snapBuffer] of this.#inboxBuckets.entriesAsync({ start: deleteFromKey })) { - const snapshot = deserializeBucketSnapshot(snapBuffer); - await this.#bucketTimestampToSeq.deleteValue(this.timestampToKey(snapshot.timestamp), seqKey); - await this.#bucketRollingHashToSeq.delete(this.rollingHashToKey(snapshot.inboxRollingHash)); - await this.#inboxBuckets.delete(seqKey); - } + await this.deleteBucketSnapshotsAbove(boundarySeq); if (lastRemaining === undefined || boundarySeq === undefined) { return; @@ -483,6 +477,60 @@ export class MessageStore { }); } + /** + * Deletes the snapshot of every bucket above `seq`, together with its timestamp and rolling-hash index entries. + * The timestamp entry is deleted by value, so the rollover siblings a bucket shares a timestamp with stay indexed. + * Passing undefined deletes every snapshot. Must run inside the transaction that rewrites the buckets. + */ + private async deleteBucketSnapshotsAbove(seq: bigint | undefined): Promise { + const deleteFromKey = seq === undefined ? 0 : this.bucketSeqToKey(seq) + 1; + for await (const [seqKey, snapBuffer] of this.#inboxBuckets.entriesAsync({ start: deleteFromKey })) { + const snapshot = deserializeBucketSnapshot(snapBuffer); + await this.#bucketTimestampToSeq.deleteValue(this.timestampToKey(snapshot.timestamp), seqKey); + await this.#bucketRollingHashToSeq.delete(this.rollingHashToKey(snapshot.inboxRollingHash)); + await this.#inboxBuckets.delete(seqKey); + } + } + + /** + * Replaces everything the store holds above the given bucket with the messages the canonical chain delivers for the + * same range, and moves the sync point to the L1 block that range was fetched up to, all in a single transaction. + * + * The caller has already compared the two views: `firstDifferingIndex` is the lowest index whose leaf changed, and + * is undefined when the canonical chain re-delivers the very same leaves, in which case not a single message is + * dropped and only the bucket metadata derived from the L1 blocks moves. The steps depend on each other in order: + * the removal rewrites the snapshot of the bucket left holding the last surviving message, so it has to run while + * that snapshot is still there; the snapshots above the last canonical bucket then go, because a re-delivery that + * renumbers or merges buckets would otherwise be rejected as incomplete or leave a stale snapshot behind; and only + * then can the canonical messages be delivered, which rewrites the snapshots from the messages themselves. + * + * `messages` must start at the first message of the last canonical bucket (or of the Inbox when there is none), so + * that every bucket it covers arrives whole. + * + * @param lastCanonicalBucketSeq - Sequence of the newest bucket known to sit on canonical L1 blocks, if any. + * @param firstDifferingIndex - Lowest message index whose leaf the canonical chain changed, if any. + * @param messages - The canonical messages from the last canonical bucket's opening L1 block onwards. + * @param syncPoint - L1 block the messages were fetched up to. + * @param finalizedL1Block - L1 finalized block to record, if it is at or below the sync point. + */ + public replaceMessagesAboveBucket(args: { + lastCanonicalBucketSeq: bigint | undefined; + firstDifferingIndex: bigint | undefined; + messages: InboxMessage[]; + syncPoint: L1BlockId; + finalizedL1Block: L1BlockId | undefined; + }): Promise { + const { lastCanonicalBucketSeq, firstDifferingIndex, messages, syncPoint, finalizedL1Block } = args; + return this.db.transactionAsync(async () => { + if (firstDifferingIndex !== undefined) { + await this.removeL1ToL2Messages(firstDifferingIndex); + } + await this.deleteBucketSnapshotsAbove(lastCanonicalBucketSeq); + await this.addL1ToL2MessageBuckets(messages); + await this.setMessageSyncState(syncPoint, finalizedL1Block); + }); + } + /** * Atomically drops every message from `startIndex` on (when given) and rewinds the message sync point, so that an * interruption cannot leave truncated messages behind a sync point that would never fetch them again. From 5f575623e5044fcfa993a244f294211c0ec7fb05 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 19:25:52 -0300 Subject: [PATCH 2/9] feat(fast-inbox): commit the inbox message swap and the proposed-chain prune together Wraps the store swap and the prune of the proposed blocks built on the leaves it changed in one transaction, the way the rewind already does, so a crash cannot leave a proposed chain standing on messages the store replaced. --- .../src/modules/data_store_updater.ts | 36 ++++++++++++++++--- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/yarn-project/archiver/src/modules/data_store_updater.ts b/yarn-project/archiver/src/modules/data_store_updater.ts index 2ebe1839cfc7..127937f192bd 100644 --- a/yarn-project/archiver/src/modules/data_store_updater.ts +++ b/yarn-project/archiver/src/modules/data_store_updater.ts @@ -24,6 +24,7 @@ import type { UInt64 } from '@aztec/stdlib/types'; import type { ArchiverDataStores } from '../store/data_stores.js'; import type { L2TipsCache } from '../store/l2_tips_cache.js'; +import type { InboxMessage } from '../structs/inbox_message.js'; /** Operation type for contract data updates. */ enum Operation { @@ -303,17 +304,42 @@ export class ArchiverDataStoreUpdater { return prunedBlocks; } + /** + * Replaces the messages above the last canonical Inbox bucket with the ones the canonical chain delivers, and + * prunes the proposed blocks that consumed a leaf the swap changed, in a single transaction. + * + * Splitting the two would make the prune unrecoverable, for the same reason + * {@link rewindMessagesAndPruneProposedBlocks} keeps them together: the next sync pass would find the local + * messages consistent with L1 and never come back to the blocks built on the leaves that are gone. + * + * @returns The pruned blocks. + */ + public async replaceMessagesAndPruneProposedBlocks(args: { + lastCanonicalBucketSeq: bigint | undefined; + firstDifferingIndex: bigint | undefined; + messages: InboxMessage[]; + syncPoint: L1BlockId; + finalizedL1Block: L1BlockId | undefined; + }): Promise { + const prunedBlocks = await this.stores.db.transactionAsync(async () => { + await this.stores.messages.replaceMessagesAboveBucket(args); + return args.firstDifferingIndex === undefined + ? [] + : await this.removeProposedBlocksConsumingMessagesFrom(args.firstDifferingIndex); + }); + await this.l2TipsCache?.refresh(); + 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. + * The index is the first leaf whose value actually changed, so a reorg that re-mines the same messages in a + * different L1 block prunes nothing, and one that replaces the tail of a bucket keeps the blocks that stopped + * below the replaced leaf. * * 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 From c1ac72653bc84db2d401d63b2a03b3de1220632a Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 19:25:59 -0300 Subject: [PATCH 3/9] feat(fast-inbox): make the inbox rollback content-aware The rollback to the last canonical Inbox bucket now fetches the canonical messages for the range it is about to replace before writing anything, compares them against the ones it holds, and prunes leaves and proposed blocks only from the first leaf whose value actually changed. A reorg that re-mines the same messages in another L1 block costs the bucket metadata and nothing else, and one that replaces the tail of a bucket keeps the blocks that consumed the part of it that survived. The fetch stops at the first batch that covers the last index we hold, so a node that fell far behind does not pull the whole canonical chain into one transaction, and every batch tail it did read is compared against the live chain before the swap commits; a reorg landing mid-fetch throws the messages away without writing. The finalized marker moves with the syncpoint only when it is at or below it, so the rolling-hash fallback can never skip an unsynced range. --- .../archiver/src/archiver-sync.test.ts | 325 +++++++++++++++++- .../archiver/src/modules/l1_synchronizer.ts | 223 +++++++++++- 2 files changed, 527 insertions(+), 21 deletions(-) diff --git a/yarn-project/archiver/src/archiver-sync.test.ts b/yarn-project/archiver/src/archiver-sync.test.ts index 3e60c0e6fc3f..dc1259f90614 100644 --- a/yarn-project/archiver/src/archiver-sync.test.ts +++ b/yarn-project/archiver/src/archiver-sync.test.ts @@ -1847,10 +1847,11 @@ describe('Archiver Sync', () => { expect(countReorgWarnings(warnSpy)).toEqual(1); expect(countWarnings(warnSpy, 'rolling back to bucket 37')).toEqual(1); - // The walk reads one block per orphaned bucket and stops at the first canonical one, which the re-fetch that - // follows checks once more; the buckets below it are never read, and neither are the per-message logs of the - // rolling-hash fallback. - expect(fake.countL1BlockReads(136n)).toEqual(2); + // The walk reads one block per orphaned bucket and stops at the first canonical one; the re-fetch that follows + // checks the block of the last message it read instead. The buckets below the canonical one are never read, + // and neither are the per-message logs of the rolling-hash fallback. + expect(fake.countL1BlockReads(136n)).toEqual(1); + expect(fake.countL1BlockReads(140n)).toEqual(1); expect(fake.countL1BlockReads(134n)).toEqual(0); expect(fake.countL1BlockReads(100n)).toEqual(0); expect(eventByHashSpy).not.toHaveBeenCalled(); @@ -1927,6 +1928,44 @@ describe('Archiver Sync', () => { }); }); + it('does not move the finalized block past the syncpoint of a rollback that stops short of the head', async () => { + const early = [Fr.random(), Fr.random()]; + const late = [Fr.random()]; + fake.addMessages(CheckpointNumber(1), 100n, early); + fake.addMessages(CheckpointNumber(2), 102n, late); + fake.setFinalizedL1BlockNumber(90n); + fake.setL1BlockNumber(110n); + await archiver.syncImmediate(); + expect(await archiverStore.messages.getMessagesFinalizedL1Block()).toMatchObject({ l1BlockNumber: 90n }); + + // Records the finalized block as each syncpoint write leaves it, so the swap's own commit can be told apart + // from the regular fetch that catches up after it within the same pass. + const setMessageSyncState = archiverStore.messages.setMessageSyncState.bind(archiverStore.messages); + const syncStates: { syncPoint: bigint; finalized: bigint | undefined }[] = []; + jest.spyOn(archiverStore.messages, 'setMessageSyncState').mockImplementation(async (l1Block, finalized) => { + await setMessageSyncState(l1Block, finalized); + const stored = await archiverStore.messages.getMessagesFinalizedL1Block(); + syncStates.push({ syncPoint: l1Block.l1BlockNumber, finalized: stored?.l1BlockNumber }); + }); + + // The node comes back a day later: the block its newest bucket sat on was orphaned, and L1 has moved thousands + // of blocks with more messages and a much higher finalized block. The swap only needs the canonical messages + // up to the last index it holds, so it stops at the batch that covers them, well below the head. + fake.retimeMessages(102n, 103n); + fake.addMessages(CheckpointNumber(3), 4000n, [Fr.random()]); + fake.setFinalizedL1BlockNumber(3000n); + fake.setL1BlockNumber(5000n); + await archiver.syncImmediate(); + + // The rolling-hash fallback trusts everything at or below the finalized block without asking L1, so a marker + // above the syncpoint would let it skip a range this node never synced. + expect(syncStates[0]).toEqual({ syncPoint: 2100n, finalized: 90n }); + // The regular fetch resumes from there and advances the marker once it has caught up with the head. + expect(await archiverStore.messages.getSynchedL1Block()).toMatchObject({ l1BlockNumber: 5000n }); + expect(await archiverStore.messages.getMessagesFinalizedL1Block()).toMatchObject({ l1BlockNumber: 3000n }); + expect(await getStoredLeaves()).toHaveLength(4); + }); + it('skips the fetch and fails the pass when the L1 block of the newest bucket cannot be read', async () => { const msgs = [Fr.random(), Fr.random(), Fr.random()]; fake.addMessages(CheckpointNumber(1), 100n, [msgs[0]]); @@ -2140,6 +2179,86 @@ describe('Archiver Sync', () => { expect(await getStoredLeaves()).toHaveLength(2); expect(fake.countL1BlockReads(100n)).toEqual(0); }); + + describe('a reorg landing while the rollback reads the canonical messages', () => { + /** Makes the given reorg land on every log query after the first one of a fetch, so every pass sees it. */ + const reorgAfterFirstLogQuery = (reorg: () => void) => { + const readLogs = inboxContract.getMessageSentEvents.getMockImplementation()!; + inboxContract.getMessageSentEvents.mockImplementation((fromBlock, toBlock) => { + if (fromBlock > firstQueryEnd) { + reorg(); + } + return readLogs(fromBlock, toBlock); + }); + }; + + // Rolls the newest bucket back so the pass swaps the messages above the bucket on L1 block 100, over a range + // wide enough to take more than one log query. Returns the state the swap must leave untouched. + const syncAndReorgNewestBucket = async (lateL1Block: bigint) => { + fake.addMessages(CheckpointNumber(1), 100n, [Fr.random()]); + fake.addMessages(CheckpointNumber(2), lateL1Block, [Fr.random()]); + fake.setFinalizedL1BlockNumber(90n); + fake.setL1BlockNumber(firstQueryEnd + 1000n); + await archiver.syncImmediate(); + + const before = { + leaves: await getStoredLeaves(), + buckets: [await archiverStore.messages.getInboxBucket(1n), await archiverStore.messages.getInboxBucket(2n)], + syncPoint: await archiverStore.messages.getSynchedL1Block(), + }; + return before; + }; + + const expectStoreUntouched = async (before: Awaited>) => { + expect(await getStoredLeaves()).toEqual(before.leaves); + expect(await archiverStore.messages.getInboxBucket(1n)).toEqual(before.buckets[0]); + expect(await archiverStore.messages.getInboxBucket(2n)).toEqual(before.buckets[1]); + expect(await archiverStore.messages.getSynchedL1Block()).toEqual(before.syncPoint); + }; + + it('writes nothing when an earlier batch is orphaned mid-read', async () => { + const before = await syncAndReorgNewestBucket(firstQueryEnd + 500n); + + // Bucket 2 is re-timed, so the rollback re-reads everything from bucket 1's block on. Block 100 is then + // replaced while the second log query of that read is in flight, which leaves the first batch on a fork. + fake.retimeMessages(firstQueryEnd + 500n, firstQueryEnd + 501n); + const warnSpy = jest.spyOn(syncLogger, 'warn'); + reorgAfterFirstLogQuery(() => fake.reorgL1BlocksFrom(100n)); + await expect(archiver.syncImmediate()).rejects.toThrow('Retries exhausted'); + + expect(countWarnings(warnSpy, 'was replaced while they were being read')).toBeGreaterThanOrEqual(1); + await expectStoreUntouched(before); + }); + + it('writes nothing when the last batch is orphaned mid-read', async () => { + const before = await syncAndReorgNewestBucket(firstQueryEnd + 500n); + + // The final batch is the one that decides what is pruned, so it is checked against the live chain too, + // even though the regular fetch leaves it to the next pass. + fake.retimeMessages(firstQueryEnd + 500n, firstQueryEnd + 501n); + const warnSpy = jest.spyOn(syncLogger, 'warn'); + reorgAfterFirstLogQuery(() => fake.reorgL1BlocksFrom(firstQueryEnd + 501n)); + await expect(archiver.syncImmediate()).rejects.toThrow('Retries exhausted'); + + expect(countWarnings(warnSpy, `L1 block ${firstQueryEnd + 501n} was replaced`)).toBeGreaterThanOrEqual(1); + await expectStoreUntouched(before); + }); + + it('writes nothing when the last batch is empty and the batch before it is orphaned mid-read', async () => { + const before = await syncAndReorgNewestBucket(firstQueryEnd + 500n); + + // The canonical chain dropped the message of bucket 2, so the batch covering its L1 block comes back + // empty and the tail of the batch before it is the only one there is to check. + fake.removeMessagesAfter(1); + fake.reorgL1BlocksFrom(firstQueryEnd + 500n); + const warnSpy = jest.spyOn(syncLogger, 'warn'); + reorgAfterFirstLogQuery(() => fake.reorgL1BlocksFrom(100n)); + await expect(archiver.syncImmediate()).rejects.toThrow('Retries exhausted'); + + expect(countWarnings(warnSpy, 'L1 block 100 was replaced')).toBeGreaterThanOrEqual(1); + await expectStoreUntouched(before); + }); + }); }); it('catches up over many buckets after an outage without per-message log lookups', async () => { @@ -2269,6 +2388,204 @@ describe('Archiver Sync', () => { ); }); + it('keeps the proposed chain when the reorg re-mines the same messages', async () => { + const { early, late } = addMessages(); + fake.setL1BlockNumber(l1BlockNumber); + await archiver.syncImmediate(); + + const blocks = await makeBlocksConsumingThrough([2, 4, 4]); + for (const block of blocks) { + await archiver.addBlock(block); + } + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3)); + const bucketBefore = await archiverStore.messages.getInboxBucket(2n); + + // L1 block 102 is orphaned and its two messages are re-mined in block 103, in the same order. Every leaf the + // proposed blocks consumed is still there under the same index, so L1 would accept the checkpoint carrying + // them and the local chain must survive. + fake.retimeMessages(102n, 103n); + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex([...early, ...late])); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3)); + expect(pruneSpy).not.toHaveBeenCalled(); + expect(await archiverStore.messages.getInboxBucket(2n)).toEqual({ + ...bucketBefore, + l1BlockNumber: 103n, + l1BlockHash: fake.getL1BlockHash(103n), + timestamp: fake.getTimestampAtL1Block(103n), + }); + }); + + it('keeps the proposed chain when the reorg merges the bucket it consumed into an earlier one', 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 rolling hash the checkpoint carrying these blocks committed to. + const consumedRollingHash = (await archiverStore.messages.getInboxBucket(2n))!.inboxRollingHash; + + // The messages of block 100 are re-mined into block 102, so all four end up in a single bucket. The leaves + // and their order are untouched, so the blocks that consumed them stay valid. + fake.retimeMessages(100n, 102n); + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex([...early, ...late])); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2)); + expect(pruneSpy).not.toHaveBeenCalled(); + expect(await archiverStore.messages.getInboxBucket(1n)).toMatchObject({ msgCount: 4, l1BlockNumber: 102n }); + expect(await archiverStore.messages.getInboxBucket(2n)).toBeUndefined(); + // The proposer re-resolves its publish hint to the merged bucket, so the sealed header stays publishable. + expect(await archiverStore.messages.getInboxBucketByRollingHash(consumedRollingHash)).toMatchObject({ seq: 1n }); + }); + + it('prunes from the first leaf the reorg changed, not from the start of its bucket', async () => { + const { early, late } = addMessages(); + fake.setL1BlockNumber(l1BlockNumber); + await archiver.syncImmediate(); + + // Block 2 consumed only the first of the two messages of L1 block 102, block 3 consumed both. + const blocks = await makeBlocksConsumingThrough([2, 3, 4]); + for (const block of blocks) { + await archiver.addBlock(block); + } + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3)); + + // L1 block 102 is re-mined carrying only its first message, so index 2 survives and index 3 is gone. + fake.removeMessagesAfter(3); + fake.reorgL1BlocksFrom(102n); + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex([...early, late[0]])); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2)); + expect(pruneSpy).toHaveBeenCalledWith( + expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks: [blocks[2]] }), + ); + }); + + it('prunes from the dropped leaf when a reorg cuts the tail of a bucket spanning two L1 blocks', async () => { + // Blocks 100 and 101 are mined with the same timestamp, so their messages share a bucket spanning both. + fake.shareTimestampWithL1Block(101n, 100n); + const msgs = [Fr.random(), Fr.random(), Fr.random()]; + fake.addMessages(CheckpointNumber(1), 98n, [msgs[0]]); + fake.addMessages(CheckpointNumber(1), 100n, [msgs[1]]); + fake.addMessages(CheckpointNumber(1), 101n, [msgs[2]]); + fake.setL1BlockNumber(l1BlockNumber); + await archiver.syncImmediate(); + expect(await archiverStore.messages.getInboxBucketL1Span(2n)).toMatchObject({ + openedAt: { l1BlockNumber: 100n }, + closedAt: { l1BlockNumber: 101n }, + }); + + const blocks = await makeBlocksConsumingThrough([2, 3]); + for (const block of blocks) { + await archiver.addBlock(block); + } + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2)); + + // Only block 101 is replaced, and it comes back without its message. The bucket keeps its opening block and + // its first message, so the block that stopped at index 1 is untouched. + fake.removeMessagesAfter(2); + fake.reorgL1BlocksFrom(101n); + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex(msgs.slice(0, 2))); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1)); + expect(pruneSpy).toHaveBeenCalledWith( + expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks: [blocks[1]] }), + ); + expect(await archiverStore.messages.getInboxBucket(2n)).toMatchObject({ msgCount: 1, l1BlockNumber: 100n }); + expect(await archiverStore.messages.getInboxBucketL1Span(2n)).toMatchObject({ + openedAt: { l1BlockNumber: 100n }, + closedAt: { l1BlockNumber: 100n }, + }); + }); + + it('keeps the proposed chain when the reorg re-times a bucket with rollover siblings', async () => { + // A single L1 block with more messages than a bucket holds spills into a rollover bucket sharing its opening + // block, so rolling back to the sibling re-downloads both. Neither may be disturbed. + const rollover = times(260, () => Fr.random()); + const later = [Fr.random()]; + fake.addMessages(CheckpointNumber(1), 100n, rollover); + fake.addMessages(CheckpointNumber(1), 102n, later); + fake.setL1BlockNumber(l1BlockNumber); + await archiver.syncImmediate(); + + const firstBucket = await archiverStore.messages.getInboxBucket(1n); + const secondBucket = await archiverStore.messages.getInboxBucket(2n); + expect(firstBucket).toMatchObject({ msgCount: 256, l1BlockNumber: 100n }); + expect(secondBucket).toMatchObject({ msgCount: 4, l1BlockNumber: 100n }); + + const blocks = await makeBlocksConsumingThrough([260, 261]); + for (const block of blocks) { + await archiver.addBlock(block); + } + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2)); + + fake.retimeMessages(102n, 103n); + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex([...rollover, ...later])); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2)); + expect(pruneSpy).not.toHaveBeenCalled(); + expect(await archiverStore.messages.getInboxBucket(1n)).toEqual(firstBucket); + expect(await archiverStore.messages.getInboxBucket(2n)).toEqual(secondBucket); + expect(await archiverStore.messages.getInboxBucket(3n)).toMatchObject({ msgCount: 1, l1BlockNumber: 103n }); + }); + + // The swap has to commit as a whole: a crash between the messages and the syncpoint would leave the store + // holding buckets from the canonical chain behind a syncpoint that says they were never fetched. + it('leaves the store untouched when the swap fails after writing the messages', async () => { + const { early, late } = addMessages(); + fake.setL1BlockNumber(l1BlockNumber); + await archiver.syncImmediate(); + + const blocks = await makeBlocksConsumingThrough([2, 4, 4]); + for (const block of blocks) { + await archiver.addBlock(block); + } + const bucketsBefore = [ + await archiverStore.messages.getInboxBucket(1n), + await archiverStore.messages.getInboxBucket(2n), + ]; + const syncPointBefore = await archiverStore.messages.getSynchedL1Block(); + + const syncPointFailure = new Error('cannot move the syncpoint'); + jest.spyOn(archiverStore.messages, 'setSynchedL1Block').mockImplementationOnce(() => { + throw syncPointFailure; + }); + + fake.retimeMessages(102n, 103n); + await expect(archiver.syncImmediate()).rejects.toThrow(syncPointFailure); + + expect(await getStoredLeaves()).toEqual(asHex([...early, ...late])); + expect(await archiverStore.messages.getTotalL1ToL2MessageCount()).toEqual(4n); + expect(await archiverStore.messages.getInboxBucket(1n)).toEqual(bucketsBefore[0]); + expect(await archiverStore.messages.getInboxBucket(2n)).toEqual(bucketsBefore[1]); + expect((await archiverStore.messages.getLatestInboxBucketAtOrBefore(bucketsBefore[1]!.timestamp))!.seq).toEqual( + 2n, + ); + expect( + (await archiverStore.messages.getInboxBucketByRollingHash(bucketsBefore[1]!.inboxRollingHash))!.seq, + ).toEqual(2n); + expect(await archiverStore.messages.getSynchedL1Block()).toEqual(syncPointBefore); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3)); + expect((await archiver.getL2Tips()).proposed.number).toEqual(BlockNumber(3)); + + // With the store rolled back, the next pass sees the same mismatch and completes the swap. + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex([...early, ...late])); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(3)); + expect(await archiverStore.messages.getInboxBucket(2n)).toMatchObject({ l1BlockNumber: 103n }); + expect(pruneSpy).not.toHaveBeenCalled(); + }); + // 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. diff --git a/yarn-project/archiver/src/modules/l1_synchronizer.ts b/yarn-project/archiver/src/modules/l1_synchronizer.ts index b5d53cf1ef68..1b501df1d4b2 100644 --- a/yarn-project/archiver/src/modules/l1_synchronizer.ts +++ b/yarn-project/archiver/src/modules/l1_synchronizer.ts @@ -459,7 +459,7 @@ export class ArchiverL1Synchronizer implements Traceable { // syncpoint, and we resume from the new one. If the check could not be completed the pass is retried instead: // storing newer buckets on top of one we could not check would make them the ones checked from here on, and the // unchecked one would never be looked at again. - const canonicality = await this.ensureBucketsAreCanonical(messagesSynchedTo, finalizedL1Block); + const canonicality = await this.ensureBucketsAreCanonical(messagesSynchedTo, currentL1Block, finalizedL1Block); if (canonicality.status === 'unverifiable') { return false; } @@ -487,7 +487,7 @@ export class ArchiverL1Synchronizer implements Traceable { // If that's the case, we'd get an exception out of the message store since the rolling hash of the first message // we try to insert would not match the one in the db, in which case we rollback to the last common message with L1. try { - const outcome = await this.retrieveAndStoreMessages(syncFrom, currentL1BlockNumber, finalizedL1Block); + const outcome = await this.retrieveAndStoreMessages(syncFrom, currentL1Block, finalizedL1Block); if (outcome !== 'stored') { return false; } @@ -548,6 +548,7 @@ export class ArchiverL1Synchronizer implements Traceable { */ private async ensureBucketsAreCanonical( messagesSynchedTo: L1BlockId, + currentL1Block: L1BlockId, finalizedL1Block: L1BlockId | undefined, ): Promise { let isNewest = true; @@ -566,7 +567,7 @@ export class ArchiverL1Synchronizer implements Traceable { return { status: 'unverifiable' }; } if (result.status === 'reorged') { - const syncPoint = await this.rollbackReorgedBucket(bucket.seq, result); + const syncPoint = await this.rollbackReorgedBucket(bucket.seq, result, currentL1Block, finalizedL1Block); return syncPoint === undefined ? { status: 'unverifiable' } : { status: 'rolled-back', syncPoint }; } } @@ -610,7 +611,12 @@ export class ArchiverL1Synchronizer implements Traceable { * the store is rolled back to the genesis sentinel. Returns the new message syncpoint, or undefined without rolling * anything back if an L1 block could not be read. */ - private async rollbackReorgedBucket(reorgedBucketSeq: bigint, reorg: BucketReorg): Promise { + private async rollbackReorgedBucket( + reorgedBucketSeq: bigint, + reorg: BucketReorg, + currentL1Block: L1BlockId, + finalizedL1Block: L1BlockId | undefined, + ): Promise { const bucketsBelow = this.stores.messages.iterateInboxBucketL1Spans({ end: reorgedBucketSeq - 1n, reverse: true }); for await (const bucket of bucketsBelow) { const result = await this.checkBucketCanonicality(bucket); @@ -619,10 +625,10 @@ export class ArchiverL1Synchronizer implements Traceable { return undefined; } if (result.status === 'canonical') { - return await this.rollbackToCanonicalBucket(reorgedBucketSeq, bucket, reorg); + return await this.rollbackToCanonicalBucket(reorgedBucketSeq, bucket, reorg, currentL1Block, finalizedL1Block); } } - return await this.rollbackToCanonicalBucket(reorgedBucketSeq, undefined, reorg); + return await this.rollbackToCanonicalBucket(reorgedBucketSeq, undefined, reorg, currentL1Block, finalizedL1Block); } /** @@ -645,14 +651,24 @@ export class ArchiverL1Synchronizer implements Traceable { } /** - * Drops every message absorbed after the last canonical bucket and rewinds the syncpoint so the reorged L1 blocks - * are downloaded again, with the bucket timestamps and sequence numbers the canonical chain gives them. + * Replaces every message above the last canonical bucket with the ones the canonical chain holds for the same L1 + * range, dropping leaves and the proposed blocks built on them only from the first leaf whose value changed. + * + * The canonical messages are fetched before anything is written, so the swap knows what the replacement blocks + * carry rather than assuming they carry nothing. A reorg that re-mines the same messages in a different L1 block + * then costs only the bucket metadata derived from that block, and one that replaces the tail of a bucket keeps + * the blocks that consumed the part of it that survived. Nothing is written until the fetch has been checked + * against the live chain, so a failed check leaves the store exactly as it was and the next pass starts over. + * + * Returns the new message syncpoint, or undefined without writing anything if the fetch could not be verified. */ private async rollbackToCanonicalBucket( reorgedBucketSeq: bigint, lastCanonicalBucket: InboxBucketL1Span | undefined, reorg: BucketReorg, - ): Promise { + currentL1Block: L1BlockId, + finalizedL1Block: L1BlockId | undefined, + ): Promise { this.log.warn( `Inbox bucket ${reorgedBucketSeq} sits on an L1 block that is no longer canonical; rolling back to bucket ${lastCanonicalBucket?.seq ?? 0n}`, { @@ -664,13 +680,185 @@ export class ArchiverL1Synchronizer implements Traceable { }, ); - // Resume from the block before the last canonical bucket's opening one, so that bucket is re-delivered from its - // first message and any bucket that rolled over after it within the same L1 block is downloaded again. The - // syncpoint is resolved before anything is written, so the removal and the rewind can commit together. - const syncPoint = await this.resolveMessagesSyncPoint( - lastCanonicalBucket && lastCanonicalBucket.openedAt.l1BlockNumber - 1n, + // Resume from the last canonical bucket's opening block, so that bucket is re-delivered from its first message + // and any bucket that rolled over after it within the same L1 block is downloaded again. + const fetchFromL1Block = maxBigint( + ...compactArray([lastCanonicalBucket?.openedAt.l1BlockNumber, this.l1Constants.l1StartBlock + 1n]), + ); + const lastStoredIndex = (await this.stores.messages.getLastMessage())?.index; + const fetched = await this.fetchCanonicalMessages( + fetchFromL1Block, + currentL1Block, + finalizedL1Block, + lastStoredIndex, ); - return this.rewindMessagesTo(syncPoint, lastCanonicalBucket ? lastCanonicalBucket.lastMessageIndex + 1n : 0n); + if (fetched === undefined) { + return undefined; + } + + // The last canonical bucket's own messages need no comparison: both of the L1 blocks it sits on were just + // checked against the live chain, so nothing below its last message can have changed. + const firstDifferingIndex = await this.findFirstDifferingMessageIndex( + lastCanonicalBucket ? lastCanonicalBucket.lastMessageIndex + 1n : 0n, + fetched.messages, + ); + + const prunedBlocks = await this.updater.replaceMessagesAndPruneProposedBlocks({ + lastCanonicalBucketSeq: lastCanonicalBucket?.seq, + firstDifferingIndex, + messages: fetched.messages, + syncPoint: fetched.syncPoint, + // The rolling-hash fallback trusts every message at or below the finalized marker without querying L1, so the + // marker must never be advanced past the syncpoint the swap commits. + finalizedL1Block: + finalizedL1Block && finalizedL1Block.l1BlockNumber <= fetched.syncPoint.l1BlockNumber + ? finalizedL1Block + : undefined, + }); + this.log.verbose(`Updated messages syncpoint to L1 block ${fetched.syncPoint.l1BlockNumber}`, { + ...fetched.syncPoint, + firstDifferingIndex, + messageCount: fetched.messages.length, + }); + this.reportPrunedProposedBlocks(prunedBlocks, firstDifferingIndex); + return fetched.syncPoint; + } + + /** + * Retrieves the canonical L1 to L2 messages from the given L1 block on, in the same whole-L1-block batches the + * regular retrieval uses, without writing any of them. Returns them with the L1 block they were fetched up to, or + * undefined when the fetch could not be checked against the live chain. + * + * The fetch stops at the first batch that reaches the last index we hold, since that is all the comparison needs: + * a node that fell far behind an orphaned bucket would otherwise pull every message the canonical chain added + * since into memory and commit them in one transaction. Whatever is left is picked up by the regular fetch on the + * next pass, which resumes from the syncpoint this one commits. + * + * Every batch that returned messages has the L1 block of its last message compared against the live chain, + * skipping what is at or below the finalized block. The regular retrieval leaves the final batch out because it + * becomes the anchor the next pass checks; here the final batch decides what is pruned, so it is checked now. + */ + private async fetchCanonicalMessages( + fromL1Block: bigint, + currentL1Block: L1BlockId, + finalizedL1Block: L1BlockId | undefined, + lastStoredIndex: bigint | undefined, + ): Promise<{ messages: InboxMessage[]; syncPoint: L1BlockId } | undefined> { + const toL1Block = currentL1Block.l1BlockNumber; + const messages: InboxMessage[] = []; + const batchTails: InboxMessage[] = []; + let searchEndBlock = fromL1Block - 1n; + let isLastBatch = false; + + while (!isLastBatch) { + const [searchStartBlock, batchEndBlock] = this.nextRange(searchEndBlock, toL1Block); + this.log.trace(`Retrieving canonical L1 to L2 messages in L1 blocks ${searchStartBlock}-${batchEndBlock}`); + const batch = await retrieveL1ToL2Messages(this.inbox, searchStartBlock, batchEndBlock); + messages.push(...batch); + searchEndBlock = batchEndBlock; + isLastBatch = batchEndBlock >= toL1Block; + + const batchTail = batch.at(-1); + if (batchTail !== undefined) { + batchTails.push(batchTail); + } + if (isLastBatch || batchTail === undefined) { + continue; + } + if (lastStoredIndex !== undefined && batchTail.index < lastStoredIndex) { + continue; + } + // A bucket that spans co-timestamped L1 blocks can still be open at the end of the batch that covered our last + // stored index, and committing there would write a snapshot that undercounts it, so wait for it to close. + const bucketClosed = await this.isBucketClosedAfterL1Block(batchEndBlock, batchTail); + if (bucketClosed === undefined) { + return undefined; + } + if (bucketClosed) { + break; + } + } + + for (const tail of batchTails) { + if (isFinalized(tail, finalizedL1Block)) { + continue; + } + const canonicality = await this.compareL1BlocksWithChain([ + { l1BlockNumber: tail.l1BlockNumber, l1BlockHash: tail.l1BlockHash }, + ]); + if (canonicality.status === 'unverifiable') { + this.warnBucketNotVerifiable(tail.bucketSeq, canonicality.l1BlockNumber, canonicality.error); + return undefined; + } + if (canonicality.status === 'reorged') { + this.log.warn( + `Discarding the canonical L1 to L2 messages fetched to roll back to: L1 block ${tail.l1BlockNumber} was replaced while they were being read`, + { + l1BlockNumber: tail.l1BlockNumber, + storedL1BlockHash: tail.l1BlockHash.toString(), + liveL1BlockHash: canonicality.liveL1BlockHash.toString(), + }, + ); + return undefined; + } + } + + if (isLastBatch) { + return { messages, syncPoint: currentL1Block }; + } + try { + return { + messages, + syncPoint: { l1BlockNumber: searchEndBlock, l1BlockHash: await this.getL1BlockHash(searchEndBlock) }, + }; + } catch (error) { + this.warnBucketNotVerifiable(messages.at(-1)!.bucketSeq, searchEndBlock, error); + return undefined; + } + } + + /** + * Whether the Inbox bucket the given message was absorbed into is closed at the given L1 block, that is, whether + * the block after it carries a different timestamp and so can absorb nothing more into that bucket. Undefined when + * the block cannot be read: a block we know nothing about never counts as closing a bucket. + */ + private async isBucketClosedAfterL1Block(l1BlockNumber: bigint, message: InboxMessage): Promise { + const nextL1BlockNumber = l1BlockNumber + 1n; + try { + const block = await this.publicClient.getBlock({ blockNumber: nextL1BlockNumber, includeTransactions: false }); + if (!block) { + throw new Error(`Missing L1 block ${nextL1BlockNumber}`); + } + return block.timestamp !== message.bucketTimestamp; + } catch (error) { + this.log.warn( + `Failed to read L1 block ${nextL1BlockNumber} to check whether Inbox bucket ${message.bucketSeq} is closed`, + { bucketSeq: message.bucketSeq, l1BlockNumber: nextL1BlockNumber, error }, + ); + return undefined; + } + } + + /** + * Returns the lowest index at or above `fromIndex` where the messages we hold and the canonical ones disagree, or + * undefined when every stored message from there on is re-delivered unchanged. A message that the canonical chain + * does not carry at all counts as a difference. + * + * Comparing consensus rolling hashes is enough to compare whole prefixes: each one chains the hash before it, so + * equal hashes at an index mean equal leaves at every index through it. + */ + private async findFirstDifferingMessageIndex( + fromIndex: bigint, + canonicalMessages: InboxMessage[], + ): Promise { + const canonicalByIndex = new Map(canonicalMessages.map(message => [message.index, message])); + for await (const stored of this.stores.messages.iterateL1ToL2Messages({ start: fromIndex })) { + const canonical = canonicalByIndex.get(stored.index); + if (canonical === undefined || !canonical.inboxRollingHash.equals(stored.inboxRollingHash)) { + return stored.index; + } + } + return undefined; } /** @@ -760,9 +948,10 @@ export class ArchiverL1Synchronizer implements Traceable { */ private async retrieveAndStoreMessages( syncFrom: L1BlockId, - toL1Block: bigint, + currentL1Block: L1BlockId, finalizedL1Block: L1BlockId | undefined, ): Promise { + const toL1Block = currentL1Block.l1BlockNumber; const newestBucketBeforeFetch = await this.stores.messages.getNewestInboxBucketL1Span(); let searchStartBlock: bigint = 0n; let searchEndBlock: bigint = syncFrom.l1BlockNumber; @@ -823,7 +1012,7 @@ export class ArchiverL1Synchronizer implements Traceable { continue; } if (canonicality.status === 'reorged') { - const syncPoint = await this.rollbackReorgedBucket(bucketSeq, canonicality); + const syncPoint = await this.rollbackReorgedBucket(bucketSeq, canonicality, currentL1Block, finalizedL1Block); if (syncPoint !== undefined) { return 'rolled-back'; } From 7bbbd324515bb890a3493cc6c47d09afe6a08bc4 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 19:55:39 -0300 Subject: [PATCH 4/9] feat(fast-inbox): abandon the slot when the archiver dropped the checkpoint's blocks Attestation collection takes seconds, and the archiver can prune the proposed chain in that window: a slot that closed without a checkpoint, or a checkpoint seen on L1 that conflicts with the local blocks. The publish path now reads the checkpoint's last block back from the archiver and compares its hash, rather than publishing a checkpoint this node cannot serve. The test doubles gain an archiver that serves back what the proposer pushed to it, and the mock checkpoint builder no longer hands back seeded blocks it was never asked to build. --- .../sequencer/checkpoint_proposal_job.test.ts | 58 ++++++++++++++++++- .../checkpoint_proposal_job.timing.test.ts | 3 +- .../src/sequencer/checkpoint_proposal_job.ts | 40 ++++++++++++- .../src/sequencer/sequencer.test.ts | 23 +++++--- .../src/test/mock_checkpoint_builder.ts | 12 ++-- .../sequencer-client/src/test/utils.ts | 39 ++++++++++++- 6 files changed, 158 insertions(+), 17 deletions(-) diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts index 8d579e932ffa..f9f42bbebbf4 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts @@ -16,7 +16,7 @@ import { Signature } from '@aztec/foundation/eth-signature'; import type { Logger } from '@aztec/foundation/log'; import { promiseWithResolvers } from '@aztec/foundation/promise'; import { ManualDateProvider } from '@aztec/foundation/timer'; -import type { TypedEventEmitter } from '@aztec/foundation/types'; +import { type TypedEventEmitter, unfreeze } from '@aztec/foundation/types'; import { type P2P, P2PClientState } from '@aztec/p2p'; import type { SlasherClientInterface } from '@aztec/slasher'; import { AztecAddress } from '@aztec/stdlib/aztec-address'; @@ -73,6 +73,7 @@ import { mockInboxBuckets, mockPendingTxs, mockTxIterator, + serveAddedBlocksFromSource, setupTxsAndBlock, } from '../test/utils.js'; import { CheckpointProposalJob } from './checkpoint_proposal_job.js'; @@ -107,6 +108,8 @@ describe('CheckpointProposalJob', () => { let l1ToL2MessageSource: MockProxy; let l2BlockSource: MockProxy; let blockSink: MockProxy; + /** Blocks the job pushed to the archiver, which serves them back until a test prunes them. */ + let localBlocks: Map; let slasherClient: MockProxy; let dateProvider: ManualDateProvider; let metrics: MockProxy; @@ -300,7 +303,7 @@ describe('CheckpointProposalJob', () => { }); blockSink = mock(); - blockSink.addBlock.mockResolvedValue(undefined); + localBlocks = serveAddedBlocksFromSource(l2BlockSource, blockSink); blockSink.addProposedCheckpoint.mockResolvedValue(undefined); validatorClient = mock(); @@ -1838,6 +1841,57 @@ describe('CheckpointProposalJob', () => { }); }); + describe('publishing a checkpoint whose blocks the archiver dropped', () => { + // Attestation collection takes seconds, and the archiver can prune the proposed chain in that window: a slot + // that closed without a checkpoint, or a checkpoint seen on L1 that conflicts with the local blocks. Publishing + // then puts a checkpoint on L1 that this node's own archiver cannot serve. + beforeEach(() => { + job.setTimetable(makeSingleBlockTimetable()); + dateProvider.setTime(buildFrameStartSeconds() * 1000); + }); + + const setupOneBlockCheckpoint = async () => { + const { txs, block } = await setupTxsAndBlock(p2p, globalVariables, 2, chainId); + checkpointBuilder.seedBlocks([block], [txs]); + return block; + }; + + it('does not submit when the archiver no longer holds the checkpoint blocks', async () => { + const block = await setupOneBlockCheckpoint(); + validatorClient.collectAttestations.mockImplementation(() => { + localBlocks.clear(); + return Promise.resolve(getAttestations(block)); + }); + + const checkpoint = await job.executeAndAwait(); + + // The proposal was already gossiped when the prune landed, but nothing reaches L1. + expect(checkpoint).toBeDefined(); + expect(p2p.broadcastCheckpointProposal).toHaveBeenCalledTimes(1); + expect(publisher.enqueueProposeCheckpoint).not.toHaveBeenCalled(); + expect(metrics.recordCheckpointProposalFailed).toHaveBeenCalledWith('checkpoint_blocks_pruned'); + }); + + it('does not submit when the archiver rebuilt the block number with a different block', async () => { + const block = await setupOneBlockCheckpoint(); + const replacement = L2Block.fromBuffer(block.toBuffer()); + unfreeze(replacement.header).spongeBlobHash = Fr.random(); + await replacement.header.recomputeHash(); + validatorClient.collectAttestations.mockImplementation(() => { + // The prune took the block and the chain was rebuilt to the same height with a different one, so the block + // number alone would say the checkpoint is still local. + localBlocks.set(block.number, replacement); + return Promise.resolve(getAttestations(block)); + }); + + const checkpoint = await job.executeAndAwait(); + + expect(checkpoint).toBeDefined(); + expect(publisher.enqueueProposeCheckpoint).not.toHaveBeenCalled(); + expect(metrics.recordCheckpointProposalFailed).toHaveBeenCalledWith('checkpoint_blocks_pruned'); + }); + }); + describe('build single block', () => { it('does not build a block if not enough valid txs are collected', async () => { // We have enough txs, but not enough valid ones diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.timing.test.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.timing.test.ts index 13731c62a848..e92c3b0dcf91 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.timing.test.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.timing.test.ts @@ -47,6 +47,7 @@ import { makeTx, mockInboxBuckets, mockTxIterator, + serveAddedBlocksFromSource, } from '../test/utils.js'; import { CheckpointProposalJob } from './checkpoint_proposal_job.js'; import type { CheckpointProposalJobMetricsRecorder } from './checkpoint_proposal_job_metrics.js'; @@ -498,7 +499,7 @@ describe('CheckpointProposalJob Timing Tests', () => { }); blockSink = mock(); - blockSink.addBlock.mockResolvedValue(undefined); + serveAddedBlocksFromSource(l2BlockSource, blockSink); validatorClient = mock(); validatorClient.collectAttestations.mockImplementation(() => Promise.resolve([])); diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts index bd9814828590..525c52e223ae 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts @@ -379,7 +379,7 @@ export class CheckpointProposalJob implements Traceable { if (signedAttestations && (await this.waitForValidParentCheckpointOnL1())) { // Attestation collection took seconds; re-resolve the hint once more in case L1 reorged in that window. const bucketHint = await this.resolveBucketHint(broadcast.streamingState, checkpoint.header, 'publish-failed'); - if (bucketHint !== undefined) { + if (bucketHint !== undefined && (await this.checkpointBlocksAreStillLocal(checkpoint))) { await this.enqueueCheckpointForSubmission({ checkpoint, ...signedAttestations, bucketHint }); } } @@ -450,6 +450,44 @@ export class CheckpointProposalJob implements Traceable { } } + /** + * Whether the local archiver still holds the last block of the checkpoint about to be published, under the hash it + * was built with. Attestation collection takes seconds, and the archiver can drop the proposed blocks in that + * window: the end-of-slot prune takes every block of a slot that closed without a checkpoint, and a checkpoint seen + * on L1 that conflicts with the local chain prunes the blocks it disagrees with. Publishing blocks this node no + * longer holds would put a checkpoint on L1 that its own archiver cannot serve. + * + * The hash is compared rather than the height alone, since a chain rebuilt after a prune reuses the block numbers. + */ + private async checkpointBlocksAreStillLocal(checkpoint: Checkpoint): Promise { + const lastBlock = checkpoint.blocks.at(-1); + if (lastBlock === undefined) { + return true; + } + const blockHash = await lastBlock.hash(); + const local = await this.l2BlockSource.getBlockData({ number: lastBlock.number }); + if (local !== undefined && local.blockHash.equals(blockHash)) { + return true; + } + + const context = { + slot: this.targetSlot, + checkpointNumber: this.checkpointNumber, + blockNumber: lastBlock.number, + blockHash: blockHash.toString(), + localBlockHash: local?.blockHash.toString(), + reason: 'checkpoint_blocks_pruned', + }; + this.logCheckpointEvent('publish-failed', `Checkpoint publish failed for slot ${this.targetSlot}`, context); + this.log.warn( + `The local archiver no longer holds the blocks of this checkpoint; abandoning slot ${this.targetSlot} rather ` + + `than publishing a checkpoint this node cannot serve`, + context, + ); + this.metrics.recordCheckpointProposalFailed('checkpoint_blocks_pruned'); + return false; + } + /** Enqueues the checkpoint for L1 submission. Called after pipeline sleep in execute(). */ private async enqueueCheckpointForSubmission(result: CheckpointProposalResult): Promise { const { checkpoint, attestations, attestationsSignature, bucketHint } = result; diff --git a/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts b/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts index bcd463b78fec..e04e6dc5ccb3 100644 --- a/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts +++ b/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts @@ -54,7 +54,12 @@ import { type MockProxy, mock, mockDeep, mockFn } from 'jest-mock-extended'; import type { GlobalVariableBuilder } from '../global_variable_builder/global_builder.js'; import type { AttestorPublisherPair, SequencerPublisherFactory } from '../publisher/sequencer-publisher-factory.js'; import type { InvalidateCheckpointRequest, SequencerPublisher } from '../publisher/sequencer-publisher.js'; -import { MockCheckpointBuilder, MockCheckpointsBuilder, mockInboxBuckets } from '../test/utils.js'; +import { + MockCheckpointBuilder, + MockCheckpointsBuilder, + mockInboxBuckets, + serveAddedBlocksFromSource, +} from '../test/utils.js'; import * as TestUtils from '../test/utils.js'; import { Sequencer } from './sequencer.js'; import { SequencerState } from './utils.js'; @@ -317,13 +322,6 @@ describe('sequencer', () => { checkpointBuilder.setBlockProvider(() => block); l2BlockSource = mock({ - getBlockData: mockFn().mockResolvedValue({ - header: BlockHeader.empty(), - archive: AppendOnlyTreeSnapshot.empty(), - blockHash: BlockHash.ZERO, - checkpointNumber: CheckpointNumber(0), - indexWithinCheckpoint: IndexWithinCheckpoint(0), - } satisfies BlockData), getBlockNumber: mockFn().mockResolvedValue(lastBlockNumber), getL2Tips: mockFn().mockResolvedValue({ proposed: { number: lastBlockNumber, hash }, @@ -347,6 +345,15 @@ describe('sequencer', () => { getSyncedL2SlotNumber: mockFn().mockResolvedValue(SlotNumber(Number.MAX_SAFE_INTEGER)), getProposedCheckpointData: mockFn().mockResolvedValue(undefined), }); + // The proposer pushes each block it builds to the archiver and reads the last one back before publishing; + // anything it never built (the sync check's own tip lookups) falls back to the empty block data. + serveAddedBlocksFromSource(l2BlockSource, l2BlockSource, { + header: BlockHeader.empty(), + archive: AppendOnlyTreeSnapshot.empty(), + blockHash: BlockHash.ZERO, + checkpointNumber: CheckpointNumber(0), + indexWithinCheckpoint: IndexWithinCheckpoint(0), + }); l1ToL2MessageSource = mock({ getL2Tips: mockFn().mockResolvedValue({ diff --git a/yarn-project/sequencer-client/src/test/mock_checkpoint_builder.ts b/yarn-project/sequencer-client/src/test/mock_checkpoint_builder.ts index 93f4deeccd54..b18c7886dc7f 100644 --- a/yarn-project/sequencer-client/src/test/mock_checkpoint_builder.ts +++ b/yarn-project/sequencer-client/src/test/mock_checkpoint_builder.ts @@ -101,11 +101,14 @@ export class MockCheckpointBuilder implements ICheckpointBlockBuilder { usedTxs = []; this.builtBlocks.push(block); } else { - // Seeded mode: get block from pre-seeded list + // Seeded mode: get block from pre-seeded list. A caller that asks for more blocks than were seeded gets + // undefined, which the job reports as a failed build; it never becomes part of the checkpoint. block = this.blocks[this.blockIndex]; usedTxs = this.usedTxsPerBlock[this.blockIndex] ?? []; this.blockIndex++; - this.builtBlocks.push(block); + if (block !== undefined) { + this.builtBlocks.push(block); + } } // Check that no pending tx has already been consumed @@ -132,8 +135,9 @@ export class MockCheckpointBuilder implements ICheckpointBlockBuilder { completeCheckpoint(): Promise { this.completeCheckpointCalled = true; - const allBlocks = this.blockProvider ? this.builtBlocks : this.blocks; - return this.buildCheckpoint(allBlocks); + // Only the blocks that were actually built, so a checkpoint never carries a seeded block the caller stopped + // short of and never pushed to the archiver. + return this.buildCheckpoint(this.builtBlocks); } getCheckpoint(): Promise { diff --git a/yarn-project/sequencer-client/src/test/utils.ts b/yarn-project/sequencer-client/src/test/utils.ts index 43e1cfda60e9..b22f3fbf0a1e 100644 --- a/yarn-project/sequencer-client/src/test/utils.ts +++ b/yarn-project/sequencer-client/src/test/utils.ts @@ -8,7 +8,13 @@ import type { EthAddress } from '@aztec/foundation/eth-address'; import { Signature } from '@aztec/foundation/eth-signature'; import type { P2P } from '@aztec/p2p'; import { PublicDataWrite } from '@aztec/stdlib/avm'; -import { CommitteeAttestation, L2Block } from '@aztec/stdlib/block'; +import { + type BlockData, + CommitteeAttestation, + L2Block, + type L2BlockSink, + type L2BlockSource, +} from '@aztec/stdlib/block'; import { DEFAULT_BLOCK_DURATION_MS } from '@aztec/stdlib/config'; import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers'; import type { InboxBucket, L1ToL2MessageSource } from '@aztec/stdlib/messaging'; @@ -245,3 +251,34 @@ export function mockInboxBuckets(source: MockProxy, buckets Promise.resolve(all.find(bucket => bucket.inboxRollingHash.equals(inboxRollingHash))), ); } + +/** + * Serves the blocks pushed to the sink back through the block source, the way an archiver that has not pruned + * anything does: the proposer pushes each block it builds and reads the last one back before publishing. Returns the + * map behind it, so a test can drop or replace an entry to stand in for a prune. + */ +export function serveAddedBlocksFromSource( + blockSource: MockProxy, + blockSink: MockProxy, + fallback?: BlockData, +): Map { + const blocks = new Map(); + blockSink.addBlock.mockImplementation(block => { + blocks.set(block.number, block); + return Promise.resolve(); + }); + blockSource.getBlockData.mockImplementation(async query => { + const block = 'number' in query ? blocks.get(query.number) : undefined; + if (block === undefined) { + return fallback; + } + return { + header: block.header, + archive: block.archive, + blockHash: await block.hash(), + checkpointNumber: block.checkpointNumber, + indexWithinCheckpoint: block.indexWithinCheckpoint, + }; + }); + return blocks; +} From 40441ab0ba639ec5c9665fed793da9ff94365539 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 20:12:00 -0300 Subject: [PATCH 5/9] test(fast-inbox): cover the reorg that re-mines the same inbox message end to end Withholds the propose the way the pruning test does, then rewinds L1 and replays both the message and the propose: the message keeps its leaf and index, so the bucket only moves to the replacement L1 block, the proposed chain that consumed it survives, and the withheld checkpoint still publishes and promotes the very blocks it was built on. --- .../cross-chain/streaming_inbox_reorg.test.ts | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts b/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts index 278ce82f5f72..23103060007a 100644 --- a/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts +++ b/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts @@ -5,15 +5,19 @@ import { Fr } from '@aztec/aztec.js/fields'; import type { Logger } from '@aztec/aztec.js/log'; import { isL1ToL2MessageReady, waitForL1ToL2MessageReady } from '@aztec/aztec.js/messaging'; import type { AztecNode } from '@aztec/aztec.js/node'; +import { createBlobClient } from '@aztec/blob-client/client'; +import { Blob } from '@aztec/blob-lib'; import type { Delayer } from '@aztec/ethereum/l1-tx-utils'; import type { ChainMonitor } from '@aztec/ethereum/test'; import type { ExtendedViemWalletClient } from '@aztec/ethereum/types'; import { CheckpointNumber } from '@aztec/foundation/branded-types'; import { retryUntil } from '@aztec/foundation/retry'; +import { hexToBuffer } from '@aztec/foundation/string'; import { L2BlockSourceEvents, type L2PruneUncheckpointedEvent } from '@aztec/stdlib/block'; import { WorldStateSynchronizerError } from '@aztec/world-state'; import 'jest-extended'; +import { parseTransaction } from 'viem'; import { sendL1ToL2Message } from '../../fixtures/l1_to_l2_messaging.js'; import type { EndToEndContext } from '../../fixtures/utils.js'; @@ -206,4 +210,109 @@ describe('single-node/cross-chain/streaming_inbox_reorg', () => { expect(rebuiltCheckpoints.some(published => published.checkpoint.blocks.length >= 2)).toBe(true); await test.assertMultipleBlocksPerSlot(2); }); + + it('keeps the proposed chain when a reorg re-mines the same message', async () => { + // Get the chain building multi-block checkpoints before touching the Inbox. + await t.sendTransactions(TX_COUNT, 300); + await test.waitUntilCheckpointNumber(CheckpointNumber(2), L2_SLOT_DURATION_IN_S * 6); + + const prunes: L2PruneUncheckpointedEvent[] = []; + archiver.events.on(L2BlockSourceEvents.L2PruneUncheckpointed, event => { + prunes.push(event); + }); + + // Withhold the next checkpoint's propose tx, so the blocks that consume the message below are still only + // proposed when the reorg lands. Both it and the message go back into the replacement chain further down. + sequencerDelayer.cancelNextTx(); + + const msg = await sendMessage(l1Client); + logger.warn(`Sent message on L1 block ${msg.txReceipt.blockNumber}`); + + await waitForL1ToL2MessageReady(node, msg.msgHash, { timeoutSeconds: L2_SLOT_DURATION_IN_S * 2 }); + const consumedAtBlockNumber = await archiver.getBlockNumber(); + const consumedBlock = await archiver.getBlockData({ number: consumedAtBlockNumber }); + const bucketBeforeReorg = (await archiver.dataStores.messages.getNewestInboxBucket())!; + logger.warn(`Message consumed by proposed block ${consumedAtBlockNumber}`, { + bucketSeq: bucketBeforeReorg.seq, + bucketL1BlockNumber: bucketBeforeReorg.l1BlockNumber, + }); + + // Same check as the pruning test: the cancellation is armed before the message is sent, so make sure it took + // the propose that covers the consuming block rather than the previous slot's. + await retryUntil( + () => sequencerDelayer.getCancelledTxs().length, + 'sequencer propose tx withheld', + L2_SLOT_DURATION_IN_S * 2, + 0.2, + ); + const [proposeTx] = sequencerDelayer.getCancelledTxs(); + const tipsBeforeReorg = await archiver.getL2Tips(); + expect(consumedAtBlockNumber).toBeGreaterThan(tipsBeforeReorg.checkpointed.block.number); + + // The signed bytes of the mined message tx, read while it is still on the chain. The reorg rewinds the + // sender's nonce, so the very same tx is valid again in the replacement block and keeps its hash: the leaf it + // inserts hashes the sender, recipient, content, secret hash and Inbox index, none of which depends on the L1 + // block it lands in. + const rawMessageTx = await context.cheatCodes.eth.getRawTransaction(msg.txReceipt.transactionHash); + + // Replace every L1 block from the one carrying the message onwards, re-mining the message in the first + // replacement block. Both txs are replayed by hand rather than handed to `reorgWithReplacement`, which + // silently drops the blob-carrying propose. + const l1BlockNumber = await monitor.run(true).then(m => m.l1BlockNumber); + const reorgDepth = l1BlockNumber - Number(msg.txReceipt.blockNumber) + 1; + expect(reorgDepth).toBeGreaterThanOrEqual(2); + logger.warn(`Triggering reorg of depth ${reorgDepth} re-mining the same message`); + await context.cheatCodes.eth.reorg(reorgDepth); + await l1Client.sendRawTransaction({ serializedTransaction: rawMessageTx }); + await context.cheatCodes.eth.mine(reorgDepth); + + // The rollback rewound L1 by a slot's worth of blocks, and the propose is only valid inside the slot it was + // built for, so it goes back into the mempool and the next L1 block mined at its own pace carries it. + const proposeTxHash = await l1Client.sendRawTransaction({ serializedTransaction: proposeTx }); + const proposeReceipt = await l1Client.waitForTransactionReceipt({ + hash: proposeTxHash, + timeout: L1_BLOCK_TIME_IN_S * 4 * 1000, + }); + + // The node reads the checkpoint's blocks off the blob sink, which never saw the replayed propose. + const blobs = await Promise.all( + (parseTransaction(proposeTx).sidecars || []).map(sidecar => Blob.fromBlobBuffer(hexToBuffer(sidecar.blob))), + ); + await createBlobClient(context.config).sendBlobsToFilestore(blobs); + + // The bucket the consuming block built on moves to the replacement L1 block, keeping the sequence number and + // the rolling hash the sealed checkpoint header commits to. That is what makes the checkpoint publishable. + await retryUntil( + async () => { + const bucket = await archiver.dataStores.messages.getInboxBucket(bucketBeforeReorg.seq); + return bucket !== undefined && !bucket.l1BlockHash.equals(bucketBeforeReorg.l1BlockHash); + }, + 'inbox bucket re-timed onto the replacement L1 block', + L2_SLOT_DURATION_IN_S, + 0.2, + ); + const bucketAfterReorg = (await archiver.dataStores.messages.getInboxBucket(bucketBeforeReorg.seq))!; + expect(bucketAfterReorg.inboxRollingHash).toEqual(bucketBeforeReorg.inboxRollingHash); + expect(await archiver.dataStores.messages.getInboxBucketByRollingHash(bucketBeforeReorg.inboxRollingHash)).toEqual( + bucketAfterReorg, + ); + + // Nothing the reorg did dropped a block that consumed the message, and the message is still consumable. + expect(prunes.flatMap(prune => prune.blocks).map(block => block.number)).not.toContain(consumedAtBlockNumber); + expect(await archiver.getBlockNumber()).toBeGreaterThanOrEqual(consumedAtBlockNumber); + expect(await isL1ToL2MessageReady(node, msg.msgHash)).toBe(true); + + // The replayed propose reached L1, and the archiver promoted the very blocks it had proposed rather than + // rebuilding them. + expect(proposeReceipt.status).toEqual('success'); + await retryUntil( + async () => (await archiver.getL2Tips()).checkpointed.block.number >= consumedAtBlockNumber, + 'consuming block promoted to checkpointed', + L2_SLOT_DURATION_IN_S * 2, + 0.2, + ); + expect((await archiver.getBlockData({ number: consumedAtBlockNumber }))!.blockHash).toEqual( + consumedBlock!.blockHash, + ); + }); }); From e6aba30202f011c2ef6017d344acffcfc10def10 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 20:12:37 -0300 Subject: [PATCH 6/9] docs(fast-inbox): say that a reorg re-mining the same messages costs no slot The inbox rollback is content-aware now, so immediate consumption only loses a slot when a reorg actually reorders or drops a message. --- docs/docs-operate/operators/reference/changelog/v6.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/docs-operate/operators/reference/changelog/v6.md b/docs/docs-operate/operators/reference/changelog/v6.md index 357d9b7d14dd..688eb4621f3d 100644 --- a/docs/docs-operate/operators/reference/changelog/v6.md +++ b/docs/docs-operate/operators/reference/changelog/v6.md @@ -16,7 +16,7 @@ displayed_sidebar: operatorsSidebar ### Choosing a value -`0` consumes a message as soon as your archiver sees it, so a bridged message reaches L2 in the next block your node builds. The cost is that an L1 reorg touching an L1 block whose messages your node has already consumed costs your node that slot, whether or not the messages themselves changed: the archiver rewinds the messages from that block on and drops the blocks built on them, and the checkpoint is abandoned. Your node also re-resolves the bucket it consumed by its contents before publishing, but that only prevents a false rejection when a reorg elsewhere renumbers buckets without touching their messages. At ten one-block L1 reorgs a day, with messages present in 30% of L1 blocks, that puts an upper bound of roughly 0.13% on the slots your node loses — about three a day across the whole network. Nothing else is affected: the chain carries on and the message is consumed by the next proposer. +`0` consumes a message as soon as your archiver sees it, so a bridged message reaches L2 in the next block your node builds. The cost is that an L1 reorg which changes the messages your node has already consumed costs your node that slot: the archiver drops the messages from the first one whose contents changed, along with the blocks built on them, and the checkpoint is abandoned. A reorg that re-mines the same messages in a different L1 block costs nothing — the archiver replaces only the bucket metadata derived from that block, the blocks that consumed the messages stand, and your node re-resolves the bucket it consumed by its contents before publishing. At ten one-block L1 reorgs a day, with messages present in 30% of L1 blocks, that puts an upper bound of roughly 0.13% on the slots your node loses — about three a day across the whole network, and only for the reorgs that actually reorder or drop a message. Nothing else is affected: the chain carries on and the message is consumed by the next proposer. `1` consumes a message only once a child block is observed on top of the one carrying it, or the following L1 slot was missed and the block is still canonical after it. That removes the lost slots above and adds roughly one Ethereum slot (~12s) of latency to every bridged message. From 3dde9e5ebaa55eb0da3b1aa11c8d532adc433047 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 22:27:25 -0300 Subject: [PATCH 7/9] refactor(fast-inbox): share the message batch loop and the blob replay between the reorg paths Both message fetches walk the same whole-L1-block batches, both updater writes commit alongside the same proposed-chain prune, and both reorg e2es read blobs out of a replayed tx the same way. --- .../src/modules/data_store_updater.ts | 42 +++++---- .../archiver/src/modules/l1_synchronizer.ts | 93 ++++++++++--------- .../archiver/src/store/message_store.ts | 28 +++--- .../cross-chain/streaming_inbox_reorg.test.ts | 10 +- .../l1-reorgs/blocks.parallel.test.ts | 16 +--- .../src/single-node/l1-reorgs/setup.ts | 15 +++ 6 files changed, 109 insertions(+), 95 deletions(-) diff --git a/yarn-project/archiver/src/modules/data_store_updater.ts b/yarn-project/archiver/src/modules/data_store_updater.ts index 127937f192bd..1526d197eb66 100644 --- a/yarn-project/archiver/src/modules/data_store_updater.ts +++ b/yarn-project/archiver/src/modules/data_store_updater.ts @@ -24,7 +24,7 @@ import type { UInt64 } from '@aztec/stdlib/types'; import type { ArchiverDataStores } from '../store/data_stores.js'; import type { L2TipsCache } from '../store/l2_tips_cache.js'; -import type { InboxMessage } from '../structs/inbox_message.js'; +import type { InboxMessageReplacement } from '../store/message_store.js'; /** Operation type for contract data updates. */ enum Operation { @@ -292,16 +292,14 @@ export class ArchiverDataStoreUpdater { * * @returns The pruned blocks. */ - public async rewindMessagesAndPruneProposedBlocks( + public rewindMessagesAndPruneProposedBlocks( messagesSyncPoint: L1BlockId, firstRemovedIndex: bigint, ): Promise { - 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 prunedBlocks; + return this.writeMessagesAndPruneProposedBlocks( + () => this.stores.messages.rewindMessagesTo(messagesSyncPoint, firstRemovedIndex), + firstRemovedIndex, + ); } /** @@ -314,18 +312,26 @@ export class ArchiverDataStoreUpdater { * * @returns The pruned blocks. */ - public async replaceMessagesAndPruneProposedBlocks(args: { - lastCanonicalBucketSeq: bigint | undefined; - firstDifferingIndex: bigint | undefined; - messages: InboxMessage[]; - syncPoint: L1BlockId; - finalizedL1Block: L1BlockId | undefined; - }): Promise { + public replaceMessagesAndPruneProposedBlocks(args: InboxMessageReplacement): Promise { + return this.writeMessagesAndPruneProposedBlocks( + () => this.stores.messages.replaceMessagesAboveBucket(args), + args.firstDifferingIndex, + ); + } + + /** + * Commits a write to the message store and the prune of the proposed blocks that consumed a message from + * `firstRemovedIndex` on, refreshing the tips cache only once both have landed. + */ + private async writeMessagesAndPruneProposedBlocks( + writeMessages: () => Promise, + firstRemovedIndex: bigint | undefined, + ): Promise { const prunedBlocks = await this.stores.db.transactionAsync(async () => { - await this.stores.messages.replaceMessagesAboveBucket(args); - return args.firstDifferingIndex === undefined + await writeMessages(); + return firstRemovedIndex === undefined ? [] - : await this.removeProposedBlocksConsumingMessagesFrom(args.firstDifferingIndex); + : await this.removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex); }); await this.l2TipsCache?.refresh(); return prunedBlocks; diff --git a/yarn-project/archiver/src/modules/l1_synchronizer.ts b/yarn-project/archiver/src/modules/l1_synchronizer.ts index 1b501df1d4b2..e0d6064e6143 100644 --- a/yarn-project/archiver/src/modules/l1_synchronizer.ts +++ b/yarn-project/archiver/src/modules/l1_synchronizer.ts @@ -37,6 +37,7 @@ import { InitialCheckpointNumberNotSequentialError } from '../errors.js'; import { type RetrievedCheckpointFromCalldata, getCheckpointBlobDataFromBlobs, + getL1Block, retrieveCheckpointCalldataFromRollup, retrieveL1ToL2Message, retrieveL1ToL2Messages, @@ -744,33 +745,27 @@ export class ArchiverL1Synchronizer implements Traceable { finalizedL1Block: L1BlockId | undefined, lastStoredIndex: bigint | undefined, ): Promise<{ messages: InboxMessage[]; syncPoint: L1BlockId } | undefined> { - const toL1Block = currentL1Block.l1BlockNumber; const messages: InboxMessage[] = []; const batchTails: InboxMessage[] = []; - let searchEndBlock = fromL1Block - 1n; - let isLastBatch = false; - - while (!isLastBatch) { - const [searchStartBlock, batchEndBlock] = this.nextRange(searchEndBlock, toL1Block); - this.log.trace(`Retrieving canonical L1 to L2 messages in L1 blocks ${searchStartBlock}-${batchEndBlock}`); - const batch = await retrieveL1ToL2Messages(this.inbox, searchStartBlock, batchEndBlock); - messages.push(...batch); - searchEndBlock = batchEndBlock; - isLastBatch = batchEndBlock >= toL1Block; - - const batchTail = batch.at(-1); - if (batchTail !== undefined) { - batchTails.push(batchTail); - } - if (isLastBatch || batchTail === undefined) { + let lastFetchedL1Block = fromL1Block - 1n; + let reachedHead = false; + + for await (const batch of this.iterateL1ToL2MessageBatches(fromL1Block, currentL1Block.l1BlockNumber)) { + messages.push(...batch.messages); + lastFetchedL1Block = batch.searchEndBlock; + reachedHead = batch.isLastBatch; + + const batchTail = batch.messages.at(-1); + if (batchTail === undefined) { continue; } - if (lastStoredIndex !== undefined && batchTail.index < lastStoredIndex) { + batchTails.push(batchTail); + if (reachedHead || (lastStoredIndex !== undefined && batchTail.index < lastStoredIndex)) { continue; } // A bucket that spans co-timestamped L1 blocks can still be open at the end of the batch that covered our last // stored index, and committing there would write a snapshot that undercounts it, so wait for it to close. - const bucketClosed = await this.isBucketClosedAfterL1Block(batchEndBlock, batchTail); + const bucketClosed = await this.isBucketClosedAfterL1Block(batch.searchEndBlock, batchTail); if (bucketClosed === undefined) { return undefined; } @@ -803,20 +798,39 @@ export class ArchiverL1Synchronizer implements Traceable { } } - if (isLastBatch) { + if (reachedHead) { return { messages, syncPoint: currentL1Block }; } try { return { messages, - syncPoint: { l1BlockNumber: searchEndBlock, l1BlockHash: await this.getL1BlockHash(searchEndBlock) }, + syncPoint: { l1BlockNumber: lastFetchedL1Block, l1BlockHash: await this.getL1BlockHash(lastFetchedL1Block) }, }; } catch (error) { - this.warnBucketNotVerifiable(messages.at(-1)!.bucketSeq, searchEndBlock, error); + this.warnBucketNotVerifiable(messages.at(-1)!.bucketSeq, lastFetchedL1Block, error); return undefined; } } + /** + * Retrieves L1 to L2 messages from the given L1 block on, in batches that span whole L1 blocks so that every + * message of an Inbox bucket is yielded in a single batch, as the message store requires. Yields each batch with + * the L1 block range it covers and whether it reached the end of the requested range. + */ + private async *iterateL1ToL2MessageBatches( + fromL1Block: bigint, + toL1Block: bigint, + ): AsyncIterableIterator<{ messages: InboxMessage[]; searchEndBlock: bigint; isLastBatch: boolean }> { + let searchEndBlock = fromL1Block - 1n; + do { + let searchStartBlock: bigint; + [searchStartBlock, searchEndBlock] = this.nextRange(searchEndBlock, toL1Block); + this.log.trace(`Retrieving L1 to L2 messages in L1 blocks ${searchStartBlock}-${searchEndBlock}`); + const messages = await retrieveL1ToL2Messages(this.inbox, searchStartBlock, searchEndBlock); + yield { messages, searchEndBlock, isLastBatch: searchEndBlock >= toL1Block }; + } while (searchEndBlock < toL1Block); + } + /** * Whether the Inbox bucket the given message was absorbed into is closed at the given L1 block, that is, whether * the block after it carries a different timestamp and so can absorb nothing more into that bucket. Undefined when @@ -825,16 +839,10 @@ export class ArchiverL1Synchronizer implements Traceable { private async isBucketClosedAfterL1Block(l1BlockNumber: bigint, message: InboxMessage): Promise { const nextL1BlockNumber = l1BlockNumber + 1n; try { - const block = await this.publicClient.getBlock({ blockNumber: nextL1BlockNumber, includeTransactions: false }); - if (!block) { - throw new Error(`Missing L1 block ${nextL1BlockNumber}`); - } - return block.timestamp !== message.bucketTimestamp; + const { timestamp } = await getL1Block(this.publicClient, nextL1BlockNumber); + return timestamp !== message.bucketTimestamp; } catch (error) { - this.log.warn( - `Failed to read L1 block ${nextL1BlockNumber} to check whether Inbox bucket ${message.bucketSeq} is closed`, - { bucketSeq: message.bucketSeq, l1BlockNumber: nextL1BlockNumber, error }, - ); + this.warnBucketNotVerifiable(message.bucketSeq, nextL1BlockNumber, error); return undefined; } } @@ -851,9 +859,11 @@ export class ArchiverL1Synchronizer implements Traceable { fromIndex: bigint, canonicalMessages: InboxMessage[], ): Promise { - const canonicalByIndex = new Map(canonicalMessages.map(message => [message.index, message])); + // The canonical messages are contiguous and ascending by index, so the one at any index is found by offset. + const firstCanonicalIndex = canonicalMessages.at(0)?.index; for await (const stored of this.stores.messages.iterateL1ToL2Messages({ start: fromIndex })) { - const canonical = canonicalByIndex.get(stored.index); + const canonical = + firstCanonicalIndex === undefined ? undefined : canonicalMessages[Number(stored.index - firstCanonicalIndex)]; if (canonical === undefined || !canonical.inboxRollingHash.equals(stored.inboxRollingHash)) { return stored.index; } @@ -951,20 +961,18 @@ export class ArchiverL1Synchronizer implements Traceable { currentL1Block: L1BlockId, finalizedL1Block: L1BlockId | undefined, ): Promise { - const toL1Block = currentL1Block.l1BlockNumber; const newestBucketBeforeFetch = await this.stores.messages.getNewestInboxBucketL1Span(); - let searchStartBlock: bigint = 0n; - let searchEndBlock: bigint = syncFrom.l1BlockNumber; let firstMessage: InboxMessage | undefined; let lastMessage: InboxMessage | undefined; let messageCount = 0; const unfinalizedBatchTails: InboxMessage[] = []; - do { - [searchStartBlock, searchEndBlock] = this.nextRange(searchEndBlock, toL1Block); - this.log.trace(`Retrieving L1 to L2 messages in L1 blocks ${searchStartBlock}-${searchEndBlock}`); - const messages = await retrieveL1ToL2Messages(this.inbox, searchStartBlock, searchEndBlock); + for await (const batch of this.iterateL1ToL2MessageBatches( + syncFrom.l1BlockNumber + 1n, + currentL1Block.l1BlockNumber, + )) { + const { messages } = batch; const timer = new Timer(); await this.stores.messages.addL1ToL2MessageBuckets(messages); const perMsg = timer.ms() / messages.length; @@ -976,11 +984,10 @@ export class ArchiverL1Synchronizer implements Traceable { messageCount++; } const batchTail = messages.at(-1); - const isLastBatch = searchEndBlock >= toL1Block; - if (batchTail !== undefined && !isLastBatch && !isFinalized(batchTail, finalizedL1Block)) { + if (batchTail !== undefined && !batch.isLastBatch && !isFinalized(batchTail, finalizedL1Block)) { unfinalizedBatchTails.push(batchTail); } - } while (searchEndBlock < toL1Block); + } if (firstMessage === undefined || lastMessage === undefined) { return 'stored'; diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index 50c73bda0bf3..df0f08143c2f 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -102,6 +102,20 @@ function toBucketL1Span(seq: bigint, snapshot: BucketSnapshot): InboxBucketL1Spa }; } +/** Everything the canonical chain gives back for the L1 range above a bucket that survived a reorg. */ +export type InboxMessageReplacement = { + /** Sequence of the newest bucket known to sit on canonical L1 blocks, if any. */ + lastCanonicalBucketSeq: bigint | undefined; + /** Lowest message index whose leaf the canonical chain changed, if any. */ + firstDifferingIndex: bigint | undefined; + /** The canonical messages from the last canonical bucket's opening L1 block onwards. */ + messages: InboxMessage[]; + /** L1 block the messages were fetched up to. */ + syncPoint: L1BlockId; + /** L1 finalized block to record, if it is at or below the sync point. */ + finalizedL1Block: L1BlockId | undefined; +}; + /** The messages of a single Inbox bucket within an incoming batch, in insertion order. */ type IncomingBucket = { seq: bigint; @@ -506,20 +520,8 @@ export class MessageStore { * * `messages` must start at the first message of the last canonical bucket (or of the Inbox when there is none), so * that every bucket it covers arrives whole. - * - * @param lastCanonicalBucketSeq - Sequence of the newest bucket known to sit on canonical L1 blocks, if any. - * @param firstDifferingIndex - Lowest message index whose leaf the canonical chain changed, if any. - * @param messages - The canonical messages from the last canonical bucket's opening L1 block onwards. - * @param syncPoint - L1 block the messages were fetched up to. - * @param finalizedL1Block - L1 finalized block to record, if it is at or below the sync point. */ - public replaceMessagesAboveBucket(args: { - lastCanonicalBucketSeq: bigint | undefined; - firstDifferingIndex: bigint | undefined; - messages: InboxMessage[]; - syncPoint: L1BlockId; - finalizedL1Block: L1BlockId | undefined; - }): Promise { + public replaceMessagesAboveBucket(args: InboxMessageReplacement): Promise { const { lastCanonicalBucketSeq, firstDifferingIndex, messages, syncPoint, finalizedL1Block } = args; return this.db.transactionAsync(async () => { if (firstDifferingIndex !== undefined) { diff --git a/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts b/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts index 23103060007a..b918a7931e0e 100644 --- a/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts +++ b/yarn-project/end-to-end/src/single-node/cross-chain/streaming_inbox_reorg.test.ts @@ -6,23 +6,20 @@ import type { Logger } from '@aztec/aztec.js/log'; import { isL1ToL2MessageReady, waitForL1ToL2MessageReady } from '@aztec/aztec.js/messaging'; import type { AztecNode } from '@aztec/aztec.js/node'; import { createBlobClient } from '@aztec/blob-client/client'; -import { Blob } from '@aztec/blob-lib'; import type { Delayer } from '@aztec/ethereum/l1-tx-utils'; import type { ChainMonitor } from '@aztec/ethereum/test'; import type { ExtendedViemWalletClient } from '@aztec/ethereum/types'; import { CheckpointNumber } from '@aztec/foundation/branded-types'; import { retryUntil } from '@aztec/foundation/retry'; -import { hexToBuffer } from '@aztec/foundation/string'; import { L2BlockSourceEvents, type L2PruneUncheckpointedEvent } from '@aztec/stdlib/block'; import { WorldStateSynchronizerError } from '@aztec/world-state'; import 'jest-extended'; -import { parseTransaction } from 'viem'; import { sendL1ToL2Message } from '../../fixtures/l1_to_l2_messaging.js'; import type { EndToEndContext } from '../../fixtures/utils.js'; import { waitForL1ToL2MessageSeen } from '../../shared/wait_for_l1_to_l2_message.js'; -import { L1ReorgsTest, TX_COUNT } from '../l1-reorgs/setup.js'; +import { L1ReorgsTest, TX_COUNT, getBlobsFromRawTx } from '../l1-reorgs/setup.js'; import type { SingleNodeTestContext } from '../single_node_test_context.js'; // Single-node + prover-node suite covering what happens when an L1 reorg orphans the Inbox messages that a @@ -275,10 +272,7 @@ describe('single-node/cross-chain/streaming_inbox_reorg', () => { }); // The node reads the checkpoint's blocks off the blob sink, which never saw the replayed propose. - const blobs = await Promise.all( - (parseTransaction(proposeTx).sidecars || []).map(sidecar => Blob.fromBlobBuffer(hexToBuffer(sidecar.blob))), - ); - await createBlobClient(context.config).sendBlobsToFilestore(blobs); + await createBlobClient(context.config).sendBlobsToFilestore(await getBlobsFromRawTx(proposeTx)); // The bucket the consuming block built on moves to the replacement L1 block, keeping the sequence number and // the rolling hash the sealed checkpoint header commits to. That is what makes the checkpoint publishable. diff --git a/yarn-project/end-to-end/src/single-node/l1-reorgs/blocks.parallel.test.ts b/yarn-project/end-to-end/src/single-node/l1-reorgs/blocks.parallel.test.ts index bec1217ee721..9bb79e73498d 100644 --- a/yarn-project/end-to-end/src/single-node/l1-reorgs/blocks.parallel.test.ts +++ b/yarn-project/end-to-end/src/single-node/l1-reorgs/blocks.parallel.test.ts @@ -2,22 +2,20 @@ import type { Archiver } from '@aztec/archiver'; import type { Logger } from '@aztec/aztec.js/log'; import type { AztecNode } from '@aztec/aztec.js/node'; import { createBlobClient } from '@aztec/blob-client/client'; -import { Blob } from '@aztec/blob-lib'; import type { Delayer } from '@aztec/ethereum/l1-tx-utils'; import type { ChainMonitor, ChainMonitorEventMap } from '@aztec/ethereum/test'; import { CheckpointNumber } from '@aztec/foundation/branded-types'; import { AbortError } from '@aztec/foundation/error'; import { retryUntil } from '@aztec/foundation/retry'; -import { hexToBuffer } from '@aztec/foundation/string'; import { executeTimeout } from '@aztec/foundation/timer'; import 'jest-extended'; -import { keccak256, parseTransaction } from 'viem'; +import { keccak256 } from 'viem'; import type { EndToEndContext } from '../../fixtures/utils.js'; import { waitForNodeCheckpoint, waitForNodeProvenCheckpoint } from '../../fixtures/wait_helpers.js'; import type { SingleNodeTestContext } from '../single_node_test_context.js'; -import { L1ReorgsTest, TX_COUNT } from './setup.js'; +import { L1ReorgsTest, TX_COUNT, getBlobsFromRawTx } from './setup.js'; // Single-node + prover-node suite exercising L1 reorg behavior for L2 block state: proof removal, proof // re-addition via reorg, checkpoint removal from the pending chain, and checkpoint insertion via reorg. @@ -53,14 +51,6 @@ describe('single-node/l1-reorgs/blocks', () => { await t.teardown(); }); - const getBlobs = async (serializedTx: `0x${string}`) => { - const parsedTx = parseTransaction(serializedTx); - if (parsedTx.sidecars === false) { - throw new Error('No sidecars found in tx'); - } - return await Promise.all(parsedTx.sidecars!.map(sidecar => Blob.fromBlobBuffer(hexToBuffer(sidecar.blob)))); - }; - // Waits for an initial proof to land, stops the prover, reorgs L1 to remove the proof block, // waits for the proof submission window to expire, spins up a new sync-only node, and verifies // both the new node and the old node have rolled back to the pre-proof checkpoint number. @@ -412,7 +402,7 @@ describe('single-node/l1-reorgs/blocks', () => { // We also need to send the blob to the sink, so the node can get it logger.warn(`Sending blobs to blob client`); - const blobs = await getBlobs(l2BlockTx); + const blobs = await getBlobsFromRawTx(l2BlockTx); const blobClient = createBlobClient(context.config); await blobClient.sendBlobsToFilestore(blobs); diff --git a/yarn-project/end-to-end/src/single-node/l1-reorgs/setup.ts b/yarn-project/end-to-end/src/single-node/l1-reorgs/setup.ts index 57a9ea36a398..259b14ed1801 100644 --- a/yarn-project/end-to-end/src/single-node/l1-reorgs/setup.ts +++ b/yarn-project/end-to-end/src/single-node/l1-reorgs/setup.ts @@ -4,12 +4,15 @@ import type { AztecAddress } from '@aztec/aztec.js/addresses'; import { Fr } from '@aztec/aztec.js/fields'; import type { Logger } from '@aztec/aztec.js/log'; import type { AztecNode } from '@aztec/aztec.js/node'; +import { Blob } from '@aztec/blob-lib'; import type { Delayer } from '@aztec/ethereum/l1-tx-utils'; import type { ChainMonitor } from '@aztec/ethereum/test'; +import { hexToBuffer } from '@aztec/foundation/string'; import type { TestContract } from '@aztec/noir-test-contracts.js/Test'; import type { TxHash } from '@aztec/stdlib/tx'; import { jest } from '@jest/globals'; +import { parseTransaction } from 'viem'; import type { EndToEndContext } from '../../fixtures/utils.js'; import { proveAndSendTxs } from '../../test-wallet/utils.js'; @@ -21,6 +24,18 @@ jest.setTimeout(1000 * 60 * 20); /** Number of txs to send at the start of each reorg test to trigger multi-block checkpoints. */ export const TX_COUNT = 8; +/** + * Reads the blobs out of a serialized blob-carrying L1 tx, so a tx replayed by hand after a reorg can have its blobs + * pushed back to the blob sink. A node reads a checkpoint's blocks off the sink, which never sees a replayed tx. + */ +export async function getBlobsFromRawTx(serializedTx: `0x${string}`): Promise { + const parsedTx = parseTransaction(serializedTx); + if (!parsedTx.sidecars) { + throw new Error('No sidecars found in tx'); + } + return await Promise.all(parsedTx.sidecars.map(sidecar => Blob.fromBlobBuffer(hexToBuffer(sidecar.blob)))); +} + /** * The single-node + prover-node fixture shared by the L1-reorg suites (`blocks`, `messages`). Stands * up a {@link SingleNodeTestContext} on the {@link FAST_REORG_TIMING} cadence (ethSlot=4s, From d48bea0bd7d3d0835de61ed9c2f0860a412a84bf Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Thu, 3 Sep 2026 22:35:17 -0300 Subject: [PATCH 8/9] fix(fast-inbox): keep publishing when the proposer never pushes its blocks to the archiver A node running with skipPushProposedBlocksToArchiver or in fisherman mode keeps its proposed blocks out of the local archiver on purpose, so their absence at publish time says nothing about a prune. --- .../src/sequencer/checkpoint_proposal_job.test.ts | 15 +++++++++++++++ .../src/sequencer/checkpoint_proposal_job.ts | 6 ++++++ 2 files changed, 21 insertions(+) diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts index f9f42bbebbf4..1faeaed7a4eb 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts @@ -1872,6 +1872,21 @@ describe('CheckpointProposalJob', () => { expect(metrics.recordCheckpointProposalFailed).toHaveBeenCalledWith('checkpoint_blocks_pruned'); }); + it('submits when the node was configured never to push its blocks to the archiver', async () => { + config = { ...config, skipPushProposedBlocksToArchiver: true }; + job = createCheckpointProposalJob(); + job.setTimetable(makeSingleBlockTimetable()); + const block = await setupOneBlockCheckpoint(); + validatorClient.collectAttestations.mockResolvedValue(getAttestations(block)); + + const checkpoint = await job.executeAndAwait(); + + // The blocks were never pushed, so the archiver not holding them says nothing about a prune. + expect(checkpoint).toBeDefined(); + expect(publisher.enqueueProposeCheckpoint).toHaveBeenCalledTimes(1); + expect(metrics.recordCheckpointProposalFailed).not.toHaveBeenCalledWith('checkpoint_blocks_pruned'); + }); + it('does not submit when the archiver rebuilt the block number with a different block', async () => { const block = await setupOneBlockCheckpoint(); const replacement = L2Block.fromBuffer(block.toBuffer()); diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts index 525c52e223ae..d5b3ca26cbcb 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts @@ -458,8 +458,14 @@ export class CheckpointProposalJob implements Traceable { * longer holds would put a checkpoint on L1 that its own archiver cannot serve. * * The hash is compared rather than the height alone, since a chain rebuilt after a prune reuses the block numbers. + * + * Skipped whenever proposed blocks aren't pushed (`skipPushProposedBlocksToArchiver`, fisherman mode): the archiver + * never held them in the first place, so their absence says nothing about a prune. */ private async checkpointBlocksAreStillLocal(checkpoint: Checkpoint): Promise { + if (this.config.skipPushProposedBlocksToArchiver || this.config.fishermanMode) { + return true; + } const lastBlock = checkpoint.blocks.at(-1); if (lastBlock === undefined) { return true; From 485d53bd486d6afde0d544fc8621dcdb72606650 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Fri, 4 Sep 2026 10:00:27 -0300 Subject: [PATCH 9/9] fix(fast-inbox): prune the blocks whose bucket boundary a reorg merged away A block ends on an Inbox bucket boundary, and world state, the prover and the next checkpoint's bundle selection all derive what it consumed from the buckets that boundary belongs to. A reorg that re-mines two buckets' messages into a single L1 block leaves every leaf where it was, so the content comparison found nothing to prune, but the boundary between the two buckets stopped existing and the block that ended on it could no longer be replayed. The swap now treats a boundary the canonical chain took away as a difference, one-sidedly: a merge prunes the blocks that ended on the lost boundary, while a split only adds a boundary no block ever ended on and prunes nothing. The walk starts at the last canonical bucket's own last message, since that bucket can keep its L1 blocks and still absorb the re-mined messages. --- .../operators/reference/changelog/v6.md | 2 +- .../archiver/src/archiver-sync.test.ts | 72 +++++++++++++++++-- .../src/modules/data_store_updater.ts | 35 ++++----- .../archiver/src/modules/l1_synchronizer.ts | 65 ++++++++++++----- .../archiver/src/store/message_store.test.ts | 8 +-- .../archiver/src/store/message_store.ts | 17 ++--- .../src/sequencer/checkpoint_proposal_job.ts | 4 ++ 7 files changed, 151 insertions(+), 52 deletions(-) diff --git a/docs/docs-operate/operators/reference/changelog/v6.md b/docs/docs-operate/operators/reference/changelog/v6.md index 688eb4621f3d..689e39dc665e 100644 --- a/docs/docs-operate/operators/reference/changelog/v6.md +++ b/docs/docs-operate/operators/reference/changelog/v6.md @@ -16,7 +16,7 @@ displayed_sidebar: operatorsSidebar ### Choosing a value -`0` consumes a message as soon as your archiver sees it, so a bridged message reaches L2 in the next block your node builds. The cost is that an L1 reorg which changes the messages your node has already consumed costs your node that slot: the archiver drops the messages from the first one whose contents changed, along with the blocks built on them, and the checkpoint is abandoned. A reorg that re-mines the same messages in a different L1 block costs nothing — the archiver replaces only the bucket metadata derived from that block, the blocks that consumed the messages stand, and your node re-resolves the bucket it consumed by its contents before publishing. At ten one-block L1 reorgs a day, with messages present in 30% of L1 blocks, that puts an upper bound of roughly 0.13% on the slots your node loses — about three a day across the whole network, and only for the reorgs that actually reorder or drop a message. Nothing else is affected: the chain carries on and the message is consumed by the next proposer. +`0` consumes a message as soon as your archiver sees it, so a bridged message reaches L2 in the next block your node builds. The cost is that an L1 reorg which changes the messages your node has already consumed costs your node that slot: the archiver drops the messages from the first one whose contents changed, along with the blocks built on them, and the checkpoint is abandoned. A reorg that re-mines the same messages in a different L1 block costs nothing — the archiver replaces only the bucket metadata derived from that block, the blocks that consumed the messages stand, and your node re-resolves the bucket it consumed by its contents before publishing. The exception is a re-mine that lands two buckets' messages in a single L1 block: the boundary between them stops existing, and the block that consumed through it goes, along with the blocks after it. At ten one-block L1 reorgs a day, with messages present in 30% of L1 blocks, that puts an upper bound of roughly 0.13% on the slots your node loses — about three a day across the whole network, and only for the reorgs that reorder a message, drop one, or merge two buckets into one. Nothing else is affected: the chain carries on and the message is consumed by the next proposer. `1` consumes a message only once a child block is observed on top of the one carrying it, or the following L1 slot was missed and the block is still canonical after it. That removes the lost slots above and adds roughly one Ethereum slot (~12s) of latency to every bridged message. diff --git a/yarn-project/archiver/src/archiver-sync.test.ts b/yarn-project/archiver/src/archiver-sync.test.ts index dc1259f90614..667ca1dfc984 100644 --- a/yarn-project/archiver/src/archiver-sync.test.ts +++ b/yarn-project/archiver/src/archiver-sync.test.ts @@ -2417,7 +2417,7 @@ describe('Archiver Sync', () => { }); }); - it('keeps the proposed chain when the reorg merges the bucket it consumed into an earlier one', async () => { + it('prunes when the reorg merges away the bucket boundary a block ended on', async () => { const { early, late } = addMessages(); fake.setL1BlockNumber(l1BlockNumber); await archiver.syncImmediate(); @@ -2430,20 +2430,80 @@ describe('Archiver Sync', () => { // The rolling hash the checkpoint carrying these blocks committed to. const consumedRollingHash = (await archiverStore.messages.getInboxBucket(2n))!.inboxRollingHash; - // The messages of block 100 are re-mined into block 102, so all four end up in a single bucket. The leaves - // and their order are untouched, so the blocks that consumed them stay valid. + // 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. fake.retimeMessages(100n, 102n); await archiver.syncImmediate(); expect(await getStoredLeaves()).toEqual(asHex([...early, ...late])); - expect(await archiver.getBlockNumber()).toEqual(BlockNumber(2)); - expect(pruneSpy).not.toHaveBeenCalled(); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(0)); + expect(pruneSpy).toHaveBeenCalledWith( + expect.objectContaining({ type: L2BlockSourceEvents.L2PruneUncheckpointed, blocks }), + ); expect(await archiverStore.messages.getInboxBucket(1n)).toMatchObject({ msgCount: 4, l1BlockNumber: 102n }); expect(await archiverStore.messages.getInboxBucket(2n)).toBeUndefined(); - // The proposer re-resolves its publish hint to the merged bucket, so the sealed header stays publishable. + // 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 }); }); + it('keeps the proposed chain when the reorg splits the bucket a block consumed', async () => { + const early = [Fr.random(), Fr.random()]; + const late = [Fr.random(), Fr.random()]; + // All four messages sit in one bucket, spanning the two co-timestamped L1 blocks that carry them. + fake.addMessages(CheckpointNumber(1), 100n, early); + fake.addMessages(CheckpointNumber(1), 101n, late); + fake.shareTimestampWithL1Block(101n, 100n); + fake.setL1BlockNumber(l1BlockNumber); + await archiver.syncImmediate(); + + const blocks = await makeBlocksConsumingThrough([4]); + for (const block of blocks) { + await archiver.addBlock(block); + } + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1)); + + // L1 block 101 is re-mined with a timestamp of its own, cutting the bucket in two. That adds a boundary at + // leaf count 2 that no block ever ended on, and leaves the one at 4 where it was, so nothing is pruned. + fake.splitCoTimestampedL1Block(101n); + await archiver.syncImmediate(); + + expect(await getStoredLeaves()).toEqual(asHex([...early, ...late])); + expect(await archiverStore.messages.getInboxBucket(1n)).toMatchObject({ msgCount: 2, totalMsgCount: 2n }); + expect(await archiverStore.messages.getInboxBucket(2n)).toMatchObject({ msgCount: 2, totalMsgCount: 4n }); + expect(await archiver.getBlockNumber()).toEqual(BlockNumber(1)); + expect(pruneSpy).not.toHaveBeenCalled(); + }); + + it('prunes when the reorg extends the last canonical bucket past the boundary a block ended on', 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); + 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 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. + 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 }), + ); + }); + it('prunes from the first leaf the reorg changed, not from the start of its bucket', async () => { const { early, late } = addMessages(); fake.setL1BlockNumber(l1BlockNumber); diff --git a/yarn-project/archiver/src/modules/data_store_updater.ts b/yarn-project/archiver/src/modules/data_store_updater.ts index 1526d197eb66..baefce3f36d6 100644 --- a/yarn-project/archiver/src/modules/data_store_updater.ts +++ b/yarn-project/archiver/src/modules/data_store_updater.ts @@ -304,7 +304,8 @@ export class ArchiverDataStoreUpdater { /** * Replaces the messages above the last canonical Inbox bucket with the ones the canonical chain delivers, and - * prunes the proposed blocks that consumed a leaf the swap changed, in a single transaction. + * prunes the proposed blocks that consumed a leaf the swap changed or ended on a bucket boundary it took away, in + * a single transaction. * * Splitting the two would make the prune unrecoverable, for the same reason * {@link rewindMessagesAndPruneProposedBlocks} keeps them together: the next sync pass would find the local @@ -315,23 +316,23 @@ export class ArchiverDataStoreUpdater { public replaceMessagesAndPruneProposedBlocks(args: InboxMessageReplacement): Promise { return this.writeMessagesAndPruneProposedBlocks( () => this.stores.messages.replaceMessagesAboveBucket(args), - args.firstDifferingIndex, + args.firstInvalidatedIndex, ); } /** * Commits a write to the message store and the prune of the proposed blocks that consumed a message from - * `firstRemovedIndex` on, refreshing the tips cache only once both have landed. + * `firstInvalidatedIndex` on, refreshing the tips cache only once both have landed. */ private async writeMessagesAndPruneProposedBlocks( writeMessages: () => Promise, - firstRemovedIndex: bigint | undefined, + firstInvalidatedIndex: bigint | undefined, ): Promise { const prunedBlocks = await this.stores.db.transactionAsync(async () => { await writeMessages(); - return firstRemovedIndex === undefined + return firstInvalidatedIndex === undefined ? [] - : await this.removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex); + : await this.removeProposedBlocksConsumingMessagesFrom(firstInvalidatedIndex); }); await this.l2TipsCache?.refresh(); return prunedBlocks; @@ -339,23 +340,23 @@ export class ArchiverDataStoreUpdater { /** * 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 + * `firstInvalidatedIndex`, 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. + * local chain no longer backs, and one equal to it means the block stopped below it. * - * The index is the first leaf whose value actually changed, so a reorg that re-mines the same messages in a - * different L1 block prunes nothing, and one that replaces the tail of a bucket keeps the blocks that stopped - * below the replaced leaf. + * The index is the first leaf whose value actually changed, or whose bucket boundary the reorg took away, so a + * reorg that re-mines the same messages in a different L1 block prunes nothing, and one that replaces the tail of + * a bucket keeps the blocks that stopped below the replaced leaf. * * 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. + * @param firstInvalidatedIndex - Index of the first L1-to-L2 message the local chain no longer backs. * @returns The removed blocks. */ - private async removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex: bigint): Promise { + private async removeProposedBlocksConsumingMessagesFrom(firstInvalidatedIndex: bigint): Promise { const [checkpointedBlockNumber, latestBlockNumber] = await Promise.all([ this.stores.blocks.getCheckpointedL2BlockNumber(), this.stores.blocks.getLatestL2BlockNumber(), @@ -371,10 +372,10 @@ export class ArchiverDataStoreUpdater { const checkpointedTipLeafCount = BigInt( checkpointedTip?.header.state.l1ToL2MessageTree.nextAvailableLeafIndex ?? 0, ); - if (firstRemovedIndex < checkpointedTipLeafCount) { + if (firstInvalidatedIndex < checkpointedTipLeafCount) { this.log.warn( - `Not pruning proposed blocks: rollback index ${firstRemovedIndex} is below the checkpointed tip's leaf count`, - { firstRemovedIndex, checkpointedTipLeafCount, checkpointedBlockNumber }, + `Not pruning proposed blocks: rollback index ${firstInvalidatedIndex} is below the checkpointed tip's leaf count`, + { firstInvalidatedIndex, checkpointedTipLeafCount, checkpointedBlockNumber }, ); return []; } @@ -384,7 +385,7 @@ export class ArchiverDataStoreUpdater { limit: latestBlockNumber - checkpointedBlockNumber, }); const firstAffected = proposedBlocks.find( - block => BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex) > firstRemovedIndex, + block => BigInt(block.header.state.l1ToL2MessageTree.nextAvailableLeafIndex) > firstInvalidatedIndex, ); if (firstAffected === undefined) { return []; diff --git a/yarn-project/archiver/src/modules/l1_synchronizer.ts b/yarn-project/archiver/src/modules/l1_synchronizer.ts index e0d6064e6143..fd3301c43f75 100644 --- a/yarn-project/archiver/src/modules/l1_synchronizer.ts +++ b/yarn-project/archiver/src/modules/l1_synchronizer.ts @@ -698,15 +698,16 @@ export class ArchiverL1Synchronizer implements Traceable { } // The last canonical bucket's own messages need no comparison: both of the L1 blocks it sits on were just - // checked against the live chain, so nothing below its last message can have changed. - const firstDifferingIndex = await this.findFirstDifferingMessageIndex( - lastCanonicalBucket ? lastCanonicalBucket.lastMessageIndex + 1n : 0n, + // checked against the live chain, so nothing below its last message can have changed. Its last message is + // still walked, because the boundary that closes it is only worth as much as the bucket that follows it. + const firstInvalidatedIndex = await this.findFirstInvalidatedMessageIndex( + lastCanonicalBucket ? lastCanonicalBucket.lastMessageIndex : 0n, fetched.messages, ); const prunedBlocks = await this.updater.replaceMessagesAndPruneProposedBlocks({ lastCanonicalBucketSeq: lastCanonicalBucket?.seq, - firstDifferingIndex, + firstInvalidatedIndex, messages: fetched.messages, syncPoint: fetched.syncPoint, // The rolling-hash fallback trusts every message at or below the finalized marker without querying L1, so the @@ -718,10 +719,10 @@ export class ArchiverL1Synchronizer implements Traceable { }); this.log.verbose(`Updated messages syncpoint to L1 block ${fetched.syncPoint.l1BlockNumber}`, { ...fetched.syncPoint, - firstDifferingIndex, + firstInvalidatedIndex, messageCount: fetched.messages.length, }); - this.reportPrunedProposedBlocks(prunedBlocks, firstDifferingIndex); + this.reportPrunedProposedBlocks(prunedBlocks, firstInvalidatedIndex); return fetched.syncPoint; } @@ -848,25 +849,57 @@ export class ArchiverL1Synchronizer implements Traceable { } /** - * Returns the lowest index at or above `fromIndex` where the messages we hold and the canonical ones disagree, or - * undefined when every stored message from there on is re-delivered unchanged. A message that the canonical chain - * does not carry at all counts as a difference. + * Returns the lowest index at or above `fromIndex` that the canonical chain no longer backs, or undefined when + * every stored message from there on survives the swap whole. An index is invalidated when its leaf changed, and + * also when the bucket boundary that closed on it is gone: a block ends on a bucket boundary and consumers derive + * the messages it inserted from the buckets that boundary belongs to, so a boundary that stops existing takes the + * blocks that ended on it with it. * - * Comparing consensus rolling hashes is enough to compare whole prefixes: each one chains the hash before it, so - * equal hashes at an index mean equal leaves at every index through it. + * Comparing consensus rolling hashes is enough to compare whole prefixes of leaves: each one chains the hash + * before it, so equal hashes at an index mean equal leaves at every index through it. A message that the canonical + * chain does not carry at all counts as invalidated. + * + * The boundary comparison is deliberately one-sided. A reorg that merges two buckets into one destroys the + * boundary between them, so the blocks that ended there are pruned; one that splits a bucket only adds a boundary + * no block ever ended on, and every boundary that existed still does, so it prunes nothing. + * + * The one boundary this cannot settle is the newest one: when the canonical fetch reaches the L1 head it cannot + * tell a bucket that closed from one that is still open, and a bucket that carries on absorbing messages later + * takes its boundary away. That exposure is the same one ordinary ingestion has, since it stores an open head + * bucket as readily as a closed one, and is neither created nor widened here. */ - private async findFirstDifferingMessageIndex( + private async findFirstInvalidatedMessageIndex( fromIndex: bigint, canonicalMessages: InboxMessage[], ): Promise { // The canonical messages are contiguous and ascending by index, so the one at any index is found by offset. const firstCanonicalIndex = canonicalMessages.at(0)?.index; + const canonicalAt = (index: bigint) => + firstCanonicalIndex === undefined ? undefined : canonicalMessages[Number(index - firstCanonicalIndex)]; + + let previous: { stored: InboxMessage; canonical: InboxMessage } | undefined; for await (const stored of this.stores.messages.iterateL1ToL2Messages({ start: fromIndex })) { - const canonical = - firstCanonicalIndex === undefined ? undefined : canonicalMessages[Number(stored.index - firstCanonicalIndex)]; + const canonical = canonicalAt(stored.index); if (canonical === undefined || !canonical.inboxRollingHash.equals(stored.inboxRollingHash)) { return stored.index; } + if (previous !== undefined) { + const storedBucketClosed = previous.stored.bucketSeq !== stored.bucketSeq; + const canonicalBucketClosed = previous.canonical.bucketSeq !== canonical.bucketSeq; + if (storedBucketClosed && !canonicalBucketClosed) { + return previous.stored.index; + } + } + previous = { stored, canonical }; + } + + // The last message we hold closes the last bucket we hold, so a canonical bucket that carries on past it takes + // that boundary away as much as a merge does. + if (previous !== undefined) { + const next = canonicalAt(previous.stored.index + 1n); + if (next !== undefined && next.bucketSeq === previous.canonical.bucketSeq) { + return previous.stored.index; + } } return undefined; } @@ -910,7 +943,7 @@ export class ArchiverL1Synchronizer implements Traceable { } /** Logs, counts and announces the proposed blocks a message rollback dropped. */ - private reportPrunedProposedBlocks(prunedBlocks: L2Block[], firstRemovedIndex: bigint | undefined): void { + private reportPrunedProposedBlocks(prunedBlocks: L2Block[], firstInvalidatedIndex: bigint | undefined): void { if (prunedBlocks.length === 0) { return; } @@ -918,7 +951,7 @@ export class ArchiverL1Synchronizer implements Traceable { const firstPrunedBlock = prunedBlocks[0]; this.log.warn( `Pruning ${prunedBlocks.length} proposed blocks from ${firstPrunedBlock.number} built on rolled back messages`, - { firstRemovedIndex, firstPrunedBlockNumber: firstPrunedBlock.number, prunedCount: prunedBlocks.length }, + { firstInvalidatedIndex, firstPrunedBlockNumber: firstPrunedBlock.number, prunedCount: prunedBlocks.length }, ); this.instrumentation.recordPrune('inbox_rollback'); this.events.emit(L2BlockSourceEvents.L2PruneUncheckpointed, { diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts index 42a1d833dce8..3cf21ebfb2e4 100644 --- a/yarn-project/archiver/src/store/message_store.test.ts +++ b/yarn-project/archiver/src/store/message_store.test.ts @@ -739,7 +739,7 @@ describe('MessageStore', () => { await messageStore.replaceMessagesAboveBucket({ lastCanonicalBucketSeq: 2n, - firstDifferingIndex: undefined, + firstInvalidatedIndex: undefined, messages: [retimed], syncPoint, finalizedL1Block: undefined, @@ -766,7 +766,7 @@ describe('MessageStore', () => { await messageStore.replaceMessagesAboveBucket({ lastCanonicalBucketSeq: 1n, - firstDifferingIndex: undefined, + firstInvalidatedIndex: undefined, messages: merged, syncPoint, finalizedL1Block: undefined, @@ -801,7 +801,7 @@ describe('MessageStore', () => { await messageStore.replaceMessagesAboveBucket({ lastCanonicalBucketSeq: 1n, - firstDifferingIndex: undefined, + firstInvalidatedIndex: undefined, messages: split, syncPoint, finalizedL1Block: undefined, @@ -824,7 +824,7 @@ describe('MessageStore', () => { await messageStore.replaceMessagesAboveBucket({ lastCanonicalBucketSeq: 1n, - firstDifferingIndex: msgs[4].index, + firstInvalidatedIndex: msgs[4].index, messages: [msgs[3], replacement], syncPoint, finalizedL1Block: { l1BlockNumber: 800n, l1BlockHash: makeL1BlockHash(800n) }, diff --git a/yarn-project/archiver/src/store/message_store.ts b/yarn-project/archiver/src/store/message_store.ts index df0f08143c2f..3fba05442fee 100644 --- a/yarn-project/archiver/src/store/message_store.ts +++ b/yarn-project/archiver/src/store/message_store.ts @@ -106,8 +106,8 @@ function toBucketL1Span(seq: bigint, snapshot: BucketSnapshot): InboxBucketL1Spa export type InboxMessageReplacement = { /** Sequence of the newest bucket known to sit on canonical L1 blocks, if any. */ lastCanonicalBucketSeq: bigint | undefined; - /** Lowest message index whose leaf the canonical chain changed, if any. */ - firstDifferingIndex: bigint | undefined; + /** Lowest message index whose leaf or bucket boundary the canonical chain no longer backs, if any. */ + firstInvalidatedIndex: bigint | undefined; /** The canonical messages from the last canonical bucket's opening L1 block onwards. */ messages: InboxMessage[]; /** L1 block the messages were fetched up to. */ @@ -510,9 +510,10 @@ export class MessageStore { * Replaces everything the store holds above the given bucket with the messages the canonical chain delivers for the * same range, and moves the sync point to the L1 block that range was fetched up to, all in a single transaction. * - * The caller has already compared the two views: `firstDifferingIndex` is the lowest index whose leaf changed, and - * is undefined when the canonical chain re-delivers the very same leaves, in which case not a single message is - * dropped and only the bucket metadata derived from the L1 blocks moves. The steps depend on each other in order: + * The caller has already compared the two views: `firstInvalidatedIndex` is the lowest index whose leaf changed or + * whose bucket boundary the canonical chain took away, and is undefined when the canonical chain re-delivers the + * very same leaves under the same bucket boundaries, in which case not a single message is dropped and only the + * bucket metadata derived from the L1 blocks moves. The steps depend on each other in order: * the removal rewrites the snapshot of the bucket left holding the last surviving message, so it has to run while * that snapshot is still there; the snapshots above the last canonical bucket then go, because a re-delivery that * renumbers or merges buckets would otherwise be rejected as incomplete or leave a stale snapshot behind; and only @@ -522,10 +523,10 @@ export class MessageStore { * that every bucket it covers arrives whole. */ public replaceMessagesAboveBucket(args: InboxMessageReplacement): Promise { - const { lastCanonicalBucketSeq, firstDifferingIndex, messages, syncPoint, finalizedL1Block } = args; + const { lastCanonicalBucketSeq, firstInvalidatedIndex, messages, syncPoint, finalizedL1Block } = args; return this.db.transactionAsync(async () => { - if (firstDifferingIndex !== undefined) { - await this.removeL1ToL2Messages(firstDifferingIndex); + if (firstInvalidatedIndex !== undefined) { + await this.removeL1ToL2Messages(firstInvalidatedIndex); } await this.deleteBucketSnapshotsAbove(lastCanonicalBucketSeq); await this.addL1ToL2MessageBuckets(messages); diff --git a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts index d5b3ca26cbcb..3ec51cc4ea82 100644 --- a/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts +++ b/yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts @@ -459,6 +459,10 @@ export class CheckpointProposalJob implements Traceable { * * The hash is compared rather than the height alone, since a chain rebuilt after a prune reuses the block numbers. * + * This is a preflight, not a guarantee: the proposal is enqueued here and only broadcast once the publisher's wait + * for the target slot returns, and the archiver can prune in that window as it can after the transaction is sent. + * What the check buys is the common case, where the prune is already visible by the time the slot arrives. + * * Skipped whenever proposed blocks aren't pushed (`skipPushProposedBlocksToArchiver`, fisherman mode): the archiver * never held them in the first place, so their absence says nothing about a prune. */