fix(prover-node): rebuild pruned checkpoint provers and recover the epoch

A CheckpointProver's sub-tree work forks world-state per block. When an L1 reorg
prunes a base block, those fork reads fault and the prover permanently rejects
its one-shot blockProofs promise. The store previously kept the prover alive
(markPruned) so a re-add of identical content could reuse its in-flight work,
but a prune that removes the checkpoint also breaks the fork reads, so the reuse
only ever handed back a poisoned prover: a re-add or a full EpochSession recreate
re-referenced the same rejected blockProofs and failed immediately, and the node
silently abandoned the epoch until a process restart.

Drop the reuse machinery — a pruned prover cannot survive:
- CheckpointStore.cancelAndRemoveAboveBlock cancels and removes orphaned provers
  (replacing markPrunedAboveBlock); a re-add builds a fresh prover.
- Remove the pruned flag / markPruned / markCanonical from CheckpointProver and
  the SlotWatcher that only reaped lingering pruned provers.

Recover the epoch by rebuilding on re-add rather than racing the failure. A
prune unwinds world-state before the prover-node cancels the session, so a
prover mid-fork faults while live and the session goes terminal `failed`; that
race cannot be won from the session side, so make the terminal state harmless:
- SessionManager.openFullSessionIfReady drops a terminal session and rebuilds,
  so re-adding an epoch's checkpoints reopens a fresh live session over fresh
  provers (via the checkpoint/prune triggers, ungated by the tick high-water
  mark).
- runSession skips the failure-upload when the session's checkpoints no longer
  match canonical content, so a prune-invalidated failure emits no spurious
  post-mortem.

Cancel the pruned prover's in-flight work promptly: thread the abort signal into
PublicProcessor.process so a cancel stops the current block's public execution
immediately instead of running it to completion.

The expensive proving is unaffected: broker proofs are content-addressed and
survive the prune (cancelJobsOnStop defaults to false), so an identical re-add
reuses them; only witness generation, which cannot survive a prune, is redone.

Author: PhilWindle

yarn-project/end-to-end/src/single-node/proving/optimistic.parallel.test.ts

@@ -340,22 +340,20 @@ describe('single-node/proving/optimistic', () => {
         timeout: 30,
       });
 
-      // Verify the prover-node observes the prune. `markPruned()` fires reactively when
-      // the L2BlockStream emits the prune; the SlotWatcher then reaps the (now pruned)
-      // prover on its next tick (default 1s), so checking strictly for `isPruned()` would
-      // race against the reap. Identify the original by `(checkpointNumber, slot)` —
-      // checkpoint numbers refill sequentially after a reorg, so the replacement reuses
-      // the same number but lives at a different slot. Accept either state for the
-      // original: still in the store and pruned, or already reaped.
+      // Verify the prover-node observes the prune. The prune reactively cancels and removes the
+      // orphaned prover from the store when the L2BlockStream emits it, so the original should drop
+      // out of the store (or, if observed mid-race, be cancelled). Identify the original by
+      // `(checkpointNumber, slot)` — checkpoint numbers refill sequentially after a reorg, so the
+      // replacement reuses the same number but lives at a different slot.
       await retryUntil(
         () => {
           const prover = proverNode
             .getCheckpointStore()
             .listAll()
             .find(p => p.checkpoint.number === checkpointBeforeReorg && p.slotNumber === originalSlot);
-          return Promise.resolve(!prover || prover.isPruned());
+          return Promise.resolve(!prover || prover.isCancelled());
         },
-        `prover marks original checkpoint ${checkpointBeforeReorg} (slot ${originalSlot}) as pruned (or reaps it)`,
+        `prover cancels and removes original checkpoint ${checkpointBeforeReorg} (slot ${originalSlot})`,
         30,
         0.2,
       );
