diff --git a/docs/docs-operate/operators/reference/changelog/v6.md b/docs/docs-operate/operators/reference/changelog/v6.md
index 357d9b7d14dd..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 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. 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 3e60c0e6fc3f..667ca1dfc984 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<ReturnType<typeof syncAndReorgNewestBucket>>) => {
+          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,264 @@ 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('prunes when the reorg merges away the bucket boundary a block ended on', 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. 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(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 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);
+      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/data_store_updater.ts b/yarn-project/archiver/src/modules/data_store_updater.ts
index 2ebe1839cfc7..baefce3f36d6 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 { InboxMessageReplacement } from '../store/message_store.js';
 
 /** Operation type for contract data updates. */
 enum Operation {
@@ -291,13 +292,47 @@ export class ArchiverDataStoreUpdater {
    *
    * @returns The pruned blocks.
    */
-  public async rewindMessagesAndPruneProposedBlocks(
+  public rewindMessagesAndPruneProposedBlocks(
     messagesSyncPoint: L1BlockId,
     firstRemovedIndex: bigint,
+  ): Promise<L2Block[]> {
+    return this.writeMessagesAndPruneProposedBlocks(
+      () => this.stores.messages.rewindMessagesTo(messagesSyncPoint, firstRemovedIndex),
+      firstRemovedIndex,
+    );
+  }
+
+  /**
+   * 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 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
+   * messages consistent with L1 and never come back to the blocks built on the leaves that are gone.
+   *
+   * @returns The pruned blocks.
+   */
+  public replaceMessagesAndPruneProposedBlocks(args: InboxMessageReplacement): Promise<L2Block[]> {
+    return this.writeMessagesAndPruneProposedBlocks(
+      () => this.stores.messages.replaceMessagesAboveBucket(args),
+      args.firstInvalidatedIndex,
+    );
+  }
+
+  /**
+   * Commits a write to the message store and the prune of the proposed blocks that consumed a message from
+   * `firstInvalidatedIndex` on, refreshing the tips cache only once both have landed.
+   */
+  private async writeMessagesAndPruneProposedBlocks(
+    writeMessages: () => Promise<void>,
+    firstInvalidatedIndex: bigint | undefined,
   ): Promise<L2Block[]> {
     const prunedBlocks = await this.stores.db.transactionAsync(async () => {
-      await this.stores.messages.rewindMessagesTo(messagesSyncPoint, firstRemovedIndex);
-      return await this.removeProposedBlocksConsumingMessagesFrom(firstRemovedIndex);
+      await writeMessages();
+      return firstInvalidatedIndex === undefined
+        ? []
+        : await this.removeProposedBlocksConsumingMessagesFrom(firstInvalidatedIndex);
     });
     await this.l2TipsCache?.refresh();
     return prunedBlocks;
@@ -305,25 +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 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, 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<L2Block[]> {
+  private async removeProposedBlocksConsumingMessagesFrom(firstInvalidatedIndex: bigint): Promise<L2Block[]> {
     const [checkpointedBlockNumber, latestBlockNumber] = await Promise.all([
       this.stores.blocks.getCheckpointedL2BlockNumber(),
       this.stores.blocks.getLatestL2BlockNumber(),
@@ -339,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 [];
     }
@@ -352,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 b5d53cf1ef68..fd3301c43f75 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,
@@ -459,7 +460,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 +488,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 +549,7 @@ export class ArchiverL1Synchronizer implements Traceable {
    */
   private async ensureBucketsAreCanonical(
     messagesSynchedTo: L1BlockId,
+    currentL1Block: L1BlockId,
     finalizedL1Block: L1BlockId | undefined,
   ): Promise<BucketsCanonicality> {
     let isNewest = true;
@@ -566,7 +568,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 +612,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<L1BlockId | undefined> {
+  private async rollbackReorgedBucket(
+    reorgedBucketSeq: bigint,
+    reorg: BucketReorg,
+    currentL1Block: L1BlockId,
+    finalizedL1Block: L1BlockId | undefined,
+  ): Promise<L1BlockId | undefined> {
     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 +626,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 +652,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<L1BlockId> {
+    currentL1Block: L1BlockId,
+    finalizedL1Block: L1BlockId | undefined,
+  ): Promise<L1BlockId | undefined> {
     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 +681,227 @@ 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,
+    );
+    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. 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,
     );
-    return this.rewindMessagesTo(syncPoint, lastCanonicalBucket ? lastCanonicalBucket.lastMessageIndex + 1n : 0n);
+
+    const prunedBlocks = await this.updater.replaceMessagesAndPruneProposedBlocks({
+      lastCanonicalBucketSeq: lastCanonicalBucket?.seq,
+      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
+      // 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,
+      firstInvalidatedIndex,
+      messageCount: fetched.messages.length,
+    });
+    this.reportPrunedProposedBlocks(prunedBlocks, firstInvalidatedIndex);
+    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 messages: InboxMessage[] = [];
+    const batchTails: InboxMessage[] = [];
+    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;
+      }
+      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(batch.searchEndBlock, 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 (reachedHead) {
+      return { messages, syncPoint: currentL1Block };
+    }
+    try {
+      return {
+        messages,
+        syncPoint: { l1BlockNumber: lastFetchedL1Block, l1BlockHash: await this.getL1BlockHash(lastFetchedL1Block) },
+      };
+    } catch (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
+   * 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<boolean | undefined> {
+    const nextL1BlockNumber = l1BlockNumber + 1n;
+    try {
+      const { timestamp } = await getL1Block(this.publicClient, nextL1BlockNumber);
+      return timestamp !== message.bucketTimestamp;
+    } catch (error) {
+      this.warnBucketNotVerifiable(message.bucketSeq, nextL1BlockNumber, error);
+      return undefined;
+    }
+  }
+
+  /**
+   * 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 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 findFirstInvalidatedMessageIndex(
+    fromIndex: bigint,
+    canonicalMessages: InboxMessage[],
+  ): Promise<bigint | undefined> {
+    // 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 = 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;
   }
 
   /**
@@ -712,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;
     }
@@ -720,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, {
@@ -760,22 +991,21 @@ export class ArchiverL1Synchronizer implements Traceable {
    */
   private async retrieveAndStoreMessages(
     syncFrom: L1BlockId,
-    toL1Block: bigint,
+    currentL1Block: L1BlockId,
     finalizedL1Block: L1BlockId | undefined,
   ): Promise<MessageFetchOutcome> {
     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;
@@ -787,11 +1017,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';
@@ -823,7 +1052,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';
         }
diff --git a/yarn-project/archiver/src/store/message_store.test.ts b/yarn-project/archiver/src/store/message_store.test.ts
index d25d75d4ca4e..3cf21ebfb2e4 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,
+          firstInvalidatedIndex: 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,
+          firstInvalidatedIndex: 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,
+          firstInvalidatedIndex: 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,
+          firstInvalidatedIndex: 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..3fba05442fee 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 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. */
+  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;
@@ -453,13 +467,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 +491,49 @@ 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<void> {
+    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: `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
+   * 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.
+   */
+  public replaceMessagesAboveBucket(args: InboxMessageReplacement): Promise<void> {
+    const { lastCanonicalBucketSeq, firstInvalidatedIndex, messages, syncPoint, finalizedL1Block } = args;
+    return this.db.transactionAsync(async () => {
+      if (firstInvalidatedIndex !== undefined) {
+        await this.removeL1ToL2Messages(firstInvalidatedIndex);
+      }
+      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.
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..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
@@ -5,6 +5,7 @@ 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 type { Delayer } from '@aztec/ethereum/l1-tx-utils';
 import type { ChainMonitor } from '@aztec/ethereum/test';
 import type { ExtendedViemWalletClient } from '@aztec/ethereum/types';
@@ -18,7 +19,7 @@ import 'jest-extended';
 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
@@ -206,4 +207,106 @@ 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.
+    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.
+    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,
+    );
+  });
 });
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<Blob[]> {
+  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,
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..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
@@ -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<L1ToL2MessageSource>;
   let l2BlockSource: MockProxy<L2BlockSource>;
   let blockSink: MockProxy<L2BlockSink & ProposedCheckpointSink>;
+  /** Blocks the job pushed to the archiver, which serves them back until a test prunes them. */
+  let localBlocks: Map<number, L2Block>;
   let slasherClient: MockProxy<SlasherClientInterface>;
   let dateProvider: ManualDateProvider;
   let metrics: MockProxy<SequencerMetrics>;
@@ -300,7 +303,7 @@ describe('CheckpointProposalJob', () => {
     });
 
     blockSink = mock<L2BlockSink & ProposedCheckpointSink>();
-    blockSink.addBlock.mockResolvedValue(undefined);
+    localBlocks = serveAddedBlocksFromSource(l2BlockSource, blockSink);
     blockSink.addProposedCheckpoint.mockResolvedValue(undefined);
 
     validatorClient = mock<ValidatorClient>();
@@ -1838,6 +1841,72 @@ 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('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());
+      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<L2BlockSink & ProposedCheckpointSink>();
-    blockSink.addBlock.mockResolvedValue(undefined);
+    serveAddedBlocksFromSource(l2BlockSource, blockSink);
 
     validatorClient = mock<ValidatorClient>();
     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..3ec51cc4ea82 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,54 @@ 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.
+   *
+   * 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.
+   */
+  private async checkpointBlocksAreStillLocal(checkpoint: Checkpoint): Promise<boolean> {
+    if (this.config.skipPushProposedBlocksToArchiver || this.config.fishermanMode) {
+      return true;
+    }
+    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<void> {
     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<L2BlockSource & L2BlockSink & ProposedCheckpointSink>({
-      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<L1ToL2MessageSource>({
       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<Checkpoint> {
     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<Checkpoint> {
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<L1ToL2MessageSource>, 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<L2BlockSource>,
+  blockSink: MockProxy<L2BlockSink>,
+  fallback?: BlockData,
+): Map<number, L2Block> {
+  const blocks = new Map<number, L2Block>();
+  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;
+}