diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index da17a35..8123014 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -25,6 +25,7 @@ import { FallbackModel, GeminiModel, OpenAIModel, + isBudgetExhausted, type FallbackEntry, } from '@outreachgraph/ai'; import { ImapReader, ResendMailer } from '@outreachgraph/email'; @@ -683,11 +684,23 @@ async function tick(): Promise { // The queue drains before the send sweep, so anything discovered this tick // can go out on the same tick rather than waiting for the next one. - const drained = await drainQueue(db, runJob); + const drained = await drainQueue(db, runJob, { isOutage: isBudgetExhausted }); if (drained.processed > 0 || drained.reclaimed > 0) { console.log( `jobs: ${drained.succeeded} done, ${drained.retried} retrying, ` + - `${drained.dead} failed, ${drained.reclaimed} reclaimed`, + `${drained.dead} failed, ${drained.deferred} deferred, ` + + `${drained.reclaimed} reclaimed`, + ); + } + + // Said plainly and every tick it happens, because the previous version of + // this outage was silent: the queue drained itself to death against providers + // that had no budget left, and the only trace was a 400 in `jobs.last_error` + // that read like a bug in the crawler. + if (drained.deferred > 0) { + console.log( + 'jobs: every model provider is refusing work, so the queue is holding ' + + 'rather than spending attempts — it resumes on its own once they answer', ); } diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index d25d125..8deba1e 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -245,11 +245,13 @@ export { JOB_STATUSES, type BatchItem, type BatchStatus, + type DrainOptions, type DrainSummary, type EnqueueInput, type EnqueueResult, type JobHandler, type JobStatus, + type OutageCheck, type QueuedJob, } from './queue'; diff --git a/packages/pipeline/src/queue.test.ts b/packages/pipeline/src/queue.test.ts index f7954e0..f9ac019 100644 --- a/packages/pipeline/src/queue.test.ts +++ b/packages/pipeline/src/queue.test.ts @@ -268,9 +268,109 @@ describe('drainQueue', () => { }); } - const summary = await drainQueue(db, async () => {}, 2); + const summary = await drainQueue(db, async () => {}, { limit: 2 }); expect(summary.processed).toBe(2); expect((await queueDepth(db)).pending).toBe(3); }); }); + +describe('failJob during an outage', () => { + // Every failure here is the shared model chain refusing everybody, which is + // what the real incident looked like: a 400 carrying a billing message. + const outOfBudget = (error: unknown): boolean => String(error).includes('usage limits'); + + test('an outage does not spend the last attempt', async () => { + const { db } = await setup('queue-outage-refund'); + + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'crawl_site', + maxAttempts: 1, + }); + + // Claiming charges the attempt up front, so this job is on its last one — + // the exact state in which the incident killed 4,888 rows. + const job = await claimNext(db); + expect(job?.attempts).toBe(1); + + const outcome = await failJob( + db, + job!, + new Error('You have reached your specified API usage limits.'), + outOfBudget, + ); + + expect(outcome).toBe('deferred'); + + const row = await queryOne<{ status: string; attempts: number; run_after: string }>( + db, + 'SELECT status, attempts, run_after FROM jobs WHERE id = ?', + [job!.id], + ); + + // Alive, and with its attempt handed back rather than burnt on an outage. + expect(row?.status).toBe('pending'); + expect(Number(row?.attempts)).toBe(0); + // Held for a while, so a provider that has already said no is not asked + // again on the very next tick. + expect(new Date(row!.run_after).getTime()).toBeGreaterThan(Date.now() + 60_000); + }); + + test('a genuine fault still dies, outage check or not', async () => { + const { db } = await setup('queue-outage-not-everything'); + + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'crawl_site', + maxAttempts: 1, + }); + + const job = await claimNext(db); + const outcome = await failJob(db, job!, new Error('unparseable homepage'), outOfBudget); + + // The refund is for outages only; widening it would let a poisonous payload + // retry forever. + expect(outcome).toBe('dead'); + + const row = await queryOne<{ status: string }>(db, 'SELECT status FROM jobs WHERE id = ?', [ + job!.id, + ]); + + expect(row?.status).toBe('failed'); + }); + + test('one outage holds the rest of the tick instead of burning it', async () => { + const { db } = await setup('queue-outage-stops-tick'); + + for (let i = 0; i < 4; i += 1) { + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'crawl_site', + payload: { i }, + maxAttempts: 3, + }); + } + + let calls = 0; + const summary = await drainQueue( + db, + async () => { + calls += 1; + throw new Error('You have reached your specified API usage limits.'); + }, + { isOutage: outOfBudget }, + ); + + // The outage is global, so there was no point in asking four times. + expect(calls).toBe(1); + expect(summary.processed).toBe(1); + expect(summary.deferred).toBe(1); + expect(summary.dead).toBe(0); + + // Nothing died, and the queue is intact for when the budget returns. + const depth = await queueDepth(db); + expect(depth.failed).toBe(0); + expect(depth.pending).toBe(4); + }); +}); diff --git a/packages/pipeline/src/queue.ts b/packages/pipeline/src/queue.ts index fd68cd0..6d5d626 100644 --- a/packages/pipeline/src/queue.ts +++ b/packages/pipeline/src/queue.ts @@ -23,6 +23,17 @@ const BASE_BACKOFF_MS = 30_000; /** Ceiling on the doubling, so attempt 20 is not scheduled for next year. */ const MAX_BACKOFF_MS = 3_600_000; +/** + * How long a job waits after a failure that was not its own fault. + * + * Flat rather than doubling, because there is no escalating suspicion to + * express: the job is fine and the world is not. Ten minutes is long enough + * that a provider which has already refused everyone is not asked again every + * tick, and short enough that the queue comes back to life within minutes of + * the outage clearing rather than within the hour. + */ +const OUTAGE_BACKOFF_MS = 600_000; + /** * How long a claimed job may stay `running` before it is considered abandoned. * @@ -180,7 +191,16 @@ export async function completeJob(db: Client, id: string): Promise { }); } -export type FailOutcome = 'retry' | 'dead'; +export type FailOutcome = 'retry' | 'dead' | 'deferred'; + +/** + * Recognises a failure caused by shared infrastructure rather than by this job. + * + * Injected rather than imported so this module keeps knowing nothing about what + * a job actually does: the server passes `isBudgetExhausted`, and a test passes + * whatever it needs to. + */ +export type OutageCheck = (error: unknown) => boolean; /** * Records a failed attempt, and either schedules a retry or gives up. @@ -189,9 +209,37 @@ export type FailOutcome = 'retry' | 'dead'; * its last error, which is the only way to answer "why did that URL never * produce a card". */ -export async function failJob(db: Client, job: QueuedJob, error: unknown): Promise { +export async function failJob( + db: Client, + job: QueuedJob, + error: unknown, + isOutage?: OutageCheck, +): Promise { const stamp = now(); const message = error instanceof Error ? error.message : String(error); + + // A job must not be charged an attempt for someone else's outage. Charging it + // is how three days of exhausted model budget silently emptied the queue: + // every job burned through `max_attempts` against providers that were + // refusing everybody, and nothing ever brings a `failed` row back, so the + // pipeline reached a fixed point it could not leave once the budget returned. + // + // Refunding is safe here in a way it deliberately is not in `reclaimStalled`: + // there the suspect is the payload, and the evidence is that this job killed + // its container. Here the evidence points the other way — every other job in + // the queue is failing identically, which is precisely what makes the payload + // innocent. + if (isOutage?.(error)) { + await db.execute({ + sql: `UPDATE jobs SET status = 'pending', attempts = MAX(attempts - 1, 0), + last_error = ?, run_after = ?, started_at = NULL, updated_at = ? + WHERE id = ?`, + args: [message.slice(0, 2000), plus(stamp, OUTAGE_BACKOFF_MS), stamp, job.id], + }); + + return 'deferred'; + } + const exhausted = job.attempts >= job.maxAttempts; if (exhausted) { @@ -253,9 +301,18 @@ export interface DrainSummary { readonly succeeded: number; readonly retried: number; readonly dead: number; + /** Put back untouched because the failure was an outage, not a fault. */ + readonly deferred: number; readonly reclaimed: number; } +export interface DrainOptions { + /** Ceiling on how many jobs one tick runs. */ + readonly limit?: number; + /** Recognises a failure that means "not this job's fault". */ + readonly isOutage?: OutageCheck; +} + /** * Runs up to `limit` jobs, one at a time. * @@ -268,13 +325,16 @@ export interface DrainSummary { export async function drainQueue( db: Client, handler: JobHandler, - limit = 25, + options: DrainOptions = {}, ): Promise { + const { limit = 25, isOutage } = options; + const reclaimed = await reclaimStalled(db); let processed = 0; let succeeded = 0; let retried = 0; let dead = 0; + let deferred = 0; for (let i = 0; i < limit; i += 1) { const job = await claimNext(db); @@ -289,13 +349,23 @@ export async function drainQueue( } catch (error) { // A handler throwing is an expected outcome, not a bug in the loop: it is // how a job says "not this time". The tick must survive it. - const outcome = await failJob(db, job, error); - if (outcome === 'dead') dead += 1; - else retried += 1; + const outcome = await failJob(db, job, error, isOutage); + + if (outcome === 'dead') { + dead += 1; + } else if (outcome === 'deferred') { + deferred += 1; + // An outage is global by definition, so every remaining job would fail + // the same way. Stopping now spares the rest of the tick and, more to + // the point, stops us asking a provider that has already said no. + break; + } else { + retried += 1; + } } } - return { processed, succeeded, retried, dead, reclaimed }; + return { processed, succeeded, retried, dead, deferred, reclaimed }; } export interface BatchItem {