From 7db27fe569e7f7b1a8307bcd5795618f504c860b Mon Sep 17 00:00:00 2001 From: Phil Windle Date: Fri, 10 Jul 2026 13:35:40 +0000 Subject: [PATCH] fix(prover-node): rebuild pruned checkpoint provers and recover the epoch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. - stop() awaits teardowns still in flight for removed provers, not just live ones. - Drop the now-redundant "canonical" method naming: listCanonical* -> list*, canonicalCheckpointsForSpec -> checkpointsForSpec. 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 rebuilds over current canonical content; recreateInvalidSessions deletes terminal sessions before it runs. - runSession skips the failure-upload when the session's checkpoints no longer match the store (best-effort; inherently races the prune, as the store lags the world-state unwind). 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. --- .../proving/optimistic.parallel.test.ts | 20 +- yarn-project/prover-node/README.md | 113 ++++++----- .../prover-node/src/checkpoint-store.test.ts | 178 +++++------------- .../prover-node/src/checkpoint-store.ts | 148 ++++++--------- .../src/job/checkpoint-prover.test.ts | 74 +++++--- .../prover-node/src/job/checkpoint-prover.ts | 52 ++--- .../prover-node/src/job/epoch-session.test.ts | 3 - yarn-project/prover-node/src/metrics.ts | 2 +- .../prover-node/src/prover-node.test.ts | 14 +- yarn-project/prover-node/src/prover-node.ts | 7 +- .../prover-node/src/session-manager.test.ts | 162 ++++++++++++---- .../prover-node/src/session-manager.ts | 45 ++++- 12 files changed, 408 insertions(+), 410 deletions(-) diff --git a/yarn-project/end-to-end/src/single-node/proving/optimistic.parallel.test.ts b/yarn-project/end-to-end/src/single-node/proving/optimistic.parallel.test.ts index 704acfdbb7ba..f797c763f5b1 100644 --- a/yarn-project/end-to-end/src/single-node/proving/optimistic.parallel.test.ts +++ b/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, @@ -877,7 +875,7 @@ describe('single-node/proving/optimistic', () => { // The session manager constructs a full session over the canonical content for the // anchored epoch when it completes, then proves it; the store retains the provers // until expiry. - const epochCheckpointsInStore = await proverNode.getCheckpointStore().listCanonicalForEpoch(epoch); + const epochCheckpointsInStore = await proverNode.getCheckpointStore().listForEpoch(epoch); const storedNumbers = new Set(epochCheckpointsInStore.map(p => p.checkpoint.number)); for (const n of preSpawnCheckpointNumbers) { expect(storedNumbers.has(n)).toBe(true); diff --git a/yarn-project/prover-node/README.md b/yarn-project/prover-node/README.md index d9521c69c340..abafeae777ee 100644 --- a/yarn-project/prover-node/README.md +++ b/yarn-project/prover-node/README.md @@ -35,7 +35,6 @@ flowchart TB SessionManager --> EpochTicker[(periodic tick)] SessionManager --> FullSessions[(fullSessions)] SessionManager --> PartialSessions[(partialSessions)] - CheckpointStore --> SlotWatcher FullSessions -.referenced checkpoints.-> CheckpointStore PartialSessions -.referenced checkpoints.-> CheckpointStore FullSessions --> TopTreeJob @@ -56,8 +55,9 @@ The prover-node splits responsibility between four classes: - **`CheckpointStore`** — a registry of `CheckpointProver` instances keyed by `(checkpointNumber, slot, archiveRoot)`. Each `CheckpointProver` runs its own sub-tree pipeline (tx gather → block processing → block-rollup proofs), starting eagerly the moment a - checkpoint is registered. The store is the single source of canonical-vs-pruned - checkpoint content that `EpochSession`s query when assembling their subsets. + checkpoint is registered. The store holds the canonical checkpoint content that + `EpochSession`s query when assembling their subsets; a prune cancels and removes the affected + provers, so every prover in the store is canonical. - **`SessionManager`** — owns every live `EpochSession`, the serial reconcile queue, the periodic tick, and all `EpochSession` lifecycle decisions. `ProverNode` calls into it via `onCheckpointAdded`, `onPrune`, and `startProof`. Every trigger it receives is @@ -81,43 +81,38 @@ A `CheckpointProver` is content-addressed by `(checkpoint.number, slot, archiveR where `archiveRoot` is the checkpoint's own archive root (its post-state). Keying on the post-state makes the identity precise: two checkpoints are "the same" iff they produce the same archive — so a reorg branch, or a replacement built on the same predecessor but -with different content, yields a different archive root and a distinct `CheckpointProver`, while an -identical re-add collapses to the same `CheckpointProver` and reuses its in-flight work. +with different content, yields a different archive root and a distinct `CheckpointProver`. A prune +cancels and removes a prover — its sub-tree work forks world-state per block and cannot survive the +prune — so a re-add, even of identical content, constructs a fresh `CheckpointProver` (block-rollup +jobs already completed in the proving broker are still reused, as they are content-addressed). ```mermaid stateDiagram-v2 [*] --> Created Created --> Proving: gather + execute Proving --> Proven: sub-tree resolves blockProofs - Proving --> Cancelled: cancel() + Proving --> Cancelled: cancel() (prune / shutdown / epoch-session error) + Proven --> Cancelled: cancel() (prune / reap) Proven --> Reaped: reapExpired(epoch) Cancelled --> [*] - - state "Pruned (side)" as Pruned - Proving --> Pruned: markPruned() - Pruned --> Proving: markCanonical() - Proven --> Pruned: markPruned() - Pruned --> Reaped: SlotWatcher (slot < syncedSlot) ``` -The **`Pruned`** state is a side flag, not a place in the main lifecycle: sub-tree -proving keeps running underneath, so a brief reorg that prunes and immediately -re-adds the same checkpoint avoids any re-proving. The flag only gates *eligibility* -to be included in an `EpochSession` — `EpochSession`s ask the store for *canonical* (non-pruned) -checkpoints when assembling their subsets. - -### Reaping rules - -- **Pruned**: the `SlotWatcher` (a `RunningPromise` polling - `l2BlockSource.getSyncedL2SlotNumber`) reaps a pruned `CheckpointProver` when the chain's - synced slot has moved past the `CheckpointProver`'s slot. Once the chain is past that slot, - a re-add with the same content is impossible. -- **Canonical**: `CheckpointStore.reapExpired(expiredEpoch)` drops any canonical - `CheckpointProver` whose epoch is at or below the supplied expired epoch. Once an epoch's - proof-submission window has closed, its proof can no longer be accepted on L1, - so the `CheckpointProver` is no longer needed. -- **Cancelled**: removed immediately by whichever path called `cancel()` (store - shutdown, prune past-slot, `EpochSession` error). +A prune **cancels and removes** the affected `CheckpointProver`s from the store: a prover's +sub-tree work forks world-state per block, and an L1 prune of a base block faults those reads, +so there is nothing to preserve across a reorg. A re-add — even of identical content — therefore +constructs a fresh `CheckpointProver`. `EpochSession`s ask the store for the checkpoints in a slot +range; every prover in the store is canonical. + +### Removal rules + +- **Pruned**: `CheckpointStore.cancelAndRemoveAboveBlock(target)` cancels and removes every + `CheckpointProver` whose last block is above the prune target, the moment a `chain-pruned` + event arrives. +- **Expired**: `CheckpointStore.reapExpired(expiredEpoch)` drops any `CheckpointProver` whose + epoch is at or below the supplied expired epoch. Once an epoch's proof-submission window has + closed, its proof can no longer be accepted on L1, so the `CheckpointProver` is no longer needed. +- **Shutdown**: `CheckpointStore.stop()` cancels every remaining prover and awaits its teardown + (including teardowns still in flight for provers already removed by a prune or reap). ### Eager tx gathering @@ -288,8 +283,8 @@ sequenceDiagram alt content key new CS->>CP: new CheckpointProver(args) CP->>CP: eager gather + sub-tree start - else content key matches - CS->>CP: markCanonical() + else content key already live + CS->>CP: reuse existing prover end PN->>SM: onCheckpointAdded(epoch) SM->>SM: queue reconcile({kind:'checkpoint', epoch}) @@ -306,9 +301,9 @@ sequenceDiagram participant CS as CheckpointStore participant SM as SessionManager - L2->>PN: chain-pruned{checkpoint} - PN->>CS: markPrunedAfter(checkpoint.number) - CS->>CS: flip every CheckpointProver above threshold to pruned (sub-tree keeps running) + L2->>PN: chain-pruned{block} + PN->>CS: cancelAndRemoveAboveBlock(prunedToBlock) + CS->>CS: cancel + remove every CheckpointProver whose last block is above the target PN->>SM: onPrune(affectedEpochs) SM->>SM: queue reconcile({kind:'prune', affectedEpochs}) SM->>SM: walk EpochSessions, cancel-and-recreate those with shifted content @@ -371,30 +366,28 @@ indexing) leave the mark in place and the next tick retries. ### checkpoint-added → prune → checkpoint-added (reorg resilience) -State: epoch N has checkpoints c1..c4 all canonical (slots s1..s4). `fullSessions[N]` -holds `EpochSession` **A** with spec `{kind:'full', N, fromSlot:s1, toSlot:s4}`, referencing -checkpoints `[c1, c2, c3, c4]`. +State: epoch N has checkpoints c1..c4 (slots s1..s4). `fullSessions[N]` holds `EpochSession` +**A** with spec `{kind:'full', N, fromSlot:s1, toSlot:s4}`, referencing `[c1, c2, c3, c4]`. -1. **chain-pruned arrives, target c3.** Store flips c4 to pruned. Reconcile fires: - for `EpochSession` A, canonical content for `(s1, s4)` is now `[c1, c2, c3]` (c4 pruned). - The frozen set `[c1, c2, c3, c4]` no longer matches → `A.cancel('canonical content - changed')`. Epoch N still complete on L1 → reconcile constructs `EpochSession` **B** with - the same spec `{full, N, s1, s4}` but checkpoints `[c1, c2, c3]`. +1. **chain-pruned arrives, target c3.** The store cancels and removes c4 (its last block is + above the prune target). Reconcile fires: `EpochSession` A's frozen set `[c1, c2, c3, c4]` + includes the now-cancelled c4, so it no longer matches the store's `[c1, c2, c3]` → + `A.cancel('canonical content changed')`. Epoch N still complete on L1 → reconcile constructs + `EpochSession` **B** with the same spec `{full, N, s1, s4}` but checkpoints `[c1, c2, c3]`. 2. **`EpochSession` B starts top-tree proving over [c1, c2, c3].** -3. **chain-checkpointed arrives, target c4_re (same content key as old c4).** The - store finds the existing `CheckpointProver` at `(c4.number, s4, c4.archive.root)` - and calls `markCanonical()`. The sub-tree work that never stopped is visible to - `EpochSession`s again. (A re-add with *different* content would have a different archive - root and so get a fresh `CheckpointProver` instead.) +3. **chain-checkpointed arrives, target c4_re (same content key as old c4).** The old c4 + prover was removed on the prune, so the store constructs a **fresh** `CheckpointProver` for + c4_re and starts its sub-tree work from scratch. (Its block-rollup jobs are content-addressed + in the proving broker, so any the old c4 already completed are reused rather than re-proved.) -4. **Reconcile fires.** `EpochSession` B's canonical content for `(s1, s4)` is now `[c1, c2, - c3, c4]`, doesn't match its frozen `[c1, c2, c3]` → `B.cancel(...)`. Construct - `EpochSession` **C** with same spec but checkpoints `[c1, c2, c3, c4]`. +4. **Reconcile fires.** `EpochSession` B's content for `(s1, s4)` is now `[c1, c2, c3, c4_re]`, + doesn't match its frozen `[c1, c2, c3]` → `B.cancel(...)`. Construct `EpochSession` **C** with + the same spec but checkpoints `[c1, c2, c3, c4_re]`. -5. **`EpochSession` C reuses the long-lived c1..c4 `CheckpointProver` instances.** Sub-tree - work may already be complete; only the top-tree is recomputed. The chonk cache +5. **`EpochSession` C reuses the long-lived c1..c3 `CheckpointProver` instances and the fresh + c4_re.** Only c4_re's witnesses are regenerated; the top-tree is recomputed. The chonk cache survived the reorg because no epoch in this range has expired yet. @@ -466,19 +459,19 @@ Keying by tx hash makes the cache survive any reorg up to finality; releasing on finality means we don't grow the cache indefinitely while still keeping every reorg-relevant proof. -### Why does the slot watcher only reap pruned `CheckpointProver`s? +### Why is a pruned `CheckpointProver` removed rather than kept for a possible re-add? -Canonical `CheckpointProver`s can't be reaped on a slot heuristic — they're still part of the -proven-chain story. Pruned `CheckpointProver`s, on the other hand, are only kept around in -case the chain re-adds the same content; once the synced slot has moved past, that -re-add is impossible, and the `CheckpointProver` can go. Finality is the right signal for -canonical reaping, because finality is the only state that rules out future reorgs. +A prover's sub-tree work forks world-state per block, and an L1 prune of a base block faults +those in-flight reads — so the work cannot survive the prune, and there is nothing to preserve +by keeping the prover around. Removing it on prune and rebuilding on re-add is therefore both +correct and simpler; the expensive block-rollup proofs are still reused when content re-appears, +because the proving broker is content-addressed independently of the prover's lifetime. ## Configuration | Env var | Description | |---|---| -| `PROVER_NODE_POLLING_INTERVAL_MS` | Polling interval for the L2BlockStream, the checkpoint-store slot watcher, and the SessionManager periodic tick. Default 1000 ms. | +| `PROVER_NODE_POLLING_INTERVAL_MS` | Polling interval for the L2BlockStream and the SessionManager periodic tick. Default 1000 ms. | | `PROVER_NODE_MAX_PENDING_JOBS` | Cap on the number of non-terminal `EpochSession`s (full + partial). When at limit, reconcile defers opening new full `EpochSession`s; explicit `startProof` calls throw. | | `PROVER_NODE_EPOCH_PROVING_DELAY_MS` | Optional sleep at the start of each `EpochSession`, before the TopTreeJob is constructed. Used in tests to give late events time to land. | | `TX_GATHERING_TIMEOUT_MS` | Per-block tx gather deadline used by each `CheckpointProver`. | diff --git a/yarn-project/prover-node/src/checkpoint-store.test.ts b/yarn-project/prover-node/src/checkpoint-store.test.ts index 0c9c98e8de4f..bd823eb83847 100644 --- a/yarn-project/prover-node/src/checkpoint-store.test.ts +++ b/yarn-project/prover-node/src/checkpoint-store.test.ts @@ -9,12 +9,12 @@ 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>>; + let store: CheckpointStore; + let blockSource: ReturnType>>; /** Track stub provers we hand back from the factory. */ const stubs: StubProver[] = []; @@ -22,14 +22,13 @@ describe('CheckpointStore', () => { const l1Constants = { ...EmptyL1RollupConstants, epochDuration: 1 }; beforeEach(() => { - blockSource = mock>(); + blockSource = mock>(); 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 @@ -76,20 +89,20 @@ describe('CheckpointStore', () => { const proverA = await store.addOrUpdate(a, makeRegisterData()); await expect(store.addOrUpdate(b, makeRegisterData())).rejects.toThrow( - /canonical checkpoint already occupies this slot/i, + /a different 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(store.listCanonical().map(p => p.checkpoint.number)).toEqual([1, 2]); + expect(affected.every(p => p.isCancelled())).toBe(true); + // The orphaned provers are gone from the store; only 1 and 2 remain. + expect(store.list().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(store.listCanonical()).toEqual([]); + expect(affected[0].isCancelled()).toBe(true); + expect(store.list()).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,82 +149,15 @@ 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('listForEpoch 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) }); await store.addOrUpdate(cp1, makeRegisterData()); await store.addOrUpdate(cp2, makeRegisterData()); - const epoch10 = await store.listCanonicalForEpoch(EpochNumber(10)); - const epoch11 = await store.listCanonicalForEpoch(EpochNumber(11)); + const epoch10 = await store.listForEpoch(EpochNumber(10)); + const epoch11 = await store.listForEpoch(EpochNumber(11)); expect(epoch10.map(p => p.checkpoint.number)).toEqual([1]); expect(epoch11.map(p => p.checkpoint.number)).toEqual([2]); }); @@ -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; }; @@ -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[0], - proverDeps: ConstructorParameters[1], - options: ConstructorParameters[2], - bindings: ConstructorParameters[3], - factory: CheckpointProverFactory, - ) { - super(blockSource, proverDeps, options, bindings, factory); - } - - public triggerSlotWatcherTick(): Promise { - return this.reapPrunedPastSlot(); - } -} diff --git a/yarn-project/prover-node/src/checkpoint-store.ts b/yarn-project/prover-node/src/checkpoint-store.ts index 4006041c6950..c675c1fe1a86 100644 --- a/yarn-project/prover-node/src/checkpoint-store.ts +++ b/yarn-project/prover-node/src/checkpoint-store.ts @@ -1,6 +1,5 @@ import type { BlockNumber, CheckpointNumber, EpochNumber, SlotNumber } from '@aztec/foundation/branded-types'; import { type Logger, type LoggerBindings, createLogger } from '@aztec/foundation/log'; -import { RunningPromise } from '@aztec/foundation/promise'; import type { L2BlockSource } from '@aztec/stdlib/block'; import type { Checkpoint } from '@aztec/stdlib/checkpoint'; import { type L1RollupConstants, getEpochAtSlot, getSlotRangeForEpoch } from '@aztec/stdlib/epoch-helpers'; @@ -19,54 +18,64 @@ export type CheckpointProverFactory = (args: CheckpointProverArgs, deps: Checkpo * * The store survives every epoch / session boundary. A prover lives from its first * `addOrUpdate` call until either: - * - it has been pruned and the L2 chain has moved past its slot (no re-add possible), or + * - its checkpoint is pruned by an L1 reorg (`cancelAndRemoveAboveBlock`), or * - its epoch's proof-submission window has closed (`reapExpired`), so the proof could no * longer be accepted on L1 even if produced. * - * A re-add of a checkpoint that matches an existing prover's content key reuses the - * existing prover (and flips it back to canonical); the in-flight sub-tree work never - * stops, so a prune-then-re-add of the same content avoids re-proving entirely. + * A prover's sub-tree work forks world-state per block and does not survive a prune of a base + * block, so there is nothing to preserve across a reorg: a pruned prover is cancelled and + * dropped, and a re-add (even of identical content) constructs a fresh prover. */ export class CheckpointStore { private readonly provers = new Map(); - private readonly slotWatcher: RunningPromise; + /** + * Teardowns of provers already removed from `provers` (by prune or reap), awaited on `stop()`. + * Keyed by a monotonic id rather than the prover's content id: a prune-then-re-add can leave two + * teardowns for the same content id in flight at once, which a content-id key would clobber. + */ + private readonly pendingTeardowns = new Map>(); + private nextTeardownId = 0; private readonly log: Logger; constructor( - private readonly l2BlockSource: Pick, + private readonly l2BlockSource: Pick, private readonly proverDeps: Omit, - private readonly options: { slotWatcherPollIntervalMs: number }, bindings?: LoggerBindings, private readonly proverFactoryFn: CheckpointProverFactory = (args, deps) => new CheckpointProver(args, deps), ) { this.log = createLogger('prover-node:checkpoint-store', bindings); - this.slotWatcher = new RunningPromise( - () => this.reapPrunedPastSlot(), - this.log, - this.options.slotWatcherPollIntervalMs, - ); } public start(): Promise { - this.slotWatcher.start(); return Promise.resolve(); } public async stop(): Promise { - await this.slotWatcher.stop(); - // Cancel every live prover; await teardown. + // Cancel every live prover, then await both their teardown and any still in flight for provers + // already removed by a prune or reap. const provers = Array.from(this.provers.values()); this.provers.clear(); for (const prover of provers) { prover.cancel(); } - await Promise.allSettled(provers.map(p => p.whenDone())); + await Promise.allSettled([...provers.map(p => p.whenDone()), ...this.pendingTeardowns.values()]); + } + + /** + * Tracks the teardown of a prover just removed from the store so `stop()` can await it. The entry + * removes itself once teardown settles, so the map stays bounded by the number in flight. + */ + private trackTeardown(prover: CheckpointProver): void { + const id = this.nextTeardownId++; + const done = prover.whenDone(); + this.pendingTeardowns.set(id, done); + void done.finally(() => this.pendingTeardowns.delete(id)); } /** * Registers a checkpoint with the store. If a prover already exists for the - * `(number, slot, archive root)` content key, it is reused and marked canonical; - * otherwise a new prover is constructed. + * `(number, slot, archive root)` content key it is reused (an at-least-once re-registration of + * still-canonical content); otherwise a new prover is constructed. */ public async addOrUpdate(checkpoint: Checkpoint, data: RegisterCheckpointData): Promise { const l1Constants = await this.l2BlockSource.getL1Constants(); @@ -75,18 +84,18 @@ export class CheckpointStore { const existing = this.provers.get(id); if (existing) { - existing.markCanonical(); return existing; } - // At most one canonical checkpoint per slot. A different canonical checkpoint at the - // same slot means the caller forgot to prune the old chain before adding the replacement - // — surface it rather than silently creating a parallel canonical chain. + // At most one canonical checkpoint per slot. A different checkpoint at the same slot means the + // caller forgot to prune the old chain before adding the replacement — surface it rather than + // silently creating a parallel canonical chain. A pruned checkpoint has already been removed, + // so every prover still in the store is canonical. for (const prover of this.provers.values()) { - if (prover.slotNumber === checkpoint.header.slotNumber && !prover.isPruned()) { + if (prover.slotNumber === checkpoint.header.slotNumber) { throw new Error( `Cannot add checkpoint ${checkpoint.number} (archive ${checkpoint.archive.root}) at slot ${checkpoint.header.slotNumber}: ` + - `a different canonical checkpoint already occupies this slot. Prune it first.`, + `a different checkpoint already occupies this slot. Prune it first.`, ); } } @@ -97,20 +106,25 @@ export class CheckpointStore { } /** - * Marks every canonical prover that holds a block above the prune target as pruned. A checkpoint is orphaned by a - * prune to block `targetBlockNumber` iff its last block sits above the target — including a checkpoint whose range - * straddles the target (partially orphaned), which block-range marking catches without boundary ambiguity. Keying - * off the surviving block number (rather than a checkpoint number) is correct even when the source has already + * Cancels and removes every prover that holds a block above the prune target. A checkpoint is orphaned by a prune to + * block `targetBlockNumber` iff its last block sits above the target — including a checkpoint whose range straddles + * the target (partially orphaned), which block-range marking catches without boundary ambiguity. Keying off the + * surviving block number (rather than a checkpoint number) is correct even when the source has already * re-checkpointed past the divergence: the prune event reports the highest surviving block, which by construction * survives on the source, whereas the source's current checkpointed tip can sit above the prune target. - * Sub-tree work keeps running so a re-add of the same content can pick it up. Returns the affected provers. + * + * The prover's in-flight sub-tree work forks world-state per block and faults once its base block is pruned, so it + * cannot be reused; it is cancelled (aborting the fork reads) and dropped. A subsequent re-add constructs a fresh + * prover. Returns the removed provers. */ - public markPrunedAboveBlock(targetBlockNumber: BlockNumber): CheckpointProver[] { + public cancelAndRemoveAboveBlock(targetBlockNumber: BlockNumber): CheckpointProver[] { const affected: CheckpointProver[] = []; - for (const prover of this.provers.values()) { + for (const [id, prover] of Array.from(this.provers.entries())) { const lastBlockNumber = prover.checkpoint.blocks.at(-1)!.number; - if (lastBlockNumber > targetBlockNumber && !prover.isPruned()) { - prover.markPruned(); + if (lastBlockNumber > targetBlockNumber) { + prover.cancel(); + this.trackTeardown(prover); + this.provers.delete(id); affected.push(prover); } } @@ -118,20 +132,17 @@ export class CheckpointStore { } /** - * Drops canonical (non-pruned) provers whose epoch is at or below the supplied expired - * epoch. Once an epoch's proof-submission window has closed, its proof can no longer be - * accepted on L1, so the prover is no longer needed. + * Drops provers whose epoch is at or below the supplied expired epoch. Once an epoch's + * proof-submission window has closed, its proof can no longer be accepted on L1, so the + * prover is no longer needed. */ public reapExpired(expiredEpoch: EpochNumber): void { const reaped: { id: string; checkpointNumber: CheckpointNumber; epochNumber: EpochNumber }[] = []; for (const [id, prover] of Array.from(this.provers.entries())) { - if (prover.isPruned()) { - continue; - } if (prover.epochNumber <= expiredEpoch) { reaped.push({ id, checkpointNumber: prover.checkpoint.number, epochNumber: prover.epochNumber }); prover.cancel({ routine: true }); - void prover.whenDone(); + this.trackTeardown(prover); this.provers.delete(id); } } @@ -154,63 +165,28 @@ export class CheckpointStore { return this.provers.get(CheckpointProver.idFor(checkpoint)); } - /** Every prover currently in the store (canonical and pruned), in insertion order. */ + /** Every prover currently in the store, in insertion order. */ public listAll(): CheckpointProver[] { return Array.from(this.provers.values()); } - /** Canonical (non-pruned) provers in the store, sorted by checkpoint number. */ - public listCanonical(): CheckpointProver[] { - return Array.from(this.provers.values()) - .filter(p => !p.isPruned()) - .sort((a, b) => a.checkpoint.number - b.checkpoint.number); + /** Provers in the store, sorted by checkpoint number. */ + public list(): CheckpointProver[] { + return Array.from(this.provers.values()).sort((a, b) => a.checkpoint.number - b.checkpoint.number); } /** - * Canonical provers whose slot is in the supplied epoch's slot range, sorted by - * checkpoint number. + * Provers whose slot is in the supplied epoch's slot range, sorted by checkpoint number. */ - public async listCanonicalForEpoch(epoch: EpochNumber): Promise { + public async listForEpoch(epoch: EpochNumber): Promise { const l1Constants = await this.l2BlockSource.getL1Constants(); const [fromSlot, toSlot] = getSlotRangeForEpoch(epoch, l1Constants); - return this.listCanonicalInSlotRange(fromSlot, toSlot); + return this.listInSlotRange(fromSlot, toSlot); } - /** Canonical provers whose slot falls within `[fromSlot, toSlot]`, sorted by checkpoint number. */ - public listCanonicalInSlotRange(fromSlot: SlotNumber, toSlot: SlotNumber): CheckpointProver[] { - return this.listCanonical().filter(p => p.slotNumber >= fromSlot && p.slotNumber <= toSlot); - } - - /** - * SlotWatcher tick: reap pruned provers whose slot has passed the chain's synced - * slot. Once the chain has moved past, no re-add can revive the prover and its - * content key is unique enough that an actual re-add would create a new entry. - * - * Protected so unit tests can drive a single tick without spinning up the - * `RunningPromise` and waiting on its interval. - */ - protected async reapPrunedPastSlot(): Promise { - let syncedSlot: SlotNumber | undefined; - try { - syncedSlot = await this.l2BlockSource.getSyncedL2SlotNumber(); - } catch (err) { - this.log.debug(`SlotWatcher could not read synced slot`, { error: `${err}` }); - return; - } - if (syncedSlot === undefined) { - return; - } - for (const [id, prover] of Array.from(this.provers.entries())) { - if (prover.isPruned() && prover.slotNumber < syncedSlot) { - this.log.info(`Reaping pruned CheckpointProver ${id}: slot ${prover.slotNumber} < synced ${syncedSlot}`, { - checkpointNumber: prover.checkpoint.number, - slotNumber: prover.slotNumber, - }); - prover.cancel(); - void prover.whenDone(); - this.provers.delete(id); - } - } + /** Provers whose slot falls within `[fromSlot, toSlot]`, sorted by checkpoint number. */ + public listInSlotRange(fromSlot: SlotNumber, toSlot: SlotNumber): CheckpointProver[] { + return this.list().filter(p => p.slotNumber >= fromSlot && p.slotNumber <= toSlot); } } diff --git a/yarn-project/prover-node/src/job/checkpoint-prover.test.ts b/yarn-project/prover-node/src/job/checkpoint-prover.test.ts index 4a19f8a9c9a0..f516b5f33485 100644 --- a/yarn-project/prover-node/src/job/checkpoint-prover.test.ts +++ b/yarn-project/prover-node/src/job/checkpoint-prover.test.ts @@ -96,7 +96,6 @@ describe('CheckpointProver', () => { expect(prover.l1ToL2Messages).toEqual([]); expect(prover.isCancelled()).toBe(false); expect(prover.isCompleted()).toBe(false); - expect(prover.isPruned()).toBe(false); await cleanup(prover); }); @@ -108,31 +107,6 @@ describe('CheckpointProver', () => { }); }); - // ---------------- prune/canonical flag ---------------- - - describe('markPruned / markCanonical', () => { - it('markPruned flips isPruned() and is idempotent', async () => { - const prover = makeProver(); - expect(prover.isPruned()).toBe(false); - prover.markPruned(); - expect(prover.isPruned()).toBe(true); - prover.markPruned(); - expect(prover.isPruned()).toBe(true); - await cleanup(prover); - }); - - it('markCanonical restores isPruned() to false and is idempotent on a non-pruned prover', async () => { - const prover = makeProver(); - prover.markPruned(); - prover.markCanonical(); - expect(prover.isPruned()).toBe(false); - // No-op when already canonical. - prover.markCanonical(); - expect(prover.isPruned()).toBe(false); - await cleanup(prover); - }); - }); - // ---------------- cancellation ---------------- describe('cancel', () => { @@ -188,6 +162,54 @@ describe('CheckpointProver', () => { }); }); + // ---------------- cancellation short-circuits execution ---------------- + + describe('cancellation short-circuits execution', () => { + it('threads its abort signal into public execution so a cancel stops the current block', async () => { + // Drive execution into the block loop: gather resolves, the sub-tree and forks are stubbed, + // and public processing parks until its signal aborts. The captured signal must be the + // prover's own abort signal, so cancelling the prover interrupts the in-flight block rather + // than letting it run to completion before the next `signal.aborted` check. + txProvider.getTxsForBlock.mockReset(); + txProvider.getTxsForBlock.mockResolvedValue({ txs: [], missingTxs: [] }); + + const subTree = { + getSubTreeResult: () => new Promise(() => {}), + startNewBlock: () => Promise.resolve(), + startChonkVerifierCircuits: () => Promise.resolve(), + addTxs: () => Promise.resolve(), + setBlockCompleted: () => Promise.resolve(), + cancel: () => {}, + stop: () => Promise.resolve(), + }; + proverFactory.createCheckpointSubTreeOrchestrator.mockResolvedValue(subTree as any); + dbProvider.fork.mockResolvedValue({ + appendLeaves: () => Promise.resolve(), + close: () => Promise.resolve(), + } as any); + + const processReached = promiseWithResolvers(); + const publicProcessor = { + process: (_txs: unknown, limits: { signal?: AbortSignal }) => { + processReached.resolve(limits.signal!); + // Park until the signal aborts, mirroring PublicProcessor's per-tx abort check. + return new Promise(resolve => { + limits.signal?.addEventListener('abort', () => resolve([[], [], [], [], []])); + }); + }, + }; + publicProcessorFactory.create.mockReturnValue(publicProcessor as any); + + const prover = makeProver(); + const signal = await processReached.promise; + expect(signal.aborted).toBe(false); + + prover.cancel(); + expect(signal.aborted).toBe(true); + await expect(prover.whenDone()).resolves.toBeUndefined(); + }); + }); + // ---------------- gather failure ---------------- describe('gather failures', () => { diff --git a/yarn-project/prover-node/src/job/checkpoint-prover.ts b/yarn-project/prover-node/src/job/checkpoint-prover.ts index c1cb4628c8ae..2f39dd15d8df 100644 --- a/yarn-project/prover-node/src/job/checkpoint-prover.ts +++ b/yarn-project/prover-node/src/job/checkpoint-prover.ts @@ -59,16 +59,16 @@ export type CheckpointProverArgs = { * The store creates a CheckpointProver once per content-key. Keying on the checkpoint's * own archive root (its post-state) means two checkpoints are "the same" iff they * produce the same archive — so a reorg branch, or a replacement built on the same - * predecessor but with different content, keys to a distinct prover; an identical - * re-add keys to the same one and reuses its in-flight sub-tree work. + * predecessor but with different content, keys to a distinct prover. * * The prover eagerly starts its own tx gather and sub-tree work in the constructor, so * callers only need to call `whenBlockProofsReady()` to obtain the resulting block-rollup * proofs. * - * The prover survives prune/re-add cycles via `markPruned()` / `markCanonical()` — - * sub-tree proving keeps running underneath, so a checkpoint that is re-added after - * a brief reorg can be re-consumed with no re-proving. + * A CheckpointProver does not survive a prune: its sub-tree work forks world-state per + * block, and an L1 prune of a base block faults those reads. The store therefore cancels and + * discards a prover when its checkpoint is pruned, and a re-add (even of identical content) + * constructs a fresh prover. * * `cancel()` is idempotent. It aborts the gather + sub-tree, rejects the block-proof * promise, and exposes a `whenDone()` that resolves once teardown has unwound. @@ -92,8 +92,6 @@ export class CheckpointProver { private cancelled = false; private subTree?: CheckpointSubTreeOrchestrator; private completed = false; - /** Pruned in the canonical chain but not yet reaped — sub-tree continues running. */ - private pruned = false; private readonly abortController = new AbortController(); /** Tracks the eager gather+execute task so `cancel()` and `whenDone()` can await its unwind. */ @@ -146,37 +144,6 @@ export class CheckpointProver { return this.completed; } - public isPruned(): boolean { - return this.pruned; - } - - /** - * Mark this prover as no longer present in the canonical chain. Sub-tree proving keeps - * running so the work survives if the checkpoint is re-added. Idempotent. - */ - public markPruned(): void { - if (this.pruned) { - return; - } - this.pruned = true; - this.deps.log.info(`Marking CheckpointProver ${this.id} as pruned`, { - checkpointNumber: this.checkpoint.number, - slotNumber: this.slotNumber, - }); - } - - /** Mark this prover as part of the canonical chain again after a re-add. Idempotent. */ - public markCanonical(): void { - if (!this.pruned) { - return; - } - this.pruned = false; - this.deps.log.info(`Marking CheckpointProver ${this.id} as canonical`, { - checkpointNumber: this.checkpoint.number, - slotNumber: this.slotNumber, - }); - } - /** AbortSignal that fires on cancel — for callers that want to wire their own tasks. */ public getAbortSignal(): AbortSignal { return this.abortController.signal; @@ -432,7 +399,14 @@ export class CheckpointProver { } private async processTxs(publicProcessor: PublicProcessor, txs: Tx[]): Promise { - const [processedTxs, failedTxs] = await publicProcessor.process(txs, { deadline: this.deps.deadline }); + // Pass the abort signal so a prune-driven cancel stops the current block's public execution + // immediately, rather than running it to completion before the next `signal.aborted` check. + // On abort `process` returns a partial result, the length check below throws, and + // `gatherAndExecute` swallows it via its `cancelled` guard. + const [processedTxs, failedTxs] = await publicProcessor.process(txs, { + deadline: this.deps.deadline, + signal: this.abortController.signal, + }); if (failedTxs.length) { const failedTxHashes = await Promise.all(failedTxs.map(({ tx }) => tx.getTxHash())); diff --git a/yarn-project/prover-node/src/job/epoch-session.test.ts b/yarn-project/prover-node/src/job/epoch-session.test.ts index 9a78535ddbfb..dea4db077bca 100644 --- a/yarn-project/prover-node/src/job/epoch-session.test.ts +++ b/yarn-project/prover-node/src/job/epoch-session.test.ts @@ -442,9 +442,6 @@ function makeStubProver(checkpoint: Checkpoint, opts: { blockProofsError?: Error whenBlockProofsReady: () => blockProofs, isCancelled: () => false, isCompleted: () => false, - isPruned: () => false, - markPruned: () => {}, - markCanonical: () => {}, cancel: () => {}, whenDone: () => Promise.resolve(), getAbortSignal: () => new AbortController().signal, diff --git a/yarn-project/prover-node/src/metrics.ts b/yarn-project/prover-node/src/metrics.ts index 699c75b807bb..78311379b590 100644 --- a/yarn-project/prover-node/src/metrics.ts +++ b/yarn-project/prover-node/src/metrics.ts @@ -96,7 +96,7 @@ export class ProverNodeJobMetrics { this.activeCheckpoints = this.meter.createObservableGauge(Metrics.PROVER_NODE_ACTIVE_CHECKPOINTS); this.activeEpochSessions = this.meter.createObservableGauge(Metrics.PROVER_NODE_ACTIVE_EPOCH_SESSIONS); this.stateObserver = (observer: BatchObservableResult) => { - observer.observe(this.activeCheckpoints!, checkpointStore.listCanonical().length); + observer.observe(this.activeCheckpoints!, checkpointStore.listAll().length); let full = 0; let partial = 0; for (const session of sessionManager.allSessions()) { diff --git a/yarn-project/prover-node/src/prover-node.test.ts b/yarn-project/prover-node/src/prover-node.test.ts index cb282dc6d376..b8806a612cdb 100644 --- a/yarn-project/prover-node/src/prover-node.test.ts +++ b/yarn-project/prover-node/src/prover-node.test.ts @@ -124,7 +124,7 @@ describe('ProverNode', () => { expect(proverNode.getLastProcessedCheckpoint()).toEqual(CheckpointNumber(100)); }); - it('dispatches chain-pruned through markPrunedAboveBlock and notifies the session manager only when affected', async () => { + it('dispatches chain-pruned through cancelAndRemoveAboveBlock and notifies the session manager only when affected', async () => { // No registered checkpoints — nothing to prune. await proverNode.handleBlockStreamEvent({ type: 'chain-pruned', @@ -135,7 +135,7 @@ describe('ProverNode', () => { expect(sessionManager.onPrune).not.toHaveBeenCalled(); // Register a checkpoint (cp 2 at block 2), then prune to block 1. The checkpoint's only block (2) is above the - // prune target, so it is marked pruned and its epoch (2) is reported. + // prune target, so it is cancelled and removed and its epoch (2) is reported. setupNotFullyProven(); await proverNode.handleBlockStreamEvent(mineCheckpoint(makeCheckpoint(2, 2, 2))); // The prune target (block 1) resolves to checkpoint 1, clamping the cursor to checkpoint 0. @@ -178,8 +178,9 @@ describe('ProverNode', () => { proven: makeTipId(2), }); - // The orphaned prover for checkpoint 3 is marked pruned, and the cursor was clamped below 3. - expect(originalProver.isPruned()).toBe(true); + // The orphaned prover for checkpoint 3 is cancelled and removed from the store, and the cursor was clamped below 3. + expect(originalProver.isCancelled()).toBe(true); + expect(proverNode.getCheckpointStore().getByCheckpoint(original)).toBeUndefined(); expect(proverNode.getLastProcessedCheckpoint()).toEqual(CheckpointNumber(1)); // The rebuilt checkpoint 3 (distinct archive root) is now served by the source. A fresh chain-checkpointed(3) @@ -193,7 +194,7 @@ describe('ProverNode', () => { }); it('throws on a prune whose target block data is missing, leaving provers and cursor untouched for retry', async () => { - // The cursor floor is resolved before any prover is marked, so a missing-data prune throws without side effects + // The cursor floor is resolved before any prover is removed, so a missing-data prune throws without side effects // and the next pass retries the whole handler (the tips cursor only advances on success). setupNotFullyProven(); await proverNode.handleBlockStreamEvent(mineCheckpoint(makeCheckpoint(3, 3, 3))); @@ -210,7 +211,8 @@ describe('ProverNode', () => { }), ).rejects.toThrow(/No block data found for prune target/); - expect(registeredProver.isPruned()).toBe(false); + expect(registeredProver.isCancelled()).toBe(false); + expect(proverNode.getCheckpointStore().listAll()).toContain(registeredProver); expect(sessionManager.onPrune).not.toHaveBeenCalled(); expect(proverNode.getLastProcessedCheckpoint()).toEqual(CheckpointNumber(3)); }); diff --git a/yarn-project/prover-node/src/prover-node.ts b/yarn-project/prover-node/src/prover-node.ts index 9e7c72e088b7..8911ad158515 100644 --- a/yarn-project/prover-node/src/prover-node.ts +++ b/yarn-project/prover-node/src/prover-node.ts @@ -169,7 +169,6 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra txGatheringTimeoutMs: this.config.txGatheringTimeoutMs, deadline: undefined, }, - { slotWatcherPollIntervalMs: this.config.proverNodePollingIntervalMs }, this.log.getBindings(), ); } @@ -372,8 +371,8 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra private async handlePruneEvent(prunedToBlock: L2BlockId) { this.log.warn(`Chain pruned to block ${prunedToBlock.number}`, { prunedToBlock }); - // Resolve the cursor floor BEFORE marking provers: markPrunedAboveBlock returns only newly-marked provers, so a - // throw after marking would leave a retry pass with nothing to act on. Resolving first means a throw leaves + // Resolve the cursor floor BEFORE removing provers: cancelAndRemoveAboveBlock returns only the provers it removed, + // so a throw after removing would leave a retry pass with nothing to act on. Resolving first means a throw leaves // everything untouched and the next pass retries the whole handler (the tips cursor only advances on success). let cursorFloor: CheckpointNumber; if (prunedToBlock.number === 0) { @@ -391,7 +390,7 @@ export class ProverNode implements L2BlockStreamEventHandler, ProverNodeApi, Tra cursorFloor = CheckpointNumber(Math.max(0, Number(targetData.checkpointNumber) - 1)); } - const affected = this.checkpointStore.markPrunedAboveBlock(prunedToBlock.number); + const affected = this.checkpointStore.cancelAndRemoveAboveBlock(prunedToBlock.number); if (this.lastProcessedCheckpoint > cursorFloor) { this.lastProcessedCheckpoint = cursorFloor; diff --git a/yarn-project/prover-node/src/session-manager.test.ts b/yarn-project/prover-node/src/session-manager.test.ts index c907cf8b3151..606ebac446a0 100644 --- a/yarn-project/prover-node/src/session-manager.test.ts +++ b/yarn-project/prover-node/src/session-manager.test.ts @@ -31,6 +31,8 @@ describe('SessionManager', () => { /** Mirror of fullSessions/partialSessions whose entries are stubs we control. */ let stubs: StubSession[]; + /** Sessions the manager passed to the failure-upload callback. */ + let sessionFailures: EpochSession[]; /** Resolves whenever the manager constructs a stub session. */ let onConstruct: ((stub: StubSession) => void) | undefined; @@ -51,10 +53,11 @@ describe('SessionManager', () => { l2BlockSource.getL1Constants.mockResolvedValue(l1Constants); l2BlockSource.isEpochComplete.mockResolvedValue(false); l2BlockSource.getCheckpoints.mockResolvedValue([]); - store.listCanonicalInSlotRange.mockReturnValue([]); - store.listCanonicalForEpoch.mockResolvedValue([]); + store.listInSlotRange.mockReturnValue([]); + store.listForEpoch.mockResolvedValue([]); stubs = []; + sessionFailures = []; onConstruct = undefined; manager = new TestSessionManager( @@ -67,6 +70,10 @@ describe('SessionManager', () => { metrics, dateProvider: new DateProvider(), config: { maxPendingJobs: 0, tickIntervalMs: 60_000, finalizationDelayMs: undefined }, + onSessionFailed: session => { + sessionFailures.push(session); + return Promise.resolve(); + }, }, (spec, provers) => { const stub = makeStubSession(spec, provers); @@ -114,7 +121,7 @@ describe('SessionManager', () => { l2BlockSource.isEpochComplete.mockResolvedValue(true); l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6), archiverCp(2, 7)]); // Store only has checkpoint 1. - store.listCanonicalInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]); + store.listInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]); await manager.onCheckpointAdded(epoch); expect(stubs.length).toBe(0); expect(manager.getFullSession(epoch)).toBeUndefined(); @@ -126,7 +133,7 @@ describe('SessionManager', () => { const provers = [proverForCheckpoint(1, 6), proverForCheckpoint(2, 7)]; l2BlockSource.isEpochComplete.mockResolvedValue(true); l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6), archiverCp(2, 7)]); - store.listCanonicalInSlotRange.mockReturnValue(provers); + store.listInSlotRange.mockReturnValue(provers); await manager.onCheckpointAdded(epoch); @@ -176,7 +183,7 @@ describe('SessionManager', () => { ), ), ); - store.listCanonicalInSlotRange.mockImplementation((fromSlot: SlotNumber) => { + store.listInSlotRange.mockImplementation((fromSlot: SlotNumber) => { if (Number(fromSlot) === 6) { return epoch3Provers; } @@ -203,7 +210,7 @@ describe('SessionManager', () => { const provers = [proverForCheckpoint(1, 6)]; l2BlockSource.isEpochComplete.mockResolvedValue(true); l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]); - store.listCanonicalInSlotRange.mockReturnValue(provers); + store.listInSlotRange.mockReturnValue(provers); await manager.onTick(); expect(manager.getFullSession(EpochNumber(3))).toBeDefined(); @@ -229,7 +236,7 @@ describe('SessionManager', () => { const provers = [proverForCheckpoint(1, 6)]; l2BlockSource.isEpochComplete.mockResolvedValue(true); l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]); - store.listCanonicalInSlotRange.mockReturnValue(provers); + store.listInSlotRange.mockReturnValue(provers); await manager.onTick(); expect(stubs.length).toBe(1); @@ -246,7 +253,7 @@ describe('SessionManager', () => { const provers = [proverForCheckpoint(1, 6)]; l2BlockSource.isEpochComplete.mockResolvedValue(true); l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]); - store.listCanonicalInSlotRange.mockReturnValue(provers); + store.listInSlotRange.mockReturnValue(provers); await manager.onTick(); expect(stubs.length).toBe(1); @@ -260,6 +267,37 @@ describe('SessionManager', () => { expect(stubs.length).toBe(1); }); + it('onTick does not re-attempt a failed epoch, but a checkpoint re-add recovers it', async () => { + // Pins the invariant: a genuinely-failed epoch is never re-attempted by the periodic tick + // (gated by lastTickEpoch), yet the SAME epoch is recovered when a chain event fires, because + // checkpoint/prune triggers reach openFullSessionIfReady ungated. This is how a failed proof is + // not resubmitted on a loop while a pruned-then-re-added epoch still gets reproven. + mockNextUnprovenSlot(2, 6); // proven tip block 2 → first unproven slot 6 → epoch 3 + const provers = [proverForCheckpoint(1, 6)]; + l2BlockSource.isEpochComplete.mockResolvedValue(true); + l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]); + store.listInSlotRange.mockReturnValue(provers); + + // Tick opens the session; it fails. lastTickEpoch is now pinned at epoch 3. + await manager.onTick(); + expect(stubs.length).toBe(1); + stubs[0].terminate('failed'); + await flushSessionCompletion(); + + // Further ticks must NOT re-attempt epoch 3 — the gate holds, no new session is constructed. + await manager.onTick(); + await manager.onTick(); + expect(stubs.length).toBe(1); + expect(manager.getFullSession(EpochNumber(3))).toBeUndefined(); + + // A checkpoint (re-)added to the epoch reaches openFullSessionIfReady ungated → fresh session. + await manager.onCheckpointAdded(EpochNumber(3)); + const recreated = manager.getFullSession(EpochNumber(3)) as unknown as StubSession | undefined; + expect(recreated).toBeDefined(); + expect(recreated!.isTerminal()).toBe(false); + expect(stubs.length).toBe(2); + }); + it('onTick keeps retrying the same epoch while a transient blocker prevents opening', async () => { // The archiver is still indexing — getCheckpoints returns a checkpoint we don't yet // have in the store. openFullSessionIfReady should bail without creating a session, @@ -267,7 +305,7 @@ describe('SessionManager', () => { mockNextUnprovenSlot(2, 6); l2BlockSource.isEpochComplete.mockResolvedValue(true); l2BlockSource.getCheckpoints.mockResolvedValue([archiverCp(1, 6)]); - store.listCanonicalInSlotRange.mockReturnValue([]); // store hasn't indexed it yet + store.listInSlotRange.mockReturnValue([]); // store hasn't indexed it yet await manager.onTick(); expect(stubs.length).toBe(0); // no session created @@ -275,7 +313,7 @@ describe('SessionManager', () => { expect(stubs.length).toBe(0); // still no session — the tick keeps trying // Archiver catches up; the next tick succeeds. - store.listCanonicalInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]); + store.listInSlotRange.mockReturnValue([proverForCheckpoint(1, 6)]); await manager.onTick(); expect(stubs.length).toBe(1); expect(manager.getFullSession(EpochNumber(3))).toBe(stubs[0] as unknown as EpochSession); @@ -291,7 +329,7 @@ describe('SessionManager', () => { const original = stubs[0]; // Now the store reports only the first prover. - store.listCanonicalInSlotRange.mockReturnValue([initial[0]]); + store.listInSlotRange.mockReturnValue([initial[0]]); await manager.onPrune([epoch]); expect(original.cancelled).toBe(true); @@ -313,7 +351,7 @@ describe('SessionManager', () => { await openCanonicalFullSession(epoch, [proverForCheckpoint(1, 6)]); const original = stubs[0]; - store.listCanonicalInSlotRange.mockReturnValue([]); + store.listInSlotRange.mockReturnValue([]); await manager.onPrune([epoch]); expect(original.cancelled).toBe(true); @@ -333,7 +371,7 @@ describe('SessionManager', () => { const original = stubs[0]; // Reorg removes every checkpoint of the epoch → session dropped, not recreated. - store.listCanonicalInSlotRange.mockReturnValue([]); + store.listInSlotRange.mockReturnValue([]); await manager.onPrune([epoch]); expect(original.cancelled).toBe(true); expect(original.state).toBe('cancelled'); @@ -351,6 +389,56 @@ describe('SessionManager', () => { expect(stubs.length).toBe(2); }); + it('reopens an epoch whose session failed once its checkpoints are re-added', async () => { + // A prune can fault a checkpoint mid-proof before it is reconciled, driving the session to + // `failed`. That terminal state must not strand the epoch — a re-add of its checkpoints reopens a + // fresh, live session (via the checkpoint trigger, ungated by the tick high-water mark). + const epoch = EpochNumber(3); + const prover = proverForCheckpoint(1, 6); + await openCanonicalFullSession(epoch, [prover]); + const original = stubs[0]; + + original.terminate('failed'); + await flushSessionCompletion(); + expect(original.state).toBe('failed'); + + await openCanonicalFullSession(epoch, [prover]); + const recreated = manager.getFullSession(epoch) as unknown as StubSession | undefined; + expect(recreated).toBeDefined(); + expect(recreated).not.toBe(original); + expect(recreated!.uuid).not.toBe(original.uuid); + expect(recreated!.isTerminal()).toBe(false); + expect(stubs.length).toBe(2); + }); + + it('uploads a post-mortem for a genuine proving failure (canonical content unchanged)', async () => { + const epoch = EpochNumber(3); + const prover = proverForCheckpoint(1, 6); + await openCanonicalFullSession(epoch, [prover]); + const session = manager.getFullSession(epoch); + // The session's checkpoints still match canonical → the failure was genuine, not a prune. + store.listInSlotRange.mockReturnValue([prover]); + + stubs[0].terminate('failed'); + await flushSessionCompletion(); + + expect(sessionFailures).toEqual([session]); + }); + + it('skips the failure upload when the failure coincides with a canonical content change', async () => { + const epoch = EpochNumber(3); + const prover = proverForCheckpoint(1, 6); + await openCanonicalFullSession(epoch, [prover]); + // A prune removed the checkpoint from the store, so the session's content no longer matches + // canonical — the failure was caused by the content changing under it, not a proving fault. + store.listInSlotRange.mockReturnValue([]); + + stubs[0].terminate('failed'); + await flushSessionCompletion(); + + expect(sessionFailures).toEqual([]); + }); + it('drops terminal sessions on the next reconcile', async () => { const epoch = EpochNumber(3); await openCanonicalFullSession(epoch, [proverForCheckpoint(1, 6)]); @@ -372,8 +460,8 @@ describe('SessionManager', () => { it('cancels and recreates a partial session whose canonical content changed', async () => { const epoch = EpochNumber(7); const initial = [proverForCheckpoint(1, 14)]; - store.listCanonicalForEpoch.mockResolvedValue(initial); - store.listCanonicalInSlotRange.mockReturnValue(initial); + store.listForEpoch.mockResolvedValue(initial); + store.listInSlotRange.mockReturnValue(initial); const stubPromise = awaitNextStub(); const startPromise = manager.startProof(epoch); @@ -383,7 +471,7 @@ describe('SessionManager', () => { // The store now reports a different prover at the same slot. const swapped = [proverForCheckpoint(2, 14)]; - store.listCanonicalInSlotRange.mockReturnValue(swapped); + store.listInSlotRange.mockReturnValue(swapped); const recreatePromise = awaitNextStub(); await manager.onTick(); @@ -406,14 +494,14 @@ describe('SessionManager', () => { it('drops a partial session and does not recreate when canonical content goes empty', async () => { const epoch = EpochNumber(7); const initial = [proverForCheckpoint(1, 14)]; - store.listCanonicalForEpoch.mockResolvedValue(initial); - store.listCanonicalInSlotRange.mockReturnValue(initial); + store.listForEpoch.mockResolvedValue(initial); + store.listInSlotRange.mockReturnValue(initial); const stubPromise = awaitNextStub(); const startPromise = manager.startProof(epoch); const original = await stubPromise; - store.listCanonicalInSlotRange.mockReturnValue([]); + store.listInSlotRange.mockReturnValue([]); await manager.onTick(); expect(original.cancelled).toBe(true); @@ -427,8 +515,8 @@ describe('SessionManager', () => { it('drops terminal partial sessions on the next reconcile', async () => { const epoch = EpochNumber(7); const canonical = [proverForCheckpoint(1, 14)]; - store.listCanonicalForEpoch.mockResolvedValue(canonical); - store.listCanonicalInSlotRange.mockReturnValue(canonical); + store.listForEpoch.mockResolvedValue(canonical); + store.listInSlotRange.mockReturnValue(canonical); const stubPromise = awaitNextStub(); const startPromise = manager.startProof(epoch); @@ -457,8 +545,8 @@ describe('SessionManager', () => { terminalFull.terminate('failed'); expect(terminalFull.isTerminal()).toBe(true); - store.listCanonicalForEpoch.mockResolvedValue(canonical); - store.listCanonicalInSlotRange.mockReturnValue(canonical); + store.listForEpoch.mockResolvedValue(canonical); + store.listInSlotRange.mockReturnValue(canonical); const stubPromise = awaitNextStub(); const startPromise = manager.startProof(epoch); @@ -476,8 +564,8 @@ describe('SessionManager', () => { it('startProof ignores a terminal partial session and constructs a fresh one', async () => { const epoch = EpochNumber(7); const canonical = [proverForCheckpoint(1, 14)]; - store.listCanonicalForEpoch.mockResolvedValue(canonical); - store.listCanonicalInSlotRange.mockReturnValue(canonical); + store.listForEpoch.mockResolvedValue(canonical); + store.listInSlotRange.mockReturnValue(canonical); // Open a partial, settle it terminally, then call startProof again. const firstPromise = awaitNextStub(); @@ -508,8 +596,8 @@ describe('SessionManager', () => { const epoch = EpochNumber(7); // Epoch 7 covers slots [14, 15]. Single canonical prover at slot 14. const canonical = [proverForCheckpoint(1, 14)]; - store.listCanonicalForEpoch.mockResolvedValue(canonical); - store.listCanonicalInSlotRange.mockReturnValue(canonical); + store.listForEpoch.mockResolvedValue(canonical); + store.listInSlotRange.mockReturnValue(canonical); // Arm the construction trigger before calling startProof — no need to sleep waiting // for reconcile to land. @@ -535,7 +623,7 @@ describe('SessionManager', () => { }); it('startProof throws when the epoch has no canonical content', async () => { - store.listCanonicalForEpoch.mockResolvedValue([]); + store.listForEpoch.mockResolvedValue([]); await expect(manager.startProof(EpochNumber(7))).rejects.toThrow(/No blocks found/); }); @@ -543,7 +631,7 @@ describe('SessionManager', () => { const epoch = EpochNumber(7); // proverForCheckpoint builds a checkpoint whose single block number equals the checkpoint // number (1 here). A proven tip at or beyond that block means the epoch is already proven. - store.listCanonicalForEpoch.mockResolvedValue([proverForCheckpoint(1, 14)]); + store.listForEpoch.mockResolvedValue([proverForCheckpoint(1, 14)]); l2BlockSource.getBlockNumber.mockResolvedValue(BlockNumber(1)); await expect(manager.startProof(epoch)).rejects.toThrow(/already proven/i); @@ -559,7 +647,7 @@ describe('SessionManager', () => { expect(stubs.length).toBe(1); const fullSession = stubs[0]; - store.listCanonicalForEpoch.mockResolvedValue(provers); + store.listForEpoch.mockResolvedValue(provers); const doneId = await manager.startProof(epoch); fullSession.terminate('completed'); @@ -571,8 +659,8 @@ describe('SessionManager', () => { it('startProof dedupes against an existing partial session with the same spec', async () => { const epoch = EpochNumber(7); const canonical = [proverForCheckpoint(1, 14)]; - store.listCanonicalForEpoch.mockResolvedValue(canonical); - store.listCanonicalInSlotRange.mockReturnValue(canonical); + store.listForEpoch.mockResolvedValue(canonical); + store.listInSlotRange.mockReturnValue(canonical); const firstId = await manager.startProof(epoch); expect(stubs).toHaveLength(1); @@ -605,7 +693,7 @@ describe('SessionManager', () => { ), ), ); - store.listCanonicalInSlotRange.mockImplementation((fromSlot: SlotNumber) => { + store.listInSlotRange.mockImplementation((fromSlot: SlotNumber) => { if (Number(fromSlot) === 6) { return epoch3Provers; } @@ -668,7 +756,7 @@ describe('SessionManager', () => { async function openCanonicalFullSession(epoch: EpochNumber, provers: CheckpointProver[]): Promise { l2BlockSource.isEpochComplete.mockResolvedValueOnce(true); l2BlockSource.getCheckpoints.mockResolvedValueOnce(provers.map(p => ({ checkpoint: p.checkpoint }) as any)); - store.listCanonicalInSlotRange.mockReturnValueOnce(provers); + store.listInSlotRange.mockReturnValueOnce(provers); await manager.onCheckpointAdded(epoch); } @@ -678,6 +766,11 @@ describe('SessionManager', () => { * after an action that schedules a reconcile — the manager itself signals "session ready" * via the factory call. */ + /** Lets `runSession`'s post-`start()` continuation (failure upload, logging) run after a stub terminates. */ + function flushSessionCompletion(): Promise { + return new Promise(resolve => setImmediate(resolve)); + } + function awaitNextStub(): Promise { const { promise, resolve } = promiseWithResolvers(); onConstruct = stub => { @@ -831,7 +924,6 @@ function proverForCheckpoint(number: number, slot: number): CheckpointProver { id: CheckpointProver.idFor(checkpoint), checkpoint, slotNumber: SlotNumber(slot), - isPruned: () => false, isCancelled: () => false, } as unknown as CheckpointProver; } diff --git a/yarn-project/prover-node/src/session-manager.ts b/yarn-project/prover-node/src/session-manager.ts index 3430815c205f..1162958f53c2 100644 --- a/yarn-project/prover-node/src/session-manager.ts +++ b/yarn-project/prover-node/src/session-manager.ts @@ -185,7 +185,7 @@ export class SessionManager { * slot. Dedupes against any existing session covering the same range, returning its id. */ public async startProof(epoch: EpochNumber): Promise { - const canonical = await this.deps.checkpointStore.listCanonicalForEpoch(epoch); + const canonical = await this.deps.checkpointStore.listForEpoch(epoch); if (canonical.length === 0) { throw new EmptyEpochError(epoch); } @@ -268,7 +268,7 @@ export class SessionManager { this.fullSessions.delete(key); continue; } - const canonical = this.canonicalCheckpointsForSpec(session.getSpec()); + const canonical = this.checkpointsForSpec(session.getSpec()); if (!this.checkpointsMatch(session.getCheckpoints(), canonical)) { this.fireAndForgetCancel(session, 'canonical content changed'); this.fullSessions.delete(key); @@ -284,7 +284,7 @@ export class SessionManager { this.partialSessions.delete(key); continue; } - const canonical = this.canonicalCheckpointsForSpec(session.getSpec()); + const canonical = this.checkpointsForSpec(session.getSpec()); if (!this.checkpointsMatch(session.getCheckpoints(), canonical)) { this.fireAndForgetCancel(session, 'canonical content changed'); this.partialSessions.delete(key); @@ -298,6 +298,8 @@ export class SessionManager { } private async openFullSessionIfReady(epoch: EpochNumber): Promise { + // `recreateInvalidSessions` runs at the top of every reconcile and deletes terminal sessions + // before this is called, so a session present here is live and already covers the epoch. if (this.fullSessions.has(epoch)) { return; } @@ -314,7 +316,7 @@ export class SessionManager { return; } const [fromSlot, toSlot] = getSlotRangeForEpoch(epoch, l1Constants); - const canonical = this.deps.checkpointStore.listCanonicalInSlotRange(fromSlot, toSlot); + const canonical = this.deps.checkpointStore.listInSlotRange(fromSlot, toSlot); if (!this.archiverFullyCovered(archiverCps, canonical)) { this.log.debug(`Skipping full-session open for epoch ${epoch}: archiver checkpoints not all in store`, { archiverCount: archiverCps.length, @@ -329,7 +331,7 @@ export class SessionManager { } private openPartialSession(spec: SessionSpec): void { - const canonical = this.deps.checkpointStore.listCanonicalInSlotRange(spec.fromSlot, spec.toSlot); + const canonical = this.deps.checkpointStore.listInSlotRange(spec.fromSlot, spec.toSlot); if (canonical.length === 0) { return; } @@ -403,6 +405,17 @@ export class SessionManager { const state = await session.start(); this.log.info(`Session ${session.getId()} exited with state ${state}`); if (state === 'failed' && this.deps.onSessionFailed) { + // Best-effort suppression of the spurious post-mortem upload a prune produces: if the session's + // checkpoints no longer match the store's current set, the failure was caused by the content + // changing under it, not a genuine proving fault, so skip the upload. This is inherently racy — + // the store lags the world-state unwind, so a fault observed before the prune is reconciled here + // still uploads. The epoch is recovered regardless by recreating the session on re-add. + if (!this.checkpointsMatch(session.getCheckpoints(), this.checkpointsForSpec(session.getSpec()))) { + this.log.info(`Skipping failure upload for session ${session.getId()}: canonical content changed`, { + ...session.getSpec(), + }); + return; + } try { await this.deps.onSessionFailed(session); } catch (err) { @@ -447,6 +460,24 @@ export class SessionManager { return live >= max; } + /** + * Maps a reconcile trigger to the epochs whose full session should be (re)opened. + * + * This is where the "don't retry a genuinely-failed epoch, but do recover a pruned one" invariant + * lives — enforced by which triggers are gated by `lastTickEpoch`: + * + * - The periodic `tick` IS gated: once a tick has opened a session for an epoch, `lastTickEpoch` + * advances to it and later ticks skip it (`epoch <= lastTickEpoch`). So a failed attempt is never + * resubmitted on a loop by the tick. + * - `checkpoint` and `prune` are deliberately NOT gated. They only fire when the epoch's canonical + * content actually changes — a checkpoint arrives, or a reorg prunes/replaces one — which is + * exactly when re-attempting is correct. + * + * A genuine proving failure produces no content change, hence no checkpoint/prune event, so only + * the gated tick could reopen it — and it won't. A prune + re-add fires ungated events, so the + * epoch is reopened through this path (and `openFullSessionIfReady` rebuilds over the fresh + * provers). See the "onTick does not retry ... but recovers ... re-added" test. + */ private async epochsForTrigger(trigger: ReconcileTrigger): Promise { switch (trigger.kind) { case 'checkpoint': @@ -481,8 +512,8 @@ export class SessionManager { return getEpochAtSlot(header.getSlot(), await this.getL1Constants()); } - private canonicalCheckpointsForSpec(spec: SessionSpec): CheckpointProver[] { - return this.deps.checkpointStore.listCanonicalInSlotRange(spec.fromSlot, spec.toSlot); + private checkpointsForSpec(spec: SessionSpec): CheckpointProver[] { + return this.deps.checkpointStore.listInSlotRange(spec.fromSlot, spec.toSlot); } private fireAndForgetCancel(session: EpochSession, reason: string): void {