feat(prover-node): checkpoint-driven optimistic proving

Drives the prover-node onto the split `CheckpointSubTreeOrchestrator` +
`TopTreeOrchestrator` pair, with checkpoint-driven proving that pipelines
sub-trees against tx-gathering and the top-tree against the in-flight
sub-trees.

## What's new

### `prover-node` — `EpochProvingJob` job-model rewrite

`EpochProvingJob` becomes an orchestrator over a `Map<string, CheckpointJob>`
keyed by `${number}:${slot}`. Each `CheckpointJob` owns a single
`CheckpointSubTreeOrchestrator` with its own per-checkpoint context (txs,
attestations, previous-block header, l1ToL2 messages, archive sibling path).
A `TopTreeJob` drives the epoch root rollup once all checkpoint sub-trees
have started block-level proving.

Public API:
- `registerCheckpoint` — synchronous; sets up sub-tree, kicks off chonk-verifier
  cache fill, attaches the `blockProofs` promise to the eventual top-tree job.
- `provideTxs` — supplies simulated txs, transitions the checkpoint job from
  registered → block-proving.
- `removeCheckpoint(synchronous, idempotent)` — drops a single checkpoint by
  `(number, slot)`, fire-and-forget cancels its sub-tree. Tolerates re-add of
  the same checkpoint number under a different slot.
- `removeCheckpointsAfter`, `getCheckpointCount`, `getCheckpointNumbers`,
  `cancelPendingCheckpoints`.

### `prover-node` — `L2BlockStream`-driven checkpoint pipeline

The prover-node consumes `chain-checkpointed` / `chain-pruned` events from an
`L2BlockStream` rooted at the first block of the first unproven epoch. On
each `chain-checkpointed`:
1. Resolve the epoch via `getEpochAtSlot`.
2. Get-or-create the per-epoch `EpochProvingJob`.
3. Detached-task gather txs + register the checkpoint with the job.

On `chain-pruned`: call `removeCheckpointsAfter(threshold)` on every job
whose first checkpoint sits at or above the threshold. Pending gather tasks
are cancelled via `AbortSignal`.

Finalization is driven by the union of three signals: epoch-monitor sees
the epoch close on L1, a checkpoint for a strictly later epoch arrives, or
all expected checkpoints (per archiver) are registered while the epoch is
already complete on L1.

### `prover-node` — reorg-after-finalization restart

When the L2BlockStream emits a prune that retroactively invalidates an
epoch already in finalize, the prover-node aborts the in-flight publish,
clears the job, and restarts proving from the new tip.

### e2e

- New `epochs_optimistic_proving.parallel.test.ts`: full e2e covering the
  pipelining, replacement-checkpoint reuse, and reorg-during-proving paths.
- `epochs_proof_fails`, `epochs_upload_failed_proof`, `epochs_long_proving_time`,
  `epochs_multi_proof` updated to assert the new in-flight epoch behaviour.

## What's removed

`EpochProver` interface and `ServerEpochProver` are removed: the prover-node
no longer drives a single-class epoch prover, so the legacy API has no
production callers. `ProvingOrchestrator` survives only as a base class for
`CheckpointSubTreeOrchestrator` and as the single-class driver used by
`prover-client`'s integration tests; it no longer implements `EpochProver`.

## Test plan

- `yarn workspace @aztec/prover-client test` — 261 tests pass.
- `yarn workspace @aztec/prover-node test` — 89 tests pass.
- e2e tests covering optimistic proving, reorgs during proving, failed
  proof publish, and multi-checkpoint flows are included in this PR.

Author: PhilWindle

yarn-project/aztec-node/src/aztec-node/server.ts

@@ -419,7 +419,6 @@ export class AztecNodeService implements AztecNode, AztecNodeAdmin, AztecNodeDeb
   async #getCheckpointContextsForBlocks(
     blocks: { checkpointNumber: CheckpointNumber }[],
     // TODO(palla): CheckpointNumber should be accepted by this lint rule
