feat(prover-node): make epoch proving robust to prune-induced fork faults

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

@@ -4,7 +4,7 @@ import type { Logger } from '@aztec/aztec.js/log';
 import { tryRmDir } from '@aztec/foundation/fs';
 import { promiseWithResolvers } from '@aztec/foundation/promise';
 import { sleep } from '@aztec/foundation/sleep';
-import { downloadEpochProvingJob, rerunEpochProvingJob } from '@aztec/prover-node';
+import { downloadEpochProvingJob, rerunCheckpointProvingJob, rerunEpochProvingJob } from '@aztec/prover-node';
 import type { TestProverNode } from '@aztec/prover-node/test';
 
 import { jest } from '@jest/globals';
@@ -61,11 +61,11 @@ describe('single-node/proving/upload_failed_proof', () => {
     await tryRmDir(rerunDownloadDir, logger);
   });
 
-  // Makes the prover's top-tree prove always throw (v5 uses the session's topTreeProveOverride hook;
-  // pre-v5 it patched finalizeEpoch), intercepts tryUploadSessionFailure (pre-v5 tryUploadEpochFailure)
-  // to capture the upload URL, then waits for epoch 1 to start and for the upload to complete. Tears
-  // down the live context, downloads the proving job data, and re-runs it via rerunEpochProvingJob with
-  // fake proofs on a fresh config.
+  // Makes the prover's top-tree prove always throw. Because every checkpoint prover still succeeds, the
+  // session fails on its own account (state 'failed'), which the session manager treats as a genuine,
+  // race-free failure and uploads a post-mortem eagerly via tryUploadEpochFailure(sessionId, checkpoints).
+  // Intercepts that to capture the upload URL, then tears down the live context, downloads the proving
+  // job data, and re-runs it via rerunEpochProvingJob with fake proofs on a fresh config.
   it('uploads failed proving job state and re-runs it on a fresh instance', async () => {
     // Make initial prover node fail to prove, via the session's top-tree-prove hook.
     const proverNode = test.proverNodes[0].getProverNode() as TestProverNode;
@@ -77,19 +77,20 @@ describe('single-node/proving/upload_failed_proof', () => {
       },
     });
 
-    // And track when the epoch failure upload is complete
+    // Track when the epoch failure upload is complete. It fires eagerly when a full session fails with
+    // healthy provers (a top-tree failure here).
     const { promise: epochUploaded, resolve: onEpochUploaded } = promiseWithResolvers<string>();
-    const origTryUploadEpochFailure = proverNode.tryUploadSessionFailure.bind(proverNode);
-    proverNode.tryUploadSessionFailure = async (session: any) => {
-      const url = await origTryUploadEpochFailure(session);
+    const origTryUploadEpochFailure = proverNode.tryUploadEpochFailure.bind(proverNode);
+    proverNode.tryUploadEpochFailure = async (id: any, checkpoints: any) => {
+      const url = await origTryUploadEpochFailure(id, checkpoints);
       if (url !== undefined) {
         onEpochUploaded(url);
       }
       return url;
     };
 
-    // Warp to the start of epoch one so prover node starts proving epoch 0,
-    // and wait for the data to be uploaded to the remote file store
+    // Warp to the start of epoch one so the prover node starts proving, fails at the top tree, and
+    // uploads the failed proving-job data to the remote file store.
     await test.warpToEpochStart(1);
     const epochUploadUrl = await epochUploaded;
 
@@ -122,4 +123,62 @@ describe('single-node/proving/upload_failed_proof', () => {
 
     logger.info(`Test succeeded`);
   });
+
+  // Same shape as above, one level down: forces a single checkpoint's sub-tree to fail (every checkpoint
+  // prover fails, via the test hook), which uploads that checkpoint's proving data on its own. Intercepts
+  // tryUploadCheckpointFailure for the URL, then downloads and re-proves just that checkpoint via
+  // rerunCheckpointProvingJob (no epoch top-tree / L1 submit).
+  it('uploads failed checkpoint proving state and re-proves it on a fresh instance', async () => {
+    const proverNode = test.proverNodes[0].getProverNode() as TestProverNode;
+    proverNode.setCheckpointHooks({
+      checkpointProveOverride: async () => {
+        await sleep(1000);
+        logger.warn(`Triggering error on checkpoint sub-tree prove`);
+        throw new Error(`Fake error while proving checkpoint`);
+      },
+    });
+
+    // The checkpoint post-mortem upload fires eagerly when a CheckpointProver's block proofs reject.
+    const { promise: checkpointUploaded, resolve: onCheckpointUploaded } = promiseWithResolvers<string>();
+    const origTryUploadCheckpointFailure = proverNode.tryUploadCheckpointFailure.bind(proverNode);
+    proverNode.tryUploadCheckpointFailure = async (prover: any) => {
+      const url = await origTryUploadCheckpointFailure(prover);
+      if (url !== undefined) {
+        onCheckpointUploaded(url);
+      }
+      return url;
+    };
+
+    // Warp so the prover node starts registering checkpoints; the first one's sub-tree fails and uploads.
+    await test.warpToEpochStart(1);
+    const checkpointUploadUrl = await checkpointUploaded;
+
+    await test.teardown();
+
+    const rerunDownloadPath = join(rerunDownloadDir, 'data.bin');
+    logger.warn(`Downloading checkpoint proving job data and state`, { rerunDataDir, rerunDownloadPath });
+    await downloadEpochProvingJob(checkpointUploadUrl, logger, {
+      dataDirectory: rerunDataDir,
+      jobDataDownloadPath: rerunDownloadPath,
+    });
+
+    logger.warn(`Rerunning checkpoint proving job from ${rerunDownloadPath}`);
+    await rerunCheckpointProvingJob(
+      rerunDownloadPath,
+      logger,
+      {
+        ...config,
+        realProofs: false,
+        dataStoreMapSizeKb: 1024 * 1024,
+        dataDirectory: rerunDataDir,
+        proverAgentCount: 2,
+        proverId: EthAddress.random(),
+        ...(await getACVMConfig(logger)),
+        ...(await getBBConfig(logger)),
+      },
+      context.genesis,
+    );
+
+    logger.info(`Test succeeded`);
+  });
 });

