perf(jobs): prune terminal rows on their own hourly clock, index-usable bound

apps/web/src/lib/server/jobs/JOBS.md

@@ -255,12 +255,13 @@ Read from `process.env` directly rather than through the zod config, matching
 `process-role.ts`: these must work in any context, including a worker process that
 has not loaded the full application config.
 
-| Variable               | Default | Meaning                                                             |
-| ---------------------- | ------- | ------------------------------------------------------------------- |
-| `JOB_POLL_INTERVAL_MS` | 1000    | How often each workspace loop claims work                           |
-| `JOB_BATCH_SIZE`       | 5       | Jobs claimed per drain pass                                         |
-| `JOB_REAP_INTERVAL_MS` | 15000   | How often expired leases are adjudicated                            |
-| `JOB_RETENTION_MS`     | 7 days  | How long terminal rows are kept. Must exceed any live cron slot key |
+| Variable                | Default | Meaning                                                                                       |
+| ----------------------- | ------- | --------------------------------------------------------------------------------------------- |
+| `JOB_POLL_INTERVAL_MS`  | 1000    | How often each workspace loop claims work                                                     |
+| `JOB_BATCH_SIZE`        | 5       | Jobs claimed per drain pass                                                                   |
+| `JOB_REAP_INTERVAL_MS`  | 15000   | How often expired leases are adjudicated                                                      |
+| `JOB_PRUNE_INTERVAL_MS` | 1 hour  | How often terminal rows past retention are dropped (a per-workspace table scan; keep it slow) |
+| `JOB_RETENTION_MS`      | 7 days  | How long terminal rows are kept. Must exceed any live cron slot key                           |
 
 ### Worker job logs
 

apps/web/src/lib/server/jobs/__tests__/job-queue.test.ts

@@ -689,6 +689,27 @@ describe('per-queue retention', () => {
     const rows = await rowsFor(q)
     expect(rows.map((r) => r.status)).toEqual(['failed'])
   })
+
+  it('keeps a row past the shortest window but inside its own longer one', async () => {
+    // The index-usable floor is the shortest retention anywhere; it must only
+    // ever narrow the scan, never decide the outcome for a queue kept longer.
+    const short = queue('retention-floor-short')
+    const long = queue('retention-floor-long')
+    for (const q of [short, long]) {
+      await enqueueJob({ queue: q, dedupeKey: 'done' })
+      const [job] = await claimJobs({ specs: [{ queue: q, limit: 1, leaseMs: LEASE }] })
+      await completeJob(job)
+    }
+    await testSql()`
+      UPDATE job_queue SET finished_at = now() - interval '3 days'
+      WHERE queue IN (${short}, ${long})
+    `
+    // Default one day is the floor; the long queue keeps successes thirty.
+    await pruneTerminalJobs(86_400_000, { [long]: { succeeded: 30 * 86_400_000 } })
+
+    expect(await rowsFor(short)).toHaveLength(0)
+    expect((await rowsFor(long)).map((r) => r.status)).toEqual(['succeeded'])
+  })
 })
 
 describe('transactional enqueue', () => {

apps/web/src/lib/server/jobs/__tests__/runner.test.ts

@@ -522,6 +522,25 @@ describe('maintenance', () => {
     expect(rows[0].status).toBe('failed')
     expect(rows[0].last_error).toMatch(/no attempts remaining/)
   })
+
+  it('leaves aged terminal rows alone when the prune is not due', async () => {
+    const q = queue('maintenance-no-prune')
+    __setJobDefinitionsForTests([{ name: q, maxAttempts: 1, handler: async () => async () => {} }])
+
+    await enqueueJob({ queue: q, dedupeKey: 'stranded', maxAttempts: 1 })
+    await claimJobs({ specs: [{ queue: q, limit: 1, leaseMs: 30_000 }] })
+    await expireLease(q)
+    await enqueueJob({ queue: q, dedupeKey: 'ancient', maxAttempts: 1 })
+    await testSql()`
+      UPDATE job_queue SET status = 'succeeded', finished_at = now() - interval '400 days'
+      WHERE queue = ${q} AND dedupe_key = 'ancient'
+    `
+
+    const result = await runMaintenanceTick(CONFIG, { prune: false })
+    expect(result.terminated).toBeGreaterThanOrEqual(1)
+    expect(result.pruned).toBe(0)
+    expect((await rowsFor(q)).map((r) => r.status).sort()).toEqual(['failed', 'succeeded'])
+  })
 })
 
 describe('the bounded pool', () => {

apps/web/src/lib/server/jobs/__tests__/worker.test.ts

@@ -5,6 +5,8 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 
 const POLL_MS = 50
+const REAP_MS = POLL_MS * 4
+const PRUNE_MS = REAP_MS * 10
 const WORKSPACE_KEY = 'job_loop_ws'
 
 const workspace = {
@@ -18,6 +20,8 @@ const workspace = {
 
 interface ClaimPlan {
   claimed: number
+  /** `prune` flag of every `runMaintenanceTick` call, in order, when provided. */
+  maintenance?: boolean[]
 }
 
 interface DormancyPlan {
@@ -68,15 +72,19 @@ async function bootJobWorker(plan: ClaimPlan, dormancy?: DormancyPlan) {
     runnerConfig: () => ({
       pollIntervalMs: POLL_MS,
       batchSize: 5,
-      reapIntervalMs: 15_000,
+      reapIntervalMs: REAP_MS,
+      pruneIntervalMs: PRUNE_MS,
       retentionMs: 7 * 24 * 60 * 60 * 1000,
       maxConcurrency: 4,
     }),
     createJobPool: () => ({}),
     poolSize: () => 0,
     createScheduleState: () => ({}),
     runScheduleTick: async () => ({ enqueued: 0, attempted: 0, nextSlotAt: null }),
-    runMaintenanceTick: async () => ({ requeued: 0, terminated: 0 }),
+    runMaintenanceTick: async (_config: unknown, opts: { prune?: boolean } = {}) => {
+      plan.maintenance?.push(opts.prune !== false)
+      return { requeued: 0, terminated: 0, pruned: 0 }
+    },
     dispatchPass: async () => ({ claimed: plan.claimed, saturated: true }),
     runJob: async () => 'succeeded',
     awaitPool: async () => {},
@@ -135,6 +143,24 @@ describe('pooled job worker', () => {
     expect(handle.status().claimed).toBeGreaterThanOrEqual(2)
   })
 
+  it('reaps leases on the reap clock and prunes only on the slower prune clock', async () => {
+    // Retention is measured in days; pruning on every reap tick scans every
+    // workspace's terminal rows for nothing. Only the first tick of each prune
+    // window carries the flag.
+    vi.useFakeTimers()
+    const plan: ClaimPlan = { claimed: 0, maintenance: [] }
+    handle = await bootJobWorker(plan)
+    expect(plan.maintenance).toEqual([true])
+
+    await vi.advanceTimersByTimeAsync(PRUNE_MS - REAP_MS)
+    const withinWindow = plan.maintenance!.slice(1)
+    expect(withinWindow.length).toBeGreaterThanOrEqual(2)
+    expect(withinWindow.every((prune) => prune === false)).toBe(true)
+
+    await vi.advanceTimersByTimeAsync(REAP_MS * 2)
+    expect(plan.maintenance!.filter(Boolean)).toHaveLength(2)
+  })
+
   describe('dormancy', () => {
     const HOUR = 3_600_000
     const idle = (key: string) => ({

apps/web/src/lib/server/jobs/job-queue.ts

@@ -658,6 +658,11 @@ export async function pruneTerminalJobs(
   // stays answerable long after the successful traffic has been discarded. One
   // fleet-wide window either bloats the highest-volume queue's table or throws
   // away the diagnostic history the low-volume ones were keeping on purpose.
+  const overrideSecs = Object.values(perQueueMs).flatMap((byStatus) =>
+    Object.values(byStatus)
+      .filter((ms): ms is number => typeof ms === 'number')
+      .map((ms) => ms / 1000)
+  )
   const overrides = JSON.stringify(
     Object.fromEntries(
       Object.entries(perQueueMs).map(([queue, byStatus]) => [
@@ -670,9 +675,17 @@ export async function pruneTerminalJobs(
       ])
     )
   )
+  // The per-row retention below depends on the row's own `queue` and `status`,
+  // so the planner cannot use `job_queue_terminal_idx` for it and would scan
+  // every terminal row on every prune. The shortest retention anywhere is a
+  // plain constant bound that every qualifying row also satisfies, so it goes
+  // first as the index-usable filter; the precise test then runs only on rows
+  // already past that floor.
+  const floorSecs = Math.min(olderThanMs / 1000, ...overrideSecs)
   const result = await db.execute(sql`
     DELETE FROM job_queue
     WHERE status IN ('succeeded', 'failed')
+      AND finished_at < now() - make_interval(secs => ${floorSecs})
       AND finished_at < now() - make_interval(
         secs => COALESCE(
           (${overrides}::jsonb -> queue ->> status)::numeric,

apps/web/src/lib/server/jobs/runner.ts

@@ -65,6 +65,13 @@ export interface RunnerConfig {
   batchSize: number
   /** How often expired leases are reclaimed. */
   reapIntervalMs: number
+  /**
+   * How often terminal rows past retention are pruned. Retention is measured
+   * in days, so this is deliberately much slower than `reapIntervalMs`: a lost
+   * lease has to be noticed quickly, an aged row does not, and the prune is a
+   * table scan on every workspace in a pooled fleet.
+   */
+  pruneIntervalMs: number
   /** How long terminal rows are kept. Must exceed any live cron slot key. */
   retentionMs: number
   /**
@@ -89,6 +96,7 @@ export function runnerConfig(): RunnerConfig {
     pollIntervalMs: envInt('JOB_POLL_INTERVAL_MS', 1_000, 50, 600_000),
     batchSize: envInt('JOB_BATCH_SIZE', 5, 1, 100),
     reapIntervalMs: envInt('JOB_REAP_INTERVAL_MS', 15_000, 500, 3_600_000),
+    pruneIntervalMs: envInt('JOB_PRUNE_INTERVAL_MS', 3_600_000, 1_000, 86_400_000),
     retentionMs: envInt('JOB_RETENTION_MS', 7 * 24 * 60 * 60 * 1000, 60_000, 365 * 86_400_000),
     maxConcurrency: envInt('JOB_MAX_CONCURRENCY', totalDeclaredConcurrency(), 1, 512),
   }
@@ -665,9 +673,20 @@ export interface MaintenanceResult extends ReapResult {
   pruned: number
 }
 
-/** Reclaim expired leases, then drop terminal rows past retention. */
-export async function runMaintenanceTick(config: RunnerConfig): Promise<MaintenanceResult> {
+/**
+ * Reclaim expired leases and, when the caller says the prune is due, drop
+ * terminal rows past retention.
+ *
+ * The two run on different clocks (`reapIntervalMs` vs `pruneIntervalMs`); the
+ * loop owns both and tells this function which fired. `prune` defaults to on so
+ * a caller with a single clock keeps the historical "both, every tick" shape.
+ */
+export async function runMaintenanceTick(
+  config: RunnerConfig,
+  opts: { prune?: boolean } = {}
+): Promise<MaintenanceResult> {
   const reaped = await reapExpiredLeases()
-  const pruned = await pruneTerminalJobs(config.retentionMs, retentionOverrides())
+  const pruned =
+    opts.prune === false ? 0 : await pruneTerminalJobs(config.retentionMs, retentionOverrides())
   return { ...reaped, pruned }
 }

apps/web/src/lib/server/jobs/worker.ts

@@ -123,6 +123,7 @@ function startLoop(opts: {
   let waitResolve: (() => void) | null = null
   let nextScheduleAt = 0
   let nextMaintenanceAt = 0
+  let nextPruneAt = 0
   let descriptor: WorkspaceDescriptor | null = opts.workspace
   const schedule = createScheduleState()
   const pool = createJobPool()
@@ -196,10 +197,12 @@ function startLoop(opts: {
             nextScheduleAt = tick.nextSlotAt ? tick.nextSlotAt.getTime() : now + 60_000
           }
           if (now >= nextMaintenanceAt) {
-            const maintenance = await runMaintenanceTick(opts.config)
+            const prune = now >= nextPruneAt
+            const maintenance = await runMaintenanceTick(opts.config, { prune })
             s.requeued += maintenance.requeued
             s.terminated += maintenance.terminated
             nextMaintenanceAt = now + opts.config.reapIntervalMs
+            if (prune) nextPruneAt = now + opts.config.pruneIntervalMs
           }
           return dispatchPass({
             pool,