Stop an outage from spending a job's last attempt

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<void> {
 
   // 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',
     );
   }
 

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';
 

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);
+  });
+});

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<void> {
   });
 }
 
-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<FailOutcome> {
+export async function failJob(
+  db: Client,
+  job: QueuedJob,
+  error: unknown,
+  isOutage?: OutageCheck,
+): Promise<FailOutcome> {
   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<DrainSummary> {
+  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 {