yarn-project/prover-node/README.md

@@ -48,10 +48,12 @@ The prover-node splits responsibility between four classes:
 
 - **`ProverNode`** — owns the long-lived collections, wires the L2BlockStream, and
   translates each chain event into a single method call on the `SessionManager` or
-  `ProofPublishingService`. It also performs the per-event side effects that don't
-  belong on an `EpochSession` (registering new checkpoints with the store, sweeping
-  expired epochs out of the cache and the store, etc.) and runs the failure-upload
-  action when an `EpochSession` exits with `failed`.
+  `ProofPublishingService`. It also performs side effects that don't belong on an
+  `EpochSession`: registering new checkpoints with the store per event, and sweeping expired
+  epochs out of the cache and the store on a periodic ticker (see below). It uploads post-mortem
+  snapshots on failure at two levels: per checkpoint when a `CheckpointProver` fails (its `onFailed`
+  callback), and per session when a full session fails on its own account (`onSessionFailed` /
+  `EpochSession.hasFailed()`).
 - **`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
@@ -64,8 +66,12 @@ The prover-node splits responsibility between four classes:
   translated into a `reconcile(trigger)` call, a single idempotent function that walks
   all `EpochSession`s, cancels any whose canonical content has shifted, re-creates them with
   the new content, and opens fresh full `EpochSession`s for any epoch that has become provable.
-  Reconcile runs on a `SerialQueue` (from `@aztec/foundation/queue`), so two concurrent
-  triggers can never interleave on an `await` and race on the `EpochSession` maps.
+  It never builds a session over a **failed** `CheckpointProver` (one whose block proofs rejected —
+  a sub-tree fault or a prune-induced fork fault): a session over it could only fail, so the epoch is
+  cheaply skipped until a prune/re-add installs a fresh prover in its place, at which point a rebuilt
+  session reuses every already-completed sub-proof from the content-addressed broker. Reconcile runs on
+  a `SerialQueue` (from `@aztec/foundation/queue`), so two concurrent triggers can never interleave on an
+  `await` and race on the `EpochSession` maps.
 - **`ProofPublishingService`** — central owner of L1 proof submission. `EpochSession`s hand
   their top-tree proofs to the service as `PublishCandidate`s; the service serialises
   one publish at a time against a freshly-created `ProverNodePublisher`, gates eligibility
@@ -150,9 +156,9 @@ stateDiagram-v2
   awaiting_root --> publishing_proof: epoch proof ready, submit to L1
   publishing_proof --> completed: publish succeeds
   publishing_proof --> superseded: longer same-epoch candidate wins
-  publishing_proof --> failed: L1 submission errored
-  awaiting_checkpoints --> failed: top-tree prove errored
-  awaiting_root --> failed: top-tree prove errored
+  publishing_proof --> failed: L1 submission errored (provers healthy)
+  awaiting_checkpoints --> stopped: a checkpoint prover failed
+  awaiting_root --> failed: top-tree prove errored (provers healthy)
   initialized --> timed_out: deadline
   awaiting_checkpoints --> timed_out: deadline (EpochSession or candidate)
   awaiting_root --> timed_out: deadline (EpochSession or candidate)
@@ -162,9 +168,22 @@ stateDiagram-v2
   superseded --> [*]
   cancelled --> [*]
   timed_out --> [*]
+  stopped --> [*]
   failed --> [*]
 ```
 
