diff --git a/apps/web/src/lib/server/jobs/JOBS.md b/apps/web/src/lib/server/jobs/JOBS.md index 1d86a1fae..13d3fd421 100644 --- a/apps/web/src/lib/server/jobs/JOBS.md +++ b/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 diff --git a/apps/web/src/lib/server/jobs/__tests__/job-queue.test.ts b/apps/web/src/lib/server/jobs/__tests__/job-queue.test.ts index a7d2bc513..9ce140004 100644 --- a/apps/web/src/lib/server/jobs/__tests__/job-queue.test.ts +++ b/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', () => { diff --git a/apps/web/src/lib/server/jobs/__tests__/runner.test.ts b/apps/web/src/lib/server/jobs/__tests__/runner.test.ts index c0342e5bf..70bf0594c 100644 --- a/apps/web/src/lib/server/jobs/__tests__/runner.test.ts +++ b/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', () => { diff --git a/apps/web/src/lib/server/jobs/__tests__/worker.test.ts b/apps/web/src/lib/server/jobs/__tests__/worker.test.ts index 072450abd..0af32a9a5 100644 --- a/apps/web/src/lib/server/jobs/__tests__/worker.test.ts +++ b/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,7 +72,8 @@ 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, }), @@ -76,7 +81,10 @@ async function bootJobWorker(plan: ClaimPlan, dormancy?: DormancyPlan) { 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) => ({ diff --git a/apps/web/src/lib/server/jobs/job-queue.ts b/apps/web/src/lib/server/jobs/job-queue.ts index 0a320702d..225c9f892 100644 --- a/apps/web/src/lib/server/jobs/job-queue.ts +++ b/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, diff --git a/apps/web/src/lib/server/jobs/runner.ts b/apps/web/src/lib/server/jobs/runner.ts index f81e59628..579327bde 100644 --- a/apps/web/src/lib/server/jobs/runner.ts +++ b/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 { +/** + * 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 { 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 } } diff --git a/apps/web/src/lib/server/jobs/worker.ts b/apps/web/src/lib/server/jobs/worker.ts index 56726df36..7492546f3 100644 --- a/apps/web/src/lib/server/jobs/worker.ts +++ b/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,