Skip to content

Commit 6ad8216

Browse files
committed
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.
1 parent 0a2faa0 commit 6ad8216

2 files changed

Lines changed: 70 additions & 15 deletions

File tree

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -407,6 +407,37 @@ describe.each([
407407
);
408408
});
409409

410+
it('prefers an older epoch over a higher-priority proof type from a younger epoch', async () => {
411+
const baseRollup2 = makeRandomProvingJobId();
412+
await broker.enqueueProvingJob({
413+
id: baseRollup2,
414+
type: ProvingRequestType.PRIVATE_TX_BASE_ROLLUP,
415+
epochNumber: EpochNumber(2),
416+
inputsUri: makeInputsUri(),
417+
});
418+
419+
const inboxParity1 = makeRandomProvingJobId();
420+
await broker.enqueueProvingJob({
421+
id: inboxParity1,
422+
type: ProvingRequestType.INBOX_PARITY,
423+
epochNumber: EpochNumber(1),
424+
inputsUri: makeInputsUri(),
425+
});
426+
427+
// The lowest-priority type from epoch 1 wins over the higher-priority type from epoch 2: the oldest
428+
// epoch's remaining jobs must complete rather than starve behind newer epochs' work.
429+
await getAndAssertNextJobId(
430+
inboxParity1,
431+
ProvingRequestType.INBOX_PARITY,
432+
ProvingRequestType.PRIVATE_TX_BASE_ROLLUP,
433+
);
434+
await getAndAssertNextJobId(
435+
baseRollup2,
436+
ProvingRequestType.INBOX_PARITY,
437+
ProvingRequestType.PRIVATE_TX_BASE_ROLLUP,
438+
);
439+
});
440+
410441
it('returns any job if filter is empty', async () => {
411442
const baseParity1 = makeRandomProvingJobId();
412443
await broker.enqueueProvingJob({

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

Lines changed: 39 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -365,35 +365,59 @@ export class ProvingBroker implements ProvingJobProducer, ProvingJobConsumer, Pr
365365
: Object.values(ProvingRequestType).filter((x): x is ProvingRequestType => typeof x === 'number');
366366
allowedProofs.sort(proofTypeComparator);
367367

368+
// Select the oldest-epoch job across the allowed queues, tie-breaking by proof-type priority: an epoch's
369+
// remaining jobs always outrank younger epochs' work, so the oldest epoch completes instead of starving
370+
// behind the continuous arrival of new higher-priority-type jobs. The legacy L1-to-L2 tree got this ordering
371+
// for free (block roots waited on parity outputs); the streaming inbox parity only gates the checkpoint
372+
// root, so a purely type-major order would leave it unscheduled under sustained block production.
373+
let selected: { proofType: ProvingRequestType; enqueuedJob: EnqueuedProvingJob; job: ProvingJob } | undefined;
368374
for (const proofType of allowedProofs) {
369375
const queue = this.queues[proofType];
370376
let enqueuedJob: EnqueuedProvingJob | undefined;
377+
let candidate: { enqueuedJob: EnqueuedProvingJob; job: ProvingJob } | undefined;
371378
// exhaust the queue and make sure we're not sending a job that's already in progress
372379
// or has already been completed
373380
// this can happen if the broker crashes and restarts
374381
// it's possible agents will report progress or results for jobs that are in the queue (after the restart)
375382
while ((enqueuedJob = queue.getImmediate())) {
376383
const job = this.jobsCache.get(enqueuedJob.id);
377384
if (job && !this.inProgress.has(enqueuedJob.id) && !this.resultsCache.has(enqueuedJob.id)) {
378-
const time = this.msTimeSource();
379-
this.inProgress.set(job.id, {
380-
id: job.id,
381-
startedAt: time,
382-
lastUpdatedAt: time,
383-
});
384-
const enqueuedAt = this.enqueuedAt.get(job.id);
385-
if (enqueuedAt) {
386-
this.instrumentation.recordJobWait(job.type, enqueuedAt);
387-
// we can clear this flag now.
388-
this.enqueuedAt.delete(job.id);
389-
}
390-
391-
return { job, time };
385+
candidate = { enqueuedJob, job };
386+
break;
392387
}
393388
}
389+
if (!candidate) {
390+
continue;
391+
}
392+
if (selected === undefined || candidate.enqueuedJob.epochNumber < selected.enqueuedJob.epochNumber) {
393+
if (selected) {
394+
this.queues[selected.proofType].put(selected.enqueuedJob);
395+
}
396+
selected = { proofType, ...candidate };
397+
} else {
398+
queue.put(candidate.enqueuedJob);
399+
}
400+
}
401+
402+
if (!selected) {
403+
return undefined;
404+
}
405+
406+
const { job } = selected;
407+
const time = this.msTimeSource();
408+
this.inProgress.set(job.id, {
409+
id: job.id,
410+
startedAt: time,
411+
lastUpdatedAt: time,
412+
});
413+
const enqueuedAt = this.enqueuedAt.get(job.id);
414+
if (enqueuedAt) {
415+
this.instrumentation.recordJobWait(job.type, enqueuedAt);
416+
// we can clear this flag now.
417+
this.enqueuedAt.delete(job.id);
394418
}
395419

396-
return undefined;
420+
return { job, time };
397421
}
398422

399423
async #reportProvingJobError(

0 commit comments

Comments
 (0)