+A fault settles the `EpochSession` in one of two terminal states:
+
+- **`stopped`** — a `CheckpointProver` under the session failed (its block proofs rejected). This may
+  be a prune-induced fork fault, so it is *not* a verdict on the epoch and is *not* uploaded. The
+  reconciler drops the session and, guarded by `isFailed()`, does not rebuild over the failed prover;
+  a prune/re-add installs a fresh prover and the epoch is retried.
+- **`failed`** — the session's own work (top-tree prove or L1 submit) failed while *every* checkpoint
+  prover succeeded. Because healthy provers rule out a prune, this is a genuine, race-free failure
+  (`EpochSession.hasFailed()`). The reconciler retains such a full session (so the tick does not
+  re-prove a deterministically-failing epoch) and uploads a post-mortem exactly once. No prune/fault
+  classification is needed — the two states carry it.
+
 The non-terminal states track the window between `start()` and the L1 submission:
 
 - `awaiting-checkpoints` — a `TopTreeJob` is awaiting each checkpoint's sub-tree result
@@ -323,44 +342,57 @@ sequenceDiagram
   PS->>PS: drain reads proven afresh, re-checks eligibility
 ```
 
-### Per-event expiry sweep
+### Periodic expiry sweep
 
 ```mermaid
 sequenceDiagram
-  participant L2 as L2BlockStream
+  participant T as expiryTicker (RunningPromise)
   participant PN as ProverNode
+  participant L2 as L2BlockStream
   participant CC as ChonkCache
   participant CS as CheckpointStore
 
-  L2->>PN: any event
+  T->>PN: checkEpochExpiry() (every poll interval)
   PN->>L2: getSyncedL2SlotNumber()
   PN->>PN: latestEpoch = getEpochAtSlot(latestSlot)
   PN->>PN: newlyExpiredUpTo = latestEpoch - (proofSubmissionEpochs + 1)
   loop for each newly-expired epoch