-    // eslint-disable-next-line aztec-custom/no-non-primitive-in-collections
   ): Promise<Map<CheckpointNumber, { l1?: L1PublishedData; attestations?: CommitteeAttestation[] } | undefined>> {
     const unique = Array.from(new Set(blocks.map(b => b.checkpointNumber)));
     const entries = await Promise.all(unique.map(async n => [n, await this.#getCheckpointContext(n)] as const));

yarn-project/end-to-end/bootstrap.sh

@@ -33,10 +33,12 @@ function test_cmds {
     echo "$prefix:NAME=e2e_prover_full_fake FAKE_PROOFS=1 $run_test_script simple e2e_prover/full"
   fi
   echo "$prefix:TIMEOUT=15m:NAME=e2e_block_building $(set_dump_avm e2e_block_building) $run_test_script simple e2e_block_building"
+  echo "$prefix:TIMEOUT=15m:NAME=e2e_epochs/epochs_long_proving_time $run_test_script simple src/e2e_epochs/epochs_long_proving_time.test.ts"
 
   local tests=(
     # List all standalone and nested tests, except for the ones listed above.
-    src/e2e_!(prover)/*.test.ts
+    src/e2e_!(prover|epochs)/*.test.ts
+    src/e2e_epochs/!(epochs_long_proving_time).test.ts
     src/e2e_p2p/reqresp/*.test.ts
     src/e2e_!(block_building).test.ts
   )

yarn-project/end-to-end/src/e2e_epochs/epochs_long_proving_time.test.ts

@@ -6,7 +6,9 @@ import { jest } from '@jest/globals';
 
 import { EpochsTestContext } from './epochs_test.js';
 
-jest.setTimeout(1000 * 60 * 10);
+jest.setTimeout(1000 * 60 * 15);
+
+const MAX_JOB_COUNT = 20;
 
 describe('e2e_epochs/epochs_long_proving_time', () => {
   let logger: Logger;
@@ -24,11 +26,14 @@ describe('e2e_epochs/epochs_long_proving_time', () => {
     const { aztecSlotDuration } = EpochsTestContext.getSlotDurations({ aztecEpochDuration });
     const epochDurationInSeconds = aztecSlotDuration * aztecEpochDuration;
     const proverTestDelayMs = (epochDurationInSeconds * 1000 * 3) / 4;
+    // Each epoch takes ~3 epochs to prove, so the broker needs to keep results for
+    // at least that many epochs to avoid rejecting jobs as stale.
     test = await EpochsTestContext.setup({
       aztecEpochDuration,
       aztecProofSubmissionEpochs: 1000, // Effectively don't re-org
       proverTestDelayMs,
-      proverNodeMaxPendingJobs: 1, // We test for only a single job at once
+      proverNodeMaxPendingJobs: MAX_JOB_COUNT, // Prove multiple epochs concurrently
+      proverBrokerMaxEpochsToKeepResultsFor: 10,
       enableProposerPipelining: true,
     });
     ({ logger, monitor, L1_BLOCK_TIME_IN_S } = test);
@@ -59,10 +64,7 @@ describe('e2e_epochs/epochs_long_proving_time', () => {
     // At least 3 epochs should have passed after the proven one (though we add a -1 just in case)
     expect(monitor.checkpointNumber).toBeGreaterThanOrEqual(targetProvenEpochs * test.epochDuration * 3 - 1);
 
-    // We expect maxJobCount to equal 1, since the prover node epoch monitor defines an epoch as ready to be proven
-    // only if the previous one has already been proven. We can relax this check if we want to support multiple epochs
-    // to be proven in parallel, in which case we should update the assertion below.
-    expect(maxJobCount).toEqual(1);
-    logger.info(`Test succeeded`);
+    expect(maxJobCount).toBeLessThanOrEqual(MAX_JOB_COUNT);
+    logger.info(`Test succeeded, max prover jobs ${maxJobCount}`);
   });
 });

yarn-project/end-to-end/src/e2e_epochs/epochs_multi_proof.test.ts

@@ -46,18 +46,18 @@ describe('e2e_epochs/epochs_multi_proof', () => {
     // This prevents the race condition where multiple provers submit to L1 at the same time
     test.proverNodes.forEach((proverAztecNode, index) => {
       const proverManager = proverAztecNode.getProverNode()!.getProver();
-      const origCreateEpochProver = proverManager.createEpochProver.bind(proverManager);
-      proverManager.createEpochProver = () => {
-        const epochProver = origCreateEpochProver();
-        const origFinalizeEpoch = epochProver.finalizeEpoch.bind(epochProver);
-        epochProver.finalizeEpoch = async () => {
-          const result = await origFinalizeEpoch();
+      const origCreateTopTree = proverManager.createTopTreeOrchestrator.bind(proverManager);
+      proverManager.createTopTreeOrchestrator = () => {
+        const topTree = origCreateTopTree();
+        const origProve = topTree.prove.bind(topTree);
+        topTree.prove = async (...args: Parameters<typeof origProve>) => {
+          const result = await origProve(...args);
           const sleepTime = index * 1000 * test.constants.ethereumSlotDuration;
-          logger.warn(`Delaying finalizeEpoch for prover node ${index} by ${sleepTime}ms`);
+          logger.warn(`Delaying top-tree prove for prover node ${index} by ${sleepTime}ms`);
           await sleep(sleepTime);
           return result;
         };
-        return epochProver;
+        return topTree;
       };
     });