feat(prover): revive cancelled proving jobs from a persisted aborted state (#24578)

v5 version of #24558 (same change, based on `merge-train/spartan-v5`).

## Problem

When the prover broker cancelled a proving job it settled the job as a
cached `{ status: 'rejected', reason: 'Aborted' }` result — in memory
and in the database. Because a job's id is a deterministic hash of its
inputs, any later enqueue of the same job returned that cached abort
(`Cached proving job … Not enqueuing again`) and the facade failed the
job with `Aborted`. So once a proof was aborted it stayed aborted, and a
retry — a fresh epoch attempt after a reorg/failure, or a
re-orchestration after a restart — got the cached abort instead of
re-proving. A planned mid-epoch redeploy hit this and turned into a
testnet epoch prune.

## Fix

Cancellation is now its own first-class, **revivable** state:

- A new `aborted` proving-job status (`ProvingJobStatus` /
`ProvingJobSettledResult`). `cancelProvingJob` records it, notifies the
current waiter, and **persists** it
(`ProvingBrokerDatabase.setProvingJobAborted`).
- `enqueueProvingJob` **revives** an aborted job: when the producer
re-requests it, the broker clears the aborted state — in memory and, via
the new `deleteProvingJobResult`, in the database — and re-enqueues it
as if new. So a restart mid-revival keeps the job pending rather than
resurrecting the abort. Genuine agent errors and timeouts still settle
as terminal `rejected` results.

Net effect: cancelling frees the job cleanly and records why, but
re-requesting the same proof always brings it back to life — in the same
process or after a restart — so an abort can never permanently block an
epoch from proving.

The stacked PR (prover-node clean-shutdown, v5) stops a clean
prover-node shutdown from cancelling its in-flight jobs in the first
place.

## Tests

`proving_broker.test.ts` (both DB variants): cancel leaves a job
`aborted`; re-request revives it and it completes; the aborted state
persists across a restart; and a revived job survives a restart
mid-revival as pending (not aborted).

Note: not run locally in this session — the prover-client suite needs a
full noir/wasm bootstrap that wasn't available here — so relied on CI.

---
*Created by
[claudebox](https://claudebox.work/v2/sessions/6fa5e242ed9ceae7) ·
group: `slackbot`*

---------

Co-authored-by: Phil Windle <philip.windle@gmail.com>
(cherry picked from commit a4e88ca)

Author: 2 people

yarn-project/prover-client/src/proving_broker/broker_prover_facade.ts

@@ -288,6 +288,8 @@ export class BrokerCircuitProverFacade implements ServerCircuitProver {
         }
       } else if (result.status === 'rejected') {
         return { success: false, reason: result.reason };
+      } else if (result.status === 'aborted') {
+        return { success: false, reason: 'Aborted' };
       } else {
         throw new Error(`Unexpected proving job status ${result.status}`);
       }

yarn-project/prover-client/src/proving_broker/proving_broker.test.ts

@@ -201,7 +201,7 @@ describe.each([
       await assertJobStatus(id, 'in-queue');
 
       await broker.cancelProvingJob(id);
-      await assertJobStatus(id, 'rejected');
+      await assertJobStatus(id, 'aborted');
     });
 
     it('cancels jobs in-progress', async () => {
@@ -216,7 +216,135 @@ describe.each([
       await broker.getProvingJob();
       await assertJobStatus(id, 'in-progress');
       await broker.cancelProvingJob(id);
-      await assertJobStatus(id, 'rejected');
+      await assertJobStatus(id, 'aborted');
+    });
+
+    it('revives an aborted job when its producer re-requests it', async () => {
+      const provingJob: ProvingJob = {
+        id: makeRandomProvingJobId(),
+        type: ProvingRequestType.PARITY_BASE,
+        epochNumber: EpochNumber(1),
+        inputsUri: makeInputsUri(),
+      };
+
+      await broker.enqueueProvingJob(provingJob);
+      await broker.getProvingJob();
+      await assertJobStatus(provingJob.id, 'in-progress');
+
+      await broker.cancelProvingJob(provingJob.id);
+      await assertJobStatus(provingJob.id, 'aborted');
+
+      // Re-requesting the same job (a retry without a restart) revives it rather than returning the
+      // cached abort. The job stays cached through the revive, so the start-of-call status is
+      // 'in-queue', and it can then be completed.
+      await expect(broker.enqueueProvingJob(provingJob)).resolves.toEqual({ status: 'in-queue' });
+      await assertJobStatus(provingJob.id, 'in-queue');
+      const returnedJob = await broker.getProvingJob({ allowList: [ProvingRequestType.PARITY_BASE] });
+      expect(returnedJob?.job).toEqual(provingJob);
+
+      const retryValue = makeOutputsUri();
+      await broker.reportProvingJobSuccess(provingJob.id, retryValue);
+      await assertJobStatus(provingJob.id, 'fulfilled');
+    });
+
+    it('persists the aborted state across a restart and revives on re-request', async () => {
+      const provingJob: ProvingJob = {
+        id: makeRandomProvingJobId(),
+        type: ProvingRequestType.PARITY_BASE,
+        epochNumber: EpochNumber(1),
+        inputsUri: makeInputsUri(),
+      };
+
+      await broker.enqueueProvingJob(provingJob);
+      await broker.cancelProvingJob(provingJob.id);
+      await assertJobStatus(provingJob.id, 'aborted');
+
+      // A deploy restarts both the prover node and the broker. The aborted state is persisted, so it
+      // survives the restart rather than being lost or restored as a permanent rejection.
+      await broker.stop();
+      broker = new ProvingBroker(database, {
+        proverBrokerJobTimeoutMs: jobTimeoutMs,
+        proverBrokerPollIntervalMs: brokerIntervalMs,
+        proverBrokerJobMaxRetries: maxRetries,
+        proverBrokerMaxEpochsToKeepResultsFor: 1,
+        proverBrokerDebugReplayEnabled: false,
+      });
+      await broker.start();
+      await assertJobStatus(provingJob.id, 'aborted');
+
+      // The prover re-requests the job after the restart, which revives it (returning the cached
+      // 'in-queue' status), and it can be completed.
+      await expect(broker.enqueueProvingJob(provingJob)).resolves.toEqual({ status: 'in-queue' });
+      await assertJobStatus(provingJob.id, 'in-queue');
+      const returnedJob = await broker.getProvingJob({ allowList: [ProvingRequestType.PARITY_BASE] });
+      expect(returnedJob?.job).toEqual(provingJob);
+
+      const value = makeOutputsUri();
+      await broker.reportProvingJobSuccess(provingJob.id, value);
+      await assertJobStatus(provingJob.id, 'fulfilled');
+    });
+
+    it('persists the revived (non-aborted) state, so a restart mid-revival stays revived', async () => {
+      const provingJob: ProvingJob = {
+        id: makeRandomProvingJobId(),
+        type: ProvingRequestType.PARITY_BASE,
+        epochNumber: EpochNumber(1),
+        inputsUri: makeInputsUri(),
+      };
+
+      await broker.enqueueProvingJob(provingJob);
+      await broker.cancelProvingJob(provingJob.id);
+      await assertJobStatus(provingJob.id, 'aborted');
+
+      // Revive the job but do not complete it before the broker restarts.
+      await broker.enqueueProvingJob(provingJob);
+      await assertJobStatus(provingJob.id, 'in-queue');
+
+      await broker.stop();
+      broker = new ProvingBroker(database, {
+        proverBrokerJobTimeoutMs: jobTimeoutMs,
+        proverBrokerPollIntervalMs: brokerIntervalMs,
+        proverBrokerJobMaxRetries: maxRetries,
+        proverBrokerMaxEpochsToKeepResultsFor: 1,
+        proverBrokerDebugReplayEnabled: false,
+      });
+      await broker.start();
+
+      // Reviving cleared the persisted aborted state, so the job comes back pending rather than
+      // aborted and keeps being proven without needing another re-request.
+      await assertJobStatus(provingJob.id, 'in-queue');
+    });
+
+    it('revives once when the producer re-requests an aborted job concurrently', async () => {
+      const provingJob: ProvingJob = {
+        id: makeRandomProvingJobId(),
+        type: ProvingRequestType.PARITY_BASE,
+        epochNumber: EpochNumber(1),
+        inputsUri: makeInputsUri(),
+      };
+
+      await broker.enqueueProvingJob(provingJob);
+      await broker.cancelProvingJob(provingJob.id);
+      await assertJobStatus(provingJob.id, 'aborted');
+
+      // Two overlapping re-requests race through the revive. jobsCache stays populated as the enqueue
+      // lock throughout, so exactly one of them re-enqueues the job and both observe it as in-queue
+      // rather than the stale abort. The revived job is then delivered once and completes normally.
+      const [first, second] = await Promise.all([
+        broker.enqueueProvingJob(provingJob),
+        broker.enqueueProvingJob(provingJob),
+      ]);
+      expect(first).toEqual({ status: 'in-queue' });
+      expect(second).toEqual({ status: 'in-queue' });
+      await assertJobStatus(provingJob.id, 'in-queue');
+
+      const returnedJob = await broker.getProvingJob({ allowList: [ProvingRequestType.PARITY_BASE] });
+      expect(returnedJob?.job).toEqual(provingJob);
+      await assertJobStatus(provingJob.id, 'in-progress');
+      await expect(broker.getProvingJob({ allowList: [ProvingRequestType.PARITY_BASE] })).resolves.toBeUndefined();
+
+      await broker.reportProvingJobSuccess(provingJob.id, makeOutputsUri());
+      await assertJobStatus(provingJob.id, 'fulfilled');
     });
 
     it('returns job result if successful', async () => {
@@ -523,7 +651,7 @@ describe.each([
       await broker.getProvingJob();
       await assertJobStatus(id, 'in-progress');
       await broker.cancelProvingJob(id);
-      await assertJobStatus(id, 'rejected');
+      await assertJobStatus(id, 'aborted');
 
       const id2 = makeRandomProvingJobId();
       await broker.enqueueProvingJob({

yarn-project/prover-client/src/proving_broker/proving_broker.ts

@@ -272,15 +272,40 @@ export class ProvingBroker implements ProvingJobProducer, ProvingJobConsumer, Pr
 
   async #enqueueProvingJob(job: ProvingJob): Promise<ProvingJobStatus> {
     // We return the job status at the start of this call
-    const jobStatus = this.#getProvingJobStatus(job.id);
+    let jobStatus = this.#getProvingJobStatus(job.id);
     if (this.jobsCache.has(job.id)) {
       const existing = this.jobsCache.get(job.id);
       assert.deepStrictEqual(job, existing, 'Duplicate proving job ID');
-      this.logger.warn(`Cached proving job id=${job.id} epochNumber=${job.epochNumber}. Not enqueuing again`, {
-        provingJobId: job.id,
-      });
-      this.instrumentation.incCachedJobs(job.type);
-      return jobStatus;
+
+      if (this.resultsCache.get(job.id)?.status === 'aborted') {
+        // The producer is re-requesting a job it previously cancelled: revive it rather than
+        // returning the cached abort, clearing the aborted state in memory and in the database so the
+        // revival survives a restart.
+        //
+        // Concurrency model: `jobsCache` is the enqueue lock. Every path that puts a job on the queue
+        // populates `jobsCache` *synchronously, before its first await* (see the "New proving job"
+        // block below), so a second concurrent enqueue of the same id observes the entry at the top
+        // of this method and takes a cached, no-op branch instead of enqueuing a duplicate. The revive
+        // must keep holding that lock: we tear down the settled state and re-set `jobsCache` in a
+        // single synchronous span (no await in between), and only then await the database. Because a
+        // concurrent re-request can only interleave at that await — by which point `jobsCache` is
+        // populated again and the aborted result is gone — it falls into the cached branch and no-ops,
+        // so the job is enqueued exactly once. (`cleanUpProvingJobState` also drops the promise, which
+        // was resolved with the abort, so `enqueueJobInternal` below mints a fresh one for the retry.)
+        this.logger.info(`Reviving aborted proving job id=${job.id} epochNumber=${job.epochNumber}`, {
+          provingJobId: job.id,
+        });
+        this.cleanUpProvingJobState([job.id]);
+        this.jobsCache.set(job.id, job);
+        await this.database.deleteProvingJobResult(job.id);
+        jobStatus = this.#getProvingJobStatus(job.id);
+      } else {
+        this.logger.warn(`Cached proving job id=${job.id} epochNumber=${job.epochNumber}. Not enqueuing again`, {
+          provingJobId: job.id,
+        });
+        this.instrumentation.incCachedJobs(job.type);
+        return jobStatus;
+      }
     }
 
     if (this.isJobStale(job)) {
@@ -306,15 +331,35 @@ export class ProvingBroker implements ProvingJobProducer, ProvingJobConsumer, Pr
   }
 
   async #cancelProvingJob(id: ProvingJobId): Promise<void> {
-    if (!this.jobsCache.has(id)) {
+    const job = this.jobsCache.get(id);
+    if (!job) {
       this.logger.warn(`Can't cancel a job that doesn't exist id=${id}`, { provingJobId: id });
       return;
     }
 
-    // notify listeners of the cancellation
-    if (!this.resultsCache.has(id)) {
-      this.logger.info(`Cancelling job id=${id}`, { provingJobId: id });
-      await this.#reportProvingJobError(id, 'Aborted', false, undefined, true);
+    // Leave jobs that have already settled (completed or failed) alone: those results are terminal.
+    if (this.resultsCache.has(id)) {
+      return;
+    }
+
+    this.logger.info(`Cancelling job id=${id}`, { provingJobId: id });
+    this.inProgress.delete(id);
+
+    // Record the cancellation as its own settled state and persist it, so it survives a restart and
+    // notifies the current waiter. Unlike a completion or failure this is not terminal: re-enqueuing
+    // the same job id revives it (see #enqueueProvingJob), so the abort never permanently blocks the
+    // proof.
+    const result: ProvingJobSettledResult = { status: 'aborted' };
+    this.resultsCache.set(id, result);
+    this.promises.get(id)?.resolve(result);
+    this.completedJobNotifications.push(id);
+    this.instrumentation.incAbortedJobs(job.type);
+
+    try {
+      await this.database.setProvingJobAborted(id);
+    } catch (saveErr) {
+      this.logger.error(`Failed to save proving job aborted status id=${id}`, saveErr, { provingJobId: id });
+      throw saveErr;
     }
   }
 
@@ -401,7 +446,6 @@ export class ProvingBroker implements ProvingJobProducer, ProvingJobConsumer, Pr
     err: string,
     retry = false,
     filter?: ProvingJobFilter,
-    aborted = false,
   ): Promise<GetProvingJobResponse | undefined> {
     const info = this.inProgress.get(id);
     const item = this.jobsCache.get(id);
@@ -462,11 +506,7 @@ export class ProvingBroker implements ProvingJobProducer, ProvingJobConsumer, Pr
     this.promises.get(id)!.resolve(result);
     this.completedJobNotifications.push(id);
 
-    if (aborted) {
-      this.instrumentation.incAbortedJobs(item.type);
-    } else {
-      this.instrumentation.incRejectedJobs(item.type);
-    }
+    this.instrumentation.incRejectedJobs(item.type);
     if (info) {
       const duration = this.msTimeSource() - info.startedAt;
       this.instrumentation.recordJobDuration(item.type, duration);

yarn-project/prover-client/src/proving_broker/proving_broker_database.ts

@@ -38,6 +38,22 @@ export interface ProvingBrokerDatabase {
    */
   setProvingJobError(id: ProvingJobId, err: string): Promise<void>;
 
+  /**
+   * Records that a proof request was cancelled. Unlike a result or error this is not terminal:
+   * re-enqueuing the same job id revives it, so the aborted state can survive a restart without
+   * permanently blocking the proof.
+   * @param id - The ID of the cancelled proof request
+   */
+  setProvingJobAborted(id: ProvingJobId): Promise<void>;
+
+  /**
+   * Clears any stored result for a proof request, returning it to the pending state while keeping
+   * the job itself. Used when reviving an aborted job so the revival is persisted and a restart
+   * cannot resurrect the stale aborted state.
+   * @param id - The ID of the proof request whose result should be cleared
+   */
+  deleteProvingJobResult(id: ProvingJobId): Promise<void>;
+
   /**
    * Closes the database
    */

yarn-project/prover-client/src/proving_broker/proving_broker_database/broker_persisted_database.test.ts

@@ -132,6 +132,28 @@ describe('ProvingBrokerPersistedDatabase', () => {
     }
   });
 
+  it('keeps a queued result-delete ordered after the writes it follows', async () => {
+    const id = makeRandomProvingJobId(EpochNumber(42));
+    const job: ProvingJob = {
+      id,
+      epochNumber: EpochNumber(42),
+      type: ProvingRequestType.PARITY_BASE,
+      inputsUri: makeInputsUri(),
+    };
+    await db.addProvingJob(job);
+
+    // Abort the job and then delete its result without awaiting the abort first — mimicking a revive
+    // that re-requests a job whose aborted write may not have flushed yet. Because the delete rides
+    // the same write queue, it stays ordered after the abort and the result ends up cleared. If it
+    // were applied out of order the abort would resurrect on reload.
+    const abortWrite = db.setProvingJobAborted(id);
+    const deleteWrite = db.deleteProvingJobResult(id);
+    await Promise.all([abortWrite, deleteWrite]);
+
+    const allJobs = await toArray(db.allProvingJobs());
+    expect(allJobs).toEqual([[job, undefined]]);
+  });
+
   it('can add items over multiple epochs', async () => {
     const numJobs = 5;
     const startEpoch = 12;

yarn-project/prover-client/src/proving_broker/proving_broker_database/memory.ts

@@ -36,6 +36,16 @@ export class InMemoryBrokerDatabase implements ProvingBrokerDatabase {
     return Promise.resolve();
   }
 
+  setProvingJobAborted(id: ProvingJobId): Promise<void> {
+    this.results.set(id, { status: 'aborted' });
+    return Promise.resolve();
+  }
+
+  deleteProvingJobResult(id: ProvingJobId): Promise<void> {
+    this.results.delete(id);
+    return Promise.resolve();
+  }
+
   deleteProvingJobs(ids: ProvingJobId[]): Promise<void> {
     for (const id of ids) {
       this.jobs.delete(id);