-    PN->>L2: getCheckpointsData({epoch}) + getBlocks(...)
+    PN->>L2: getBlocks({epoch, onlyCheckpointed})
     PN->>CC: releaseForBlocks(blocks)
     PN->>CS: reapExpired(epoch)
   end
 ```
 
-Expiry runs at the end of every `handleBlockStreamEvent` call (not on any specific
-event type). An epoch `E` is expired once the chain reaches the start of epoch
-`E + proofSubmissionEpochs + 1` — the deadline beyond which an L1 submission for
-`E` would be rejected. A monotonic high-water mark (`lastExpiredEpoch`) makes the
-sweep cheap: it advances per event and never revisits an epoch. It is seeded at
-`start()` from the last fully-proven epoch (computed in `computeStartupState`),
-so on a restart we never re-sweep epochs that already reached L1.
+The sweep is driven solely by the `expiryTicker` (a `RunningPromise` armed in `start()`), which
+calls `checkEpochExpiry()` every poll interval whether or not block-stream events arrive. It is
+deliberately **not** run from `handleBlockStreamEvent` — expiry is a background sweep keyed off the
+archiver's synced slot, not part of event processing — and because a `RunningPromise` never overlaps
+its own runs, no two sweeps can race. An epoch `E` is expired once the chain reaches the start of epoch
+`E + proofSubmissionEpochs + 1` — the deadline beyond which an L1 submission for `E` would be rejected.
+Expiry only releases the chonk cache and reaps `E`'s provers.
+A monotonic high-water mark (`lastExpiredEpoch`) makes the sweep cheap: it only
+advances after a sweep completes and never revisits an epoch. It is seeded at `start()` from the last
+fully-proven epoch (`resolveLastFullyProvenEpoch`), so on a restart we never re-sweep epochs that already
+reached L1.
 
 ### Periodic tick
 
 `SessionManager.start()` arms a `RunningPromise` that fires
 `reconcile({ kind: 'tick' })` every `tickIntervalMs`. The tick picks up epochs that
 became complete by time alone (no fresh checkpoint event) and advances to the
-next unproven epoch once the previous one lands on L1. A monotonic high-water
-mark (`lastTickEpoch`) prevents the tick from re-opening an epoch whose `EpochSession`
-already terminated; the mark advances only after an `EpochSession` actually exists for
-the epoch, so transient blockers (max-pending-jobs reached, archiver still
-indexing) leave the mark in place and the next tick retries.
+next unproven epoch once the previous one lands on L1. The tick is not gated by any per-epoch
+bookkeeping; two things keep it from re-proving a doomed epoch:
+
+- `openFullSessionIfReady` refuses to build a session when any `CheckpointProver` in the set has
+  **failed** (`isFailed()`). So a prover fault (possibly a prune) skips the epoch cheaply each tick —
+  no session, no proving — until a prune/re-add installs a fresh prover.
+- A session that failed on its own account (`hasFailed()` — top-tree/submit failed with every prover
+  healthy) is **retained** by `recreateInvalidSessions` rather than deleted, so the tick's
+  `fullSessions.has(epoch)` check skips it. It is replaced only when its canonical content changes (a
+  re-add), so a deterministically-failing epoch is not re-proved every tick.
+
+Transient blockers (max-pending-jobs reached, archiver still indexing) create no session and no failed
+prover, so the next tick simply tries again.
 
 ## Walkthroughs
 
@@ -467,6 +499,20 @@ by keeping the prover around.  Removing it on prune and rebuilding on re-add is
 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.
 
+### How is a genuine failure told apart from a prune, without a racy classification?
+
+Knowledge of an L1 reorg reaches the prover-node on two causally unordered channels: the control
+plane (`chain-pruned` → `onPrune`) and the data plane (world-state unwinds, and an in-flight fork
+read faults inside a `CheckpointProver` mid-proof). If the `EpochSession` reacted to a fault by asking
+"was this a prune or a genuine failure?", every answer would sample control-plane state that lags the
+data-plane unwind — racy by construction. We remove the question by pushing the fact down to where it
+is unambiguous: a `CheckpointProver` records `isFailed()` when *its* block proofs reject. The session
+then reads a clean signal: if any prover failed, the fault might be a prune (`stopped`) — don't upload,
+don't rebuild over it, recover on re-add; if *no* prover failed yet the session's own top-tree/submit
+work failed, that is definitively **not** a prune (`failed`) — a genuine, race-free failure that is
+retained (so a deterministic failure isn't re-proved every tick) and uploaded once. No control-plane
+sampling, no timing race — the two prover-level and session-level facts carry the whole decision.
+
 ## Configuration
 
 | Env var | Description |
@@ -475,7 +521,7 @@ because the proving broker is content-addressed independently of the prover's li
 | `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`. |
-| `PROVER_NODE_FAILED_EPOCH_STORE` | If set, failed `EpochSession`s upload their proving data (every `CheckpointProver`'s txs + register-time data, regardless of sub-tree completion) to this file store. |
+| `PROVER_NODE_FAILED_EPOCH_STORE` | If set, a full session that fails on its own account (top-tree/submit, provers healthy) uploads its proving data (every `CheckpointProver`'s txs + register-time data, regardless of sub-tree completion) to this file store. |
 | `PROVER_NODE_DISABLE_PROOF_PUBLISH` | If true, the publishing service runs `analyzeEpochProofSubmission` (estimates L1 fees) instead of actually submitting. |
 
 ## Failure handling and observability
@@ -493,14 +539,29 @@ Loggers:
 - `prover-node:checkpoint-prover` — sub-tree pipeline (gather, block processing).
 - `prover-client:chonk-cache` — chonk-verifier cache enqueue / release events.
 
-On `failed` exit, `SessionManager.runSession` invokes the `onSessionFailed` callback
-the manager was constructed with. `ProverNode` wires this to `tryUploadSessionFailure`,
-which calls `SessionManager.buildSessionProvingData(session)` to walk every `CheckpointProver`
-referenced by the `EpochSession` and assemble an `EpochProvingJobData` snapshot — including
-every `CheckpointProver`'s txs and register-time data even if its sub-tree never reached
-`isCompleted()`. This snapshot is what `uploadEpochProofFailure` ships to the
-configured file store along with a world-state + archiver backup, so the failure
-can be reproduced offline via `rerunEpochProvingJob`.
+Failed proving data is uploaded to the configured file store (`PROVER_NODE_FAILED_EPOCH_STORE`) at two
+levels, and re-proved offline by the matching downloader/rerunner: an epoch job via
+`downloadEpochProvingJob` + `rerunEpochProvingJob`, and a single checkpoint via the same download plus
+`rerunCheckpointProvingJob` (which rebuilds just that checkpoint's sub-tree prover and awaits its block
+proofs — no epoch top-tree or L1 submit). Both upload levels build an `EpochProvingJobData` snapshot with
+`SessionManager.buildProvingData(...)` — every `CheckpointProver`'s txs and register-time data, even if
+its sub-tree never reached `isCompleted()` — and ship it via `uploadEpochProofFailure` alongside a
+world-state + archiver backup:
+
+- **Per checkpoint.** When a `CheckpointProver`'s block proofs reject for a non-cancel reason (a sub-tree
+  fault or a prune-induced fork fault), it fires its `onFailed` callback and `ProverNode` uploads that one
+  checkpoint via `tryUploadCheckpointFailure(prover)`. This fires for prune-induced faults too — that is
+  intentional; a prune-caused checkpoint snapshot is harmless, and *not* trying to tell prune from genuine
+  failure is exactly what keeps this race-free. A cancelled prover (control-plane prune / shutdown) is not
+  a failure and does not upload.
+- **Per session.** When a full session ends in its own genuine failure (`EpochSession.hasFailed()` —
+  top-tree/submit failed with every prover healthy), the session manager fires `onSessionFailed` and
+  `ProverNode` uploads all the session's checkpoints via `tryUploadEpochFailure(...)`. This is race-free
+  (healthy provers rule out a prune) and fires once, since the failed session is retained and never re-run.
+
+A `stopped` session (a prover under it failed) does not upload at the session level — its failed checkpoint
+already uploaded itself. Each backup is a full world-state + archiver snapshot, so a prune that faults
+several in-flight provers produces several uploads; this is accepted for the diagnostic value.
 
 Metrics emitted by `EpochSession`s: