From 81f0bb3037b525688cf4d926f9713ad5e9fc0830 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nathan=20=F0=9F=94=B6=20Tarbert?= Date: Mon, 21 Sep 2026 15:00:54 -0400 Subject: [PATCH 1/2] fix(queue): claim every job type before dispatching, and rotate the start claimJobsByType charged the global budget after processClaimedJobs had already awaited the batch, so work that had finished still held slots. With a broad backlog the budget ran out partway down this.handlers insertion order and the types past that point were never claimed. A productive poll reschedules at 0ms against the same state, so the next poll made the identical choice: indefinite starvation of a fixed tail, not delay. PENDING_RESPONSE_SWEEP, the handler that returns stranded AI responses to the queue, was in that tail. Awaiting each batch inside the loop also meant type N+1 was not claimed until type N had drained, so one slow AI_RESPONSE at a 120s timeout held up ESCALATION. Claims now run for every type first, the budget is charged as each claim returns, and the starting offset rotates per poll so a spent budget does not strand the same types every time. Dispatch sits in a finally so a claim that rejects partway through cannot discard rows already claimed, which are PROCESSING with a live token and would otherwise wait for the reclaim sweep and burn an attempt. Does not address the concurrency accounting or /health: the poll still awaits the batch at the end, so activeJobs is 0 when availableSlots is computed. That needs dispatch-without-await and belongs with #182/#184. Refs #232 --- .../src/__tests__/worker-concurrency.test.ts | 204 ++++++++++++++++++ packages/outpost/queue/src/worker.ts | 95 ++++---- 2 files changed, 260 insertions(+), 39 deletions(-) diff --git a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts index 20eec81..2c7ddd9 100644 --- a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts +++ b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts @@ -318,3 +318,207 @@ describe('Worker per-type concurrency', () => { expect(callCount).toBeGreaterThanOrEqual(1); }); }); + +// ─── Claim-before-process (#232) ──────────────────────────────────────────── + +/** + * The global budget used to be charged after `processClaimedJobs` had already + * awaited the batch, so work that had finished still held slots. With a broad + * backlog the budget ran out partway down `this.handlers` insertion order and + * the same tail types were never reached — deterministically, because the next + * poll started from the same place against the same state. + */ +describe('Worker claim budget and ordering', () => { + let worker: InstanceType; + + /** Types in registration order; the last few are the ones that starved. */ + const TYPES = [ + JobType.AI_RESPONSE, + JobType.ESCALATION, + JobType.SLA_CHECK, + JobType.TRACKER_SYNC, + JobType.JOB_CLEANUP, + ]; + + /** `claimJobsForType` interpolates type third: jsonb map, default timeout, type, limit. */ + const claimedType = (call: unknown[]) => call[3] as string; + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + mockPrisma.$executeRaw.mockResolvedValue(0); + mockPrismaJob.updateMany.mockResolvedValue({ count: 1 }); + mockPrismaJob.update.mockResolvedValue({}); + }); + + afterEach(async () => { + if (worker) await worker.stop(); + vi.useRealTimers(); + }); + + it('reaches every registered type across polls when every type has a backlog', async () => { + worker = new Worker({ + pollIntervalMs: 10, + maxConcurrency: 4, // smaller than the backlog, so the budget must run out + batchSize: 2, + concurrencyByType: Object.fromEntries(TYPES.map((t) => [t, 2])), + }); + + for (const t of TYPES) worker.on(t, async () => ({ success: true })); + + // Every type always has work waiting. + let seq = 0; + mockPrisma.$queryRaw.mockImplementation((...call: unknown[]) => { + const type = claimedType(call); + return Promise.resolve([ + makeJobRow({ id: `${type}-${seq++}`, type }), + makeJobRow({ id: `${type}-${seq++}`, type }), + ]); + }); + + worker.start(); + // Several polls: any fair scheme reaches all five well inside this. + await vi.advanceTimersByTimeAsync(200); + + const reached = new Set(mockPrisma.$queryRaw.mock.calls.map(claimedType)); + for (const t of TYPES) { + expect(reached, `type ${t} was never claimed`).toContain(t); + } + }); + + it('issues every claim before any handler runs, so a slow type cannot delay the rest', async () => { + worker = new Worker({ + pollIntervalMs: 10, + maxConcurrency: 10, + batchSize: 1, + concurrencyByType: { [JobType.AI_RESPONSE]: 2, [JobType.ESCALATION]: 2 }, + }); + + const events: string[] = []; + let releaseSlow: () => void = () => {}; + const slowDone = new Promise((r) => (releaseSlow = r)); + + worker.on(JobType.AI_RESPONSE, async () => { + events.push('ai-handler-start'); + await slowDone; + return { success: true }; + }); + worker.on(JobType.ESCALATION, async () => { + events.push('esc-handler-start'); + return { success: true }; + }); + + mockPrisma.$queryRaw.mockImplementation((...call: unknown[]) => { + const type = claimedType(call); + events.push(`claim:${type}`); + if (type === JobType.AI_RESPONSE) + return Promise.resolve([makeJobRow({ id: 'ai-1', type })]); + if (type === JobType.ESCALATION) + return Promise.resolve([makeJobRow({ id: 'esc-1', type })]); + return Promise.resolve([]); + }); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + // The ESCALATION claim must be issued even though the AI handler is parked. + // Both events must be present, or the index comparison below proves nothing. + expect(events).toContain(`claim:${JobType.ESCALATION}`); + expect(events).toContain('ai-handler-start'); + expect(events.indexOf(`claim:${JobType.ESCALATION}`)).toBeLessThan( + events.indexOf('ai-handler-start'), + ); + + releaseSlow(); + await vi.advanceTimersByTimeAsync(0); + }); + + it('never claims more than the global ceiling across all types in one poll', async () => { + // Budget 4 with a per-type limit of 2 means the ceiling can only be + // honoured by summing across types: no single type can reach it alone. + worker = new Worker({ + pollIntervalMs: 10, + maxConcurrency: 4, + batchSize: 5, + concurrencyByType: Object.fromEntries(TYPES.map((t) => [t, 2])), + }); + + // Every handler parks, so the first poll cannot finish and cannot + // schedule a second one. Whatever is claimed is one poll's worth. + // Released before the assertions so `stop()` can drain in afterEach. + let release: () => void = () => {}; + const parked = new Promise((r) => (release = r)); + for (const t of TYPES) + worker.on(t, async () => { + await parked; + return { success: true }; + }); + + let claimed = 0; + const typesClaimed = new Set(); + mockPrisma.$queryRaw.mockImplementation((...call: unknown[]) => { + const type = claimedType(call); + const limit = call[4] as number; + const rows = Array.from({ length: limit }, (_, i) => + makeJobRow({ id: `${type}-${i}`, type }), + ); + claimed += rows.length; + typesClaimed.add(type); + return Promise.resolve(rows); + }); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + const claimedInOnePoll = claimed; + const spannedTypes = typesClaimed.size; + release(); + await vi.advanceTimersByTimeAsync(0); + + // Lower bound first: a claim path that did nothing would satisfy the + // ceiling trivially. + expect(claimedInOnePoll).toBeGreaterThan(0); + expect(spannedTypes).toBeGreaterThan(1); + expect(claimedInOnePoll).toBeLessThanOrEqual(4); + }); + + it('dispatches jobs already claimed when a later claim fails', async () => { + // Claiming every type before processing any of them means a rejection + // partway through must not discard the rows already claimed: they are + // PROCESSING with a live token, and dropping them leaves them for the + // reclaim sweep, which consumes an attempt. + worker = new Worker({ + pollIntervalMs: 10, + maxConcurrency: 10, + batchSize: 1, + concurrencyByType: { + [JobType.AI_RESPONSE]: 2, + [JobType.ESCALATION]: 2, + [JobType.SLA_CHECK]: 2, + }, + }); + + const ran: string[] = []; + for (const t of [JobType.AI_RESPONSE, JobType.ESCALATION, JobType.SLA_CHECK]) + worker.on(t, async (payload) => { + ran.push(String((payload as { id?: string }).id ?? 'x')); + return { success: true }; + }); + + let call = 0; + mockPrisma.$queryRaw.mockImplementation((...args: unknown[]) => { + const type = claimedType(args); + call += 1; + if (call === 3) return Promise.reject(new Error('connection pool timeout')); + return Promise.resolve([ + makeJobRow({ id: `${type}-1`, type, payload: { id: `${type}-1` } }), + ]); + }); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + // The two claims that succeeded before the failure must still have run. + expect(ran).toHaveLength(2); + }); +}); diff --git a/packages/outpost/queue/src/worker.ts b/packages/outpost/queue/src/worker.ts index 8ca273e..fefb263 100644 --- a/packages/outpost/queue/src/worker.ts +++ b/packages/outpost/queue/src/worker.ts @@ -109,6 +109,8 @@ export class Worker { private lastPollTime: Date | null = null; private upSince: Date | null = null; private shutdownResolve: (() => void) | null = null; + /** Rotates the per-type claim order so a spent budget starves a different tail each poll. */ + private claimCursor = 0; private stopPromise: Promise | null = null; private signalHandlers: { signal: string; handler: () => void }[] = []; @@ -440,54 +442,69 @@ export class Worker { /** * Claim jobs respecting per-type concurrency limits. - * For each registered job type that has available capacity, claim up to - * the available slots for that type. + * + * Every type is claimed before any handler runs, and the global budget is + * charged as each claim returns rather than after its batch has been + * processed. Both halves matter: + * + * - Charging after `processClaimedJobs` meant finished work still held + * slots, so a broad backlog exhausted the budget partway down + * `this.handlers` insertion order. The types past that point were never + * claimed, and because a productive poll reschedules at 0ms against the + * same state, the next poll made the identical choice. That is indefinite + * starvation of a fixed tail, not delay. + * - Awaiting each batch inside the loop meant type N+1 was not claimed until + * type N had fully drained, so one slow `AI_RESPONSE` (120s timeout) held + * up `ESCALATION`, the path that exists for when things are going wrong. + * + * The starting offset rotates each poll so that when the budget genuinely + * does run out, it does not run out at the same place every time. Without + * it a permanently saturated queue still starves whatever sorts last. */ private async claimJobsByType(globalSlots: number): Promise { - let totalProcessed = 0; - let remainingGlobalSlots = globalSlots; - - // Determine which types have capacity - const typesWithCapacity: Array<{ type: string; available: number }> = []; - - for (const [type] of this.handlers) { - if (remainingGlobalSlots <= 0) break; - - const typeLimit = this.concurrencyByType[type]; - const activeForType = this.activeJobsByType.get(type) ?? 0; - - if (typeLimit !== undefined) { - const available = typeLimit - activeForType; - if (available > 0) { - typesWithCapacity.push({ - type, - available: Math.min(available, remainingGlobalSlots), - }); - } - } else { - // No per-type limit; bound by global slots only - typesWithCapacity.push({ - type, - available: remainingGlobalSlots, - }); - } - } + const types = Array.from(this.handlers.keys()); + if (types.length === 0) return 0; - // Claim jobs for each type that has capacity - for (const { type, available } of typesWithCapacity) { - if (remainingGlobalSlots <= 0) break; + const offset = this.claimCursor % types.length; + this.claimCursor = (this.claimCursor + 1) % types.length; + const ordered = [...types.slice(offset), ...types.slice(0, offset)]; - const limit = Math.min(available, remainingGlobalSlots, this.batchSize); - const jobs = await this.claimJobsForType(type, limit); + let remainingGlobalSlots = globalSlots; + const claimed: Array = []; - if (jobs.length > 0) { - await this.processClaimedJobs(jobs); - totalProcessed += jobs.length; + try { + for (const type of ordered) { + if (remainingGlobalSlots <= 0) break; + + const typeLimit = this.concurrencyByType[type]; + const available = + typeLimit === undefined + ? remainingGlobalSlots + : typeLimit - (this.activeJobsByType.get(type) ?? 0); + if (available <= 0) continue; + + const limit = Math.min(available, remainingGlobalSlots, this.batchSize); + const jobs = await this.claimJobsForType(type, limit); + + claimed.push(...jobs); + // Charged here, against rows this poll actually holds, and before + // anything is dispatched. remainingGlobalSlots -= jobs.length; } + } finally { + // Dispatch whatever was claimed even if a later claim threw. Claiming + // every type before processing any of them means one failed claim + // mid-loop would otherwise drop rows that are already PROCESSING with + // a live token. Nothing would be holding them, so they would wait for + // the reclaim sweep, which consumes an attempt and moves them toward + // DEAD_LETTER. The previous shape processed each batch as it was + // claimed and so had no such window. + if (claimed.length > 0) { + await this.processClaimedJobs(claimed); + } } - return totalProcessed; + return claimed.length; } /** From 87fff09f0dec9c757e5bcc957e68058d1cca2f74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nathan=20=F0=9F=94=B6=20Tarbert?= <66887028+NathanTarbert@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:36:48 -0400 Subject: [PATCH 2/2] Describe claim-then-dispatch in the worker health comments --- apps/worker/src/__tests__/health.test.ts | 27 ++++++++++--------- apps/worker/src/health.ts | 33 ++++++++++++------------ packages/outpost/queue/src/types.ts | 8 +++--- packages/outpost/queue/src/worker.ts | 14 +++++----- 4 files changed, 43 insertions(+), 39 deletions(-) diff --git a/apps/worker/src/__tests__/health.test.ts b/apps/worker/src/__tests__/health.test.ts index 59eb174..85c5f88 100644 --- a/apps/worker/src/__tests__/health.test.ts +++ b/apps/worker/src/__tests__/health.test.ts @@ -148,10 +148,12 @@ describe('buildHealthResponse', () => { // ─── busy is not stalled ─────────────────────────────────────────────── // // THE REGRESSION THIS MODEL EXISTS FOR, and the one that got the previous - // attempt reverted. Two facts together: poll() stamps lastPollTime and then - // awaits its jobs, and claimJobsByType awaits each TYPE's batch sequentially - // (worker.ts) — so a single poll may legitimately run the SUM of every - // registered type's timeout. Against the config in index.ts that is 930s. + // attempt reverted. poll() stamps lastPollTime and then awaits every job it + // claimed, so a single poll legitimately runs as long as its slowest job. + // When claimJobsByType awaited each type's batch in turn, that was the sum + // of every registered type's timeout, 930s against the config in index.ts. + // Claims now all happen before dispatch, but the status writes after each + // handler are untimed, so the poll still has no fixed upper bound. // // Both earlier bounds were therefore wrong: `now - lastPollTime > 60s`, and // its replacement `pollDuration > max(jobTimeouts) + 60s` = 360s. Either one @@ -168,7 +170,8 @@ describe('buildHealthResponse', () => { ...over, }); - // 930_000 is the real sequential worst case for the ten registered types. + // 930_000 was the worst case when each type's batch ran in turn. It stays + // as a long-poll case: the probe must not care how the poll batches. it.each([61_000, 360_001, 600_000, 930_000, 1_200_000])( 'reports 200 after %ims of polling while jobs are in flight and none is overdue', (pollAgeMs) => { @@ -740,11 +743,11 @@ describe('a poll that keeps failing is not healthy', () => { }); // The claim clock runs from the later of "this poll started" and "a job last -// settled". claimJobsByType awaits each type's batch sequentially, so after a -// long batch finishes, the remaining claim queries each run with -// activeJobCount === 0 while pollDuration already carries that batch. Measuring -// the whole poll reported a healthy worker stalled — the reverted defect, moved -// rather than removed. +// settled". Claims all happen before dispatch, but after the last job of a long +// batch settles there is a short gap before the poll returns. In that gap +// activeJobCount === 0 while pollDuration already carries the batch. Measuring +// the whole poll would report a healthy worker stalled, the reverted defect +// moved rather than removed. describe('a settled job resets the claim clock', () => { it('stays 200 when a batch settled recently, however long the poll has run', () => { const afterLongBatch: WorkerHealthStatus = { @@ -752,8 +755,8 @@ describe('a settled job resets the claim clock', () => { activeJobCount: 0, pollStartedAt: new Date(NOW - 118_000), lastPollTime: new Date(NOW - 118_000), - // The AI_RESPONSE batch finished a moment ago; the poll is now - // issuing the next type's claim query. + // The AI_RESPONSE batch finished a moment ago and the poll is + // about to return. lastJobSettledAt: new Date(NOW - 200), }; diff --git a/apps/worker/src/health.ts b/apps/worker/src/health.ts index 33a1099..bdd5004 100644 --- a/apps/worker/src/health.ts +++ b/apps/worker/src/health.ts @@ -78,14 +78,14 @@ export const STALE_POLL_MS = 60_000; * query, so a poll that has not returned in a minute with no job to blame is * blocked on the database rather than working. * - * Deliberately NOT a bound on poll duration in general. claimJobsByType awaits - * each type's batch sequentially, so one poll can legitimately run the sum of - * every registered type's timeout — 930s against the real worker config. An - * earlier version of this module bounded the whole poll at - * `max(jobTimeouts) + 60s` = 360s and reported a healthy worker stalled, which - * is what got the previous attempt at this endpoint reverted. Whether the jobs - * themselves are overdue is the worker's own question to answer, and it answers - * it per job in `overdueJobCount`. + * Deliberately NOT a bound on poll duration in general. A poll claims every + * type up front and then waits for the whole batch, so it legitimately runs as + * long as its slowest job, and longer if a status write is slow. An earlier + * version of this module bounded the whole poll at `max(jobTimeouts) + 60s`, + * back when claiming awaited each type's batch in turn, and reported a healthy + * worker stalled. That is what got the previous attempt at this endpoint + * reverted. Whether the jobs themselves are overdue is the worker's own + * question to answer, and it answers it per job in `overdueJobCount`. */ export const CLAIM_STALL_MS = 60_000; @@ -439,16 +439,15 @@ export function buildHealthResponse( if (workerHealth.pollStartedAt) { // Nothing claimed and still not back: the claim query itself is blocked. - // With jobs in flight the duration says nothing, since a poll waits out - // every type's batch in turn. + // With jobs in flight the duration says nothing, since a poll waits for + // its whole claimed batch to settle. // Measured from the later of "this poll started" and "a job last - // settled", NOT from the start of the poll. claimJobsByType awaits each - // type's batch sequentially, so after a 118s batch of AI_RESPONSE jobs - // finishes, the nine remaining claim queries each run with - // activeJobCount === 0 and a pollDuration already carrying those 118s. - // Bounding the whole poll reported a healthy worker as stalled — the - // same mistake that got the previous two attempts reverted, moved rather - // than removed. + // settled", NOT from the start of the poll. Claims all happen before + // dispatch, but once the last job of a 118s batch settles there is a + // short gap before the poll returns. In that gap activeJobCount is 0 and + // the poll already carries those 118s. Bounding the whole poll would read + // that as a blocked claim and report a healthy worker as stalled, the + // same mistake that got the previous two attempts reverted. const claimingSince = Math.max( workerHealth.pollStartedAt.getTime(), workerHealth.lastJobSettledAt?.getTime() ?? 0, diff --git a/packages/outpost/queue/src/types.ts b/packages/outpost/queue/src/types.ts index d32b4a5..c4ed4a5 100644 --- a/packages/outpost/queue/src/types.ts +++ b/packages/outpost/queue/src/types.ts @@ -262,10 +262,10 @@ export interface WorkerHealthStatus { /** * In-flight jobs that have outlived their OWN timeout plus a grace. * - * The honest liveness signal. Poll duration is not: claimJobsByType awaits - * each type's batch sequentially, so a single poll may legitimately run the - * sum of every registered type's timeout. Non-zero here means the timeout - * machinery failed, not that the worker is slow. + * The honest liveness signal. Poll duration is not: a poll waits for its + * whole claimed batch, so it legitimately runs as long as its slowest job + * plus an untimed status write. Non-zero here means the timeout machinery + * failed, not that the worker is slow. */ overdueJobCount: number; /** When a job last settled, whatever the outcome. Null before the first one. */ diff --git a/packages/outpost/queue/src/worker.ts b/packages/outpost/queue/src/worker.ts index 3e29226..fe19e0e 100644 --- a/packages/outpost/queue/src/worker.ts +++ b/packages/outpost/queue/src/worker.ts @@ -299,12 +299,14 @@ export class Worker { * * This is the liveness signal, and it is deliberately per-job rather than * derived from poll duration. An earlier version of the health check bounded - * the whole poll at `max(jobTimeouts) + grace`, which is simply not what a - * poll is: claimJobsByType awaits each type's batch SEQUENTIALLY, so one poll - * can legitimately run the SUM of every registered type's timeout — 930s - * against the worker's real configuration, versus a 360s bound. A healthy - * worker working through a backlog was reported stalled and the container - * probe killed it mid-job. + * the whole poll at `max(jobTimeouts) + grace`. At the time claimJobsByType + * awaited each type's batch in turn, so one poll could run the sum of every + * registered type's timeout, and a healthy worker working through a backlog + * was reported stalled and killed mid-job. claimJobsByType now claims every + * type up front and runs the whole batch at once, so a poll lasts about as + * long as its slowest job. That is still not a bound worth encoding: the + * status write after each handler is untimed, and any poll-level bound has + * to be kept in step with the timeout table and with how poll() batches. * * Asking whether any single job has outlived its own timeout needs no * scheduling arithmetic, so it cannot drift out of step with how poll()