@@ -385,7 +383,7 @@ describe('single-node/proving/optimistic', () => {
             proverNode
               .getCheckpointStore()
               .listAll()
-              .some(p => p.checkpoint.number === replacementCheckpoint && !p.isPruned()),
+              .some(p => p.checkpoint.number === replacementCheckpoint),
           ),
         `prover re-creates sub-tree for replacement checkpoint ${replacementCheckpoint}`,
         30,

yarn-project/prover-node/src/checkpoint-store.test.ts

@@ -9,27 +9,26 @@ import { EmptyL1RollupConstants } from '@aztec/stdlib/epoch-helpers';
 
 import { mock } from 'jest-mock-extended';
 
-import { type CheckpointProverFactory, CheckpointStore } from './checkpoint-store.js';
+import { CheckpointStore } from './checkpoint-store.js';
 import type { CheckpointProver } from './job/checkpoint-prover.js';
 
 describe('CheckpointStore', () => {
-  let store: TestCheckpointStore;
-  let blockSource: ReturnType<typeof mock<Pick<L2BlockSource, 'getSyncedL2SlotNumber' | 'getL1Constants'>>>;
+  let store: CheckpointStore;
+  let blockSource: ReturnType<typeof mock<Pick<L2BlockSource, 'getL1Constants'>>>;
   /** Track stub provers we hand back from the factory. */
   const stubs: StubProver[] = [];
 
   // Single-slot epochs make every checkpoint live in its own epoch and slot range.
   const l1Constants = { ...EmptyL1RollupConstants, epochDuration: 1 };
 
   beforeEach(() => {
-    blockSource = mock<Pick<L2BlockSource, 'getSyncedL2SlotNumber' | 'getL1Constants'>>();
+    blockSource = mock<Pick<L2BlockSource, 'getL1Constants'>>();
     blockSource.getL1Constants.mockResolvedValue(l1Constants);
     stubs.length = 0;
-    store = new TestCheckpointStore(
+    store = new CheckpointStore(
       blockSource,
       // The deps are not exercised — the factory below ignores them.
       {} as any,
-      { slotWatcherPollIntervalMs: 100 },
       undefined,
       (args, _deps) => {
         const stub = makeStubProver(args.checkpoint, args.epochNumber);
@@ -51,21 +50,35 @@ describe('CheckpointStore', () => {
     expect(stubs.length).toBe(1);
   });
 
-  it('addOrUpdate is idempotent for the same content key (re-add after prune)', async () => {
+  it('addOrUpdate reuses the prover for a still-canonical duplicate registration', async () => {
+    // An at-least-once re-registration of a checkpoint that has NOT been pruned reuses the running
+    // prover rather than rebuilding its in-flight work.
     const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1 });
 
     const first = await store.addOrUpdate(cp, makeRegisterData());
-    expect(first.isPruned()).toBe(false);
-    store.markPrunedAboveBlock(BlockNumber(0));
-    expect(first.isPruned()).toBe(true);
-
-    // Re-adding the identical checkpoint (same archive root) reuses the existing prover.
     const second = await store.addOrUpdate(cp, makeRegisterData());
     expect(second).toBe(first);
-    expect(second.isPruned()).toBe(false);
     expect(stubs.length).toBe(1);
   });
 
+  it('addOrUpdate rebuilds a fresh prover for a re-add after prune', async () => {
+    const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1 });
+
+    const first = await store.addOrUpdate(cp, makeRegisterData());
+    expect(first.isCancelled()).toBe(false);
+    store.cancelAndRemoveAboveBlock(BlockNumber(0));
+    // The pruned prover is cancelled and dropped from the store.
+    expect(first.isCancelled()).toBe(true);
+    expect(store.getByCheckpoint(cp)).toBeUndefined();
+
+    // Re-adding the identical checkpoint (same archive root) constructs a fresh prover — the pruned
+    // one's forked world-state reads did not survive, so there is nothing to reuse.
+    const second = await store.addOrUpdate(cp, makeRegisterData());
+    expect(second).not.toBe(first);
+    expect(second.isCancelled()).toBe(false);
+    expect(stubs.length).toBe(2);
+  });
+
   it('addOrUpdate refuses a conflicting canonical checkpoint at the same slot', async () => {
     // Two canonical checkpoints sharing a slot would be a parallel chain. The store rejects
     // the second; the caller must prune the first (via the chain-pruned event) before the
@@ -79,17 +92,17 @@ describe('CheckpointStore', () => {
       /canonical checkpoint already occupies this slot/i,
     );
 
-    // After the predecessor is pruned, the replacement is accepted and keys to a distinct
-    // prover (different archive root → different content id).
-    store.markPrunedAboveBlock(BlockNumber(0));
-    expect(proverA.isPruned()).toBe(true);
+    // After the predecessor is pruned (cancelled and removed), the replacement is accepted and keys
+    // to a distinct prover (different archive root → different content id).
+    store.cancelAndRemoveAboveBlock(BlockNumber(0));
+    expect(proverA.isCancelled()).toBe(true);
     const proverB = await store.addOrUpdate(b, makeRegisterData());
     expect(proverB).not.toBe(proverA);
-    expect(proverB.isPruned()).toBe(false);
+    expect(proverB.isCancelled()).toBe(false);
     expect(stubs.length).toBe(2);
   });
 
-  it('markPrunedAboveBlock marks every prover holding a block above the target and returns them', async () => {
+  it('cancelAndRemoveAboveBlock cancels and removes every prover holding a block above the target and returns them', async () => {
     // Four single-block checkpoints occupying blocks 1..4 (one block each). Pruning to block 2 orphans the
     // checkpoints whose last block is above 2 — checkpoints 3 and 4 — and leaves 1 and 2 canonical.
     const cps = await timesAsync(4, i =>
@@ -102,23 +115,26 @@ describe('CheckpointStore', () => {
     for (const cp of cps) {
       await store.addOrUpdate(cp, makeRegisterData());
     }
-    const affected = store.markPrunedAboveBlock(BlockNumber(2));
+    const affected = store.cancelAndRemoveAboveBlock(BlockNumber(2));
     expect(affected.map(p => p.checkpoint.number)).toEqual([3, 4]);
+    expect(affected.every(p => p.isCancelled())).toBe(true);
+    // The orphaned provers are gone from the store; only 1 and 2 remain.
     expect(store.listCanonical().map(p => p.checkpoint.number)).toEqual([1, 2]);
   });
 
-  it('markPrunedAboveBlock marks a checkpoint whose block range straddles the target (partially orphaned)', async () => {
+  it('cancelAndRemoveAboveBlock removes a checkpoint whose block range straddles the target (partially orphaned)', async () => {
     // A single checkpoint spanning blocks 5..8. A prune to block 6 lands mid-checkpoint: the checkpoint is partially
-    // orphaned (blocks 7, 8 are gone) and must be marked, since its last block (8) is above the target.
+    // orphaned (blocks 7, 8 are gone) and must be removed, since its last block (8) is above the target.
     const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 4, startBlockNumber: 5 });
     await store.addOrUpdate(cp, makeRegisterData());
 
-    const affected = store.markPrunedAboveBlock(BlockNumber(6));
+    const affected = store.cancelAndRemoveAboveBlock(BlockNumber(6));
     expect(affected.map(p => p.checkpoint.number)).toEqual([1]);
+    expect(affected[0].isCancelled()).toBe(true);
     expect(store.listCanonical()).toEqual([]);
   });
 
-  it('reapExpired drops canonical provers whose epoch is ≤ expiredEpoch', async () => {
+  it('reapExpired drops provers whose epoch is ≤ expiredEpoch', async () => {
     // With epochDuration=1 each checkpoint's slot is also its epoch number.
     const cps = await Promise.all([
       Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(1) }),
@@ -133,74 +149,7 @@ describe('CheckpointStore', () => {
     expect(remainingNumbers).toEqual([3]);
   });
 
-  it('reapExpired leaves pruned provers in place', async () => {
-    const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(1) });
-    await store.addOrUpdate(cp, makeRegisterData());
-    store.markPrunedAboveBlock(BlockNumber(0));
-    store.reapExpired(EpochNumber(10));
-    expect(store.listAll().map(p => p.checkpoint.number)).toEqual([1]);
-  });
-
-  // ---------------- slot watcher ----------------
-
-  it('slot watcher reaps pruned provers whose slot is strictly before the synced slot', async () => {
-    const cp1 = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(1) });
-    const cp2 = await Checkpoint.random(CheckpointNumber(2), { numBlocks: 1, slotNumber: SlotNumber(2) });
-    const cp3 = await Checkpoint.random(CheckpointNumber(3), { numBlocks: 1, slotNumber: SlotNumber(3) });
-    for (const cp of [cp1, cp2, cp3]) {
-      await store.addOrUpdate(cp, makeRegisterData());
-    }
-    // Prune everything above checkpoint 0 ⇒ all three flip to pruned.
-    store.markPrunedAboveBlock(BlockNumber(0));
-    blockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(3));
-
-    await store.triggerSlotWatcherTick();
-
-    // Slots 1 and 2 are < 3 and get reaped; slot 3 is not strictly less, so it stays.
-    expect(store.listAll().map(p => p.checkpoint.number)).toEqual([3]);
-    // Reaped stubs were cancelled by the watcher.
-    expect(stubs.find(s => s.checkpoint.number === 1)!.cancelled).toBe(true);
-    expect(stubs.find(s => s.checkpoint.number === 2)!.cancelled).toBe(true);
-    expect(stubs.find(s => s.checkpoint.number === 3)!.cancelled).toBe(false);
-  });
-
-  it('slot watcher leaves canonical provers in place even when their slot is past the synced slot', async () => {
-    // Canonical provers must survive — only pruned provers are eligible for reaping.
-    const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(1) });
-    await store.addOrUpdate(cp, makeRegisterData());
-    blockSource.getSyncedL2SlotNumber.mockResolvedValue(SlotNumber(10));
-
-    await store.triggerSlotWatcherTick();
-
-    expect(store.listAll().map(p => p.checkpoint.number)).toEqual([1]);
-    expect(stubs[0].cancelled).toBe(false);
-  });
-
-  it('slot watcher no-ops when getSyncedL2SlotNumber returns undefined', async () => {
-    const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(1) });
-    await store.addOrUpdate(cp, makeRegisterData());
-    store.markPrunedAboveBlock(BlockNumber(0));
-    blockSource.getSyncedL2SlotNumber.mockResolvedValue(undefined);
-
-    await store.triggerSlotWatcherTick();
-
-    // No synced slot yet ⇒ watcher doesn't know whether the chain has moved past, so it
-    // keeps the pruned prover around for a possible re-add.
-    expect(store.listAll().map(p => p.checkpoint.number)).toEqual([1]);
-    expect(stubs[0].cancelled).toBe(false);
-  });
-
-  it('slot watcher swallows getSyncedL2SlotNumber errors instead of crashing the tick', async () => {
-    const cp = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(1) });
-    await store.addOrUpdate(cp, makeRegisterData());
-    store.markPrunedAboveBlock(BlockNumber(0));
-    blockSource.getSyncedL2SlotNumber.mockRejectedValue(new Error('archiver unavailable'));
-
-    await expect(store.triggerSlotWatcherTick()).resolves.toBeUndefined();
-    expect(store.listAll().map(p => p.checkpoint.number)).toEqual([1]);
-  });
-
-  it('listCanonicalForEpoch returns only canonical provers in the epoch slot range', async () => {
+  it('listCanonicalForEpoch returns only provers in the epoch slot range', async () => {
     // With epochDuration=1, each epoch's slot range is exactly [slot, slot].
     const cp1 = await Checkpoint.random(CheckpointNumber(1), { numBlocks: 1, slotNumber: SlotNumber(10) });
     const cp2 = await Checkpoint.random(CheckpointNumber(2), { numBlocks: 1, slotNumber: SlotNumber(11) });
@@ -220,12 +169,8 @@ type StubProver = {
   checkpoint: Checkpoint;
   slotNumber: SlotNumber;
   epochNumber: EpochNumber;
-  pruned: boolean;
   cancelled: boolean;
-  isPruned(): boolean;
   isCancelled(): boolean;
-  markPruned(): void;
-  markCanonical(): void;
   cancel(opts?: { routine?: boolean }): void;
   whenDone(): Promise<void>;
 };
@@ -237,20 +182,10 @@ function makeStubProver(checkpoint: Checkpoint, epochNumber: EpochNumber): StubP
     checkpoint,
     slotNumber: checkpoint.header.slotNumber,
     epochNumber,
-    pruned: false,
     cancelled: false,
-    isPruned() {
-      return this.pruned;
-    },
     isCancelled() {
       return this.cancelled;
     },
-    markPruned() {
-      this.pruned = true;
-    },
-    markCanonical() {
-      this.pruned = false;
-    },
     cancel() {
       this.cancelled = true;
     },
@@ -268,24 +203,3 @@ function makeRegisterData() {
     previousArchiveSiblingPath: makeTuple(ARCHIVE_HEIGHT, () => Fr.ZERO),
   };
 }
-
-/**
- * Subclass that exposes the protected `reapPrunedPastSlot` so tests can drive a single
- * SlotWatcher tick directly — avoids spinning up the underlying `RunningPromise` and
- * waiting on its polling interval.
- */
-class TestCheckpointStore extends CheckpointStore {
-  constructor(
-    blockSource: ConstructorParameters<typeof CheckpointStore>[0],
-    proverDeps: ConstructorParameters<typeof CheckpointStore>[1],
-    options: ConstructorParameters<typeof CheckpointStore>[2],
-    bindings: ConstructorParameters<typeof CheckpointStore>[3],
-    factory: CheckpointProverFactory,
-  ) {
-    super(blockSource, proverDeps, options, bindings, factory);
-  }
-
-  public triggerSlotWatcherTick(): Promise<void> {
-    return this.reapPrunedPastSlot();
-  }
-}