fix(fast-inbox): schedule proving jobs oldest-epoch-first across proof types (A-1427)

The legacy L1-to-L2 tree gated block roots on parity outputs, so the broker's
type-major priority still progressed one epoch at a time. The streaming inbox
parity only gates the checkpoint root: under sustained block production the
type-major order kept serving younger epochs' block roots and never scheduled
INBOX_PARITY, stalling every epoch at awaiting-root. Select the oldest-epoch
job across the allowed queues and tie-break by proof-type priority.

Author: spalladino

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

@@ -535,6 +535,29 @@ describe.each([
       );
     });
 
+    it('prefers an older epoch over a higher-priority proof type from a younger epoch', async () => {
+      const baseRollup2 = makeRandomProvingJobId();
+      await broker.enqueueProvingJob({
+        id: baseRollup2,
+        type: ProvingRequestType.PRIVATE_TX_BASE_ROLLUP,
+        epochNumber: EpochNumber(2),
+        inputsUri: makeInputsUri(),
+      });
+
+      const publicVm1 = makeRandomProvingJobId();
+      await broker.enqueueProvingJob({
+        id: publicVm1,
+        type: ProvingRequestType.PUBLIC_VM,
+        epochNumber: EpochNumber(1),
+        inputsUri: makeInputsUri(),
+      });
+
+      // A lower-priority type from epoch 1 wins over the higher-priority type from epoch 2: the oldest
+      // epoch's remaining jobs must complete rather than starve behind newer epochs' work.
+      await getAndAssertNextJobId(publicVm1, ProvingRequestType.PUBLIC_VM, ProvingRequestType.PRIVATE_TX_BASE_ROLLUP);
+      await getAndAssertNextJobId(baseRollup2, ProvingRequestType.PUBLIC_VM, ProvingRequestType.PRIVATE_TX_BASE_ROLLUP);
+    });
+
     it('returns any job if filter is empty', async () => {
       const baseParity1 = makeRandomProvingJobId();
       await broker.enqueueProvingJob({

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

@@ -410,35 +410,59 @@ export class ProvingBroker implements ProvingJobProducer, ProvingJobConsumer, Pr
         : Object.values(ProvingRequestType).filter((x): x is ProvingRequestType => typeof x === 'number');
     allowedProofs.sort(proofTypeComparator);
 
+    // Select the oldest-epoch job across the allowed queues, tie-breaking by proof-type priority: an epoch's
+    // remaining jobs always outrank younger epochs' work, so the oldest epoch completes instead of starving
+    // behind the continuous arrival of new higher-priority-type jobs. The legacy L1-to-L2 tree got this ordering
+    // for free (block roots waited on parity outputs); the streaming inbox parity only gates the checkpoint
+    // root, so a purely type-major order would leave it unscheduled under sustained block production.
+    let selected: { proofType: ProvingRequestType; enqueuedJob: EnqueuedProvingJob; job: ProvingJob } | undefined;
     for (const proofType of allowedProofs) {
       const queue = this.queues[proofType];
       let enqueuedJob: EnqueuedProvingJob | undefined;
+      let candidate: { enqueuedJob: EnqueuedProvingJob; job: ProvingJob } | undefined;
       // exhaust the queue and make sure we're not sending a job that's already in progress
       // or has already been completed
       // this can happen if the broker crashes and restarts
       // it's possible agents will report progress or results for jobs that are in the queue (after the restart)
       while ((enqueuedJob = queue.getImmediate())) {
         const job = this.jobsCache.get(enqueuedJob.id);
         if (job && !this.inProgress.has(enqueuedJob.id) && !this.resultsCache.has(enqueuedJob.id)) {
-          const time = this.msTimeSource();
-          this.inProgress.set(job.id, {
-            id: job.id,
-            startedAt: time,
-            lastUpdatedAt: time,
-          });
-          const enqueuedAt = this.enqueuedAt.get(job.id);
-          if (enqueuedAt) {
-            this.instrumentation.recordJobWait(job.type, enqueuedAt);
-            // we can clear this flag now.
-            this.enqueuedAt.delete(job.id);
-          }
-
-          return { job, time };
+          candidate = { enqueuedJob, job };
+          break;
         }
       }
+      if (!candidate) {
+        continue;
+      }
+      if (selected === undefined || candidate.enqueuedJob.epochNumber < selected.enqueuedJob.epochNumber) {
+        if (selected) {
+          this.queues[selected.proofType].put(selected.enqueuedJob);
+        }
+        selected = { proofType, ...candidate };
+      } else {
+        queue.put(candidate.enqueuedJob);
+      }
+    }
+
+    if (!selected) {
+      return undefined;
+    }
+
+    const { job } = selected;
+    const time = this.msTimeSource();
+    this.inProgress.set(job.id, {
+      id: job.id,
+      startedAt: time,
+      lastUpdatedAt: time,
+    });
+    const enqueuedAt = this.enqueuedAt.get(job.id);
+    if (enqueuedAt) {
+      this.instrumentation.recordJobWait(job.type, enqueuedAt);
+      // we can clear this flag now.
+      this.enqueuedAt.delete(job.id);
     }
 
-    return undefined;
+    return { job, time };
   }
 
   async #reportProvingJobError(