From 01b09a1f459930cddc99fa54531f3ebd1ab082de 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: Thu, 10 Sep 2026 09:40:35 -0400 Subject: [PATCH 1/3] fix(worker): report boot failures on /health, and stop calling a busy worker stalled Restores the worker /health endpoint deferred by d0c6f99, with the three incident-class defects that deferral named actually fixed rather than re-landed. main's drift guard, start.sh and repair migration are untouched -- the regex classifier from defect 2 is NOT reintroduced. THE LIVENESS MODEL Defect 1 was a healthy worker answering 503 "stalled". poll() stamps lastPollTime and then awaits the jobs it claimed, so the stamp freezes for as long as the work takes. b8a68e3 compared it against STALE_POLL_MS=60s; a 300s job therefore read as wedged and the container probe killed it mid-job. Bounding the poll instead does not work either, and this is worth recording because it is the trap: claimJobsByType awaits each TYPE's batch sequentially, so one poll can legitimately run the SUM of every registered type's timeout -- 930s against this worker's ten types, against a max-plus-grace bound of 360s. Same incident, different arithmetic. So poll duration is not the signal. Worker now publishes overdueJobCount: in-flight jobs past their OWN timeout plus a grace, which needs no scheduling arithmetic and cannot drift out of step with how poll() batches. A poll with nothing claimed is separately bounded by CLAIM_STALL_MS, since that is the one case no job can be blamed for. lastPollCompletedAt measures the gap BETWEEN polls, which is what STALE_POLL_MS was always trying to measure. Also fixed, per review: an explicitly-set but unusable PORT is now fatal rather than defaulted. Number.parseInt reads a numeric prefix, so "3003abc" bound 3003 and "0" bound an OS-assigned ephemeral port -- and the Dockerfile probes ${PORT:-...}, where shell :- substitutes only for unset or EMPTY, so serving on a fallback guarantees a probe that can never connect. A wrong port is unreportable over that port, exactly like EADDRINUSE. DEFECT 3 Number(process.env.X ?? default) collapsed to 0 on a cleared variable: `??` guards undefined and null, and Number('') is 0, not NaN. resolveDurationMs replaces it for SHUTDOWN_WATCHDOG_MS and BOOT_FAILURE_LINGER_MS, and also bounds the top -- setTimeout clamps past 2**31-1 to 1ms, so "effectively never" became "immediately", which is the same bug from the other end. HONESTY failFatally now stops the scheduler and worker before exiting. It was leaving every job claimed in its 5s window at status='PROCESSING', and nothing requeues a stale lockedAt, so those jobs were lost rather than retried -- the hazard the boot-after-shutdown guard exists to prevent, reached by another path. It also reports a post-boot crash as 'crashed', not 'failed': the failed-boot advice points at schema.prisma and the drift guard, so filing a handler's stray rejection under it sends whoever is paged to audit migrations. Config warnings now ride on /health instead of dying in a boot log. The watchdog one prints at boot and bites at the next deploy, potentially weeks later. VERIFICATION turbo build/typecheck/lint/test 10/10, lint 0 errors. apps/worker 22 -> 70 tests, packages/outpost 1196 -> 1208. Root vitest.config.ts (which no turbo task runs) 4 files / 64 passing. Prettier clean on every changed file. Mutation-tested, since vacuous tests are what sank the first attempt: the old max-bound model fails 7; reordering the overdue check fails 1; resolvePort accepting 0 fails 1; lenient parseInt fails 3; dropping the duration upper bound fails 3; crash-as-failed fails 2; a second exit timer fails 1; removing the overdue grace fails 1. Two guards were REMOVED after mutation showed no input could reach them: a poll re-entrancy check (poll() only runs from a timer armed at the end of a poll) and a poll-generation token (its only scenario needs a stop()/start() overlap nothing performs, and it mutually masked with the !running check). reschedule() keeps its !running guard, and the test comment says plainly that removing it leaves the suite green rather than implying coverage. --- apps/worker/src/__tests__/health.test.ts | 639 ++++++++++++++++++ apps/worker/src/health.ts | 445 ++++++++++++ apps/worker/src/index.ts | 430 +++++++++--- .../src/__tests__/worker-concurrency.test.ts | 301 ++++++++- packages/outpost/queue/src/types.ts | 23 + packages/outpost/queue/src/worker.ts | 144 +++- 6 files changed, 1880 insertions(+), 102 deletions(-) create mode 100644 apps/worker/src/__tests__/health.test.ts create mode 100644 apps/worker/src/health.ts diff --git a/apps/worker/src/__tests__/health.test.ts b/apps/worker/src/__tests__/health.test.ts new file mode 100644 index 00000000..a860838c --- /dev/null +++ b/apps/worker/src/__tests__/health.test.ts @@ -0,0 +1,639 @@ +import { describe, it, expect } from 'vitest'; +import type { WorkerHealthStatus } from '@copilotkit/outpost/queue'; +import { + buildHealthResponse, + classifyFatalError, + resolveDurationMs, + resolvePort, + summarizeBootError, + CLAIM_STALL_MS, + STALE_POLL_MS, + type BootState, +} from '../health.js'; + +const NOW = new Date('2026-08-12T22:00:00.000Z').getTime(); + +// Typed as the real contract, so a rename or removal in WorkerHealthStatus fails +// this file instead of leaving it green against a shape /health never serves. +const WORKER_HEALTH: WorkerHealthStatus = { + running: true, + activeJobCount: 2, + activeJobsByType: { AI_RESPONSE: 2 }, + lastPollTime: new Date(NOW - 1_000), + pollStartedAt: null, + lastPollCompletedAt: new Date(NOW - 1_000), + overdueJobCount: 0, + lastJobSettledAt: new Date(NOW - 1_000), + registeredHandlers: ['AI_RESPONSE', 'TRACKER_SYNC'], + upSince: new Date(NOW - 3_600_000), +}; + +describe('buildHealthResponse', () => { + it('reports 200 with the worker snapshot once boot is ready and the worker is polling', () => { + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, WORKER_HEALTH, NOW); + + expect(statusCode).toBe(200); + expect(body).toMatchObject({ status: 'ok', running: true, activeJobCount: 2 }); + }); + + // `status` is the envelope's field. Spreading the snapshot over it would let a + // future WorkerHealthStatus.status redefine "ok" for every probe silently. + it('keeps its own status field even if the snapshot carries one', () => { + const shadowed = { ...WORKER_HEALTH, status: 'degraded' } as unknown as WorkerHealthStatus; + + const { body } = buildHealthResponse({ phase: 'ready' }, shadowed, NOW); + + expect(body.status).toBe('ok'); + }); + + // The regression this pins: the worker used to await the database at module + // scope, above the health server, so a boot failure exited the process before + // anything bound the port. Railway could only say "1/1 replicas never became + // healthy" — indistinguishable from a broken image, and it hid a missing + // SystemConfig table for nine days. A failed boot must now answer, and the + // answer must carry the reason. + it('reports 503 AND the reason when boot failed', () => { + const boot: BootState = { + phase: 'failed', + error: 'P2021: missing database object `public.SystemConfig`', + }; + + const { statusCode, body } = buildHealthResponse(boot, null, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('failed'); + expect(body.error).toContain('SystemConfig'); + }); + + it('reports 503 with a reason while boot is still in progress', () => { + const { statusCode, body } = buildHealthResponse({ phase: 'starting' }, null, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('starting'); + expect(body.error).toBeTruthy(); + }); + + // Fail-fast is retained on purpose: a worker whose sync mappings could not be + // read must never be routed to, because it would write wrong statuses to + // Linear. This pins that a failed boot is not quietly downgraded to healthy. + it('never returns 200 for a failed boot, even with a worker snapshot present', () => { + const boot: BootState = { phase: 'failed', error: 'connection refused' }; + + const { statusCode, body } = buildHealthResponse(boot, WORKER_HEALTH, NOW); + + expect(statusCode).toBe(503); + expect(body.error).toBe('connection refused'); + }); + + // The gap that actually reaches production. Worker.stop() sets running=false + // without exiting the process — including from Worker's OWN signal handlers — + // so gating the 200 on the snapshot's existence alone answered + // 200 {"status":"ok","running":false} for a worker processing nothing. + it('does not report 200 for a worker that has stopped', () => { + const stopped: WorkerHealthStatus = { ...WORKER_HEALTH, running: false, upSince: null }; + + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, stopped, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stopped'); + expect(body.error).toContain('not running'); + }); + + // Worker.poll() catches every error and reschedules, so a poll blocked on a + // hung database call leaves running=true forever with lastPollTime frozen. + it('does not report 200 for a worker whose poll loop has stalled', () => { + const stalled: WorkerHealthStatus = { + ...WORKER_HEALTH, + pollStartedAt: null, + activeJobCount: 0, + lastPollTime: new Date(NOW - STALE_POLL_MS - 1), + lastPollCompletedAt: new Date(NOW - STALE_POLL_MS - 1), + }; + + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, stalled, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stalled'); + }); + + it('does not report 200 for a worker that has never polled', () => { + const neverPolled: WorkerHealthStatus = { + ...WORKER_HEALTH, + lastPollTime: null, + lastPollCompletedAt: null, + }; + + const { statusCode } = buildHealthResponse({ phase: 'ready' }, neverPolled, NOW); + + expect(statusCode).toBe(503); + }); + + // ─── 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. + // + // Both earlier bounds were therefore wrong: `now - lastPollTime > 60s`, and + // its replacement `pollDuration > max(jobTimeouts) + 60s` = 360s. Either one + // reports a healthy worker stalled, and the container probe kills it mid-job + // ~90s later. So poll duration is not the signal at all — the worker reports + // per-job overrun instead, and that needs no scheduling arithmetic. + describe('a long-running poll is busy, not stalled', () => { + const midPoll = (pollAgeMs: number, over: Partial = {}) => ({ + ...WORKER_HEALTH, + pollStartedAt: new Date(NOW - pollAgeMs), + lastPollTime: new Date(NOW - pollAgeMs), + lastPollCompletedAt: new Date(NOW - pollAgeMs - 1_000), + activeJobCount: 1, + ...over, + }); + + // 930_000 is the real sequential worst case for the ten registered types. + 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) => { + const { statusCode, body } = buildHealthResponse( + { phase: 'ready' }, + midPoll(pollAgeMs), + NOW, + ); + + expect(statusCode).toBe(200); + expect(body.status).toBe('busy'); + }, + ); + + // The worker judges each job against ITS OWN timeout, so this is the + // signal that a long poll has stopped being honest work. + it('reports 503 when the worker says a job has outlived its own timeout', () => { + const overdue = midPoll(600_000, { overdueJobCount: 1 }); + + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, overdue, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stalled'); + expect(body.error).toContain('outlived their own timeout'); + }); + + // Ordering test: overdue is checked BEFORE any timing branch, so a + // snapshot whose every timing signal looks healthy still reports 503 when + // the worker says its jobs are not progressing. Whether the worker can + // currently produce fresh poll stamps while saturated is beside the point + // — the health check must not depend on that being impossible. + it('lets overdue jobs override otherwise-healthy timing signals', () => { + const wedgedAtCapacity: WorkerHealthStatus = { + ...WORKER_HEALTH, + activeJobCount: 10, + pollStartedAt: null, + lastPollCompletedAt: new Date(NOW - 500), + lastPollTime: new Date(NOW - 500), + lastJobSettledAt: new Date(NOW - 7_200_000), + overdueJobCount: 10, + }; + + const { statusCode, body } = buildHealthResponse( + { phase: 'ready' }, + wedgedAtCapacity, + NOW, + ); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stalled'); + }); + + // A poll with nothing claimed is doing one bounded thing. Long silence + // there is a blocked claim query, and no job can be blamed for it. + it('reports 503 when a poll runs long with no job in flight', () => { + const claimBlocked = midPoll(CLAIM_STALL_MS + 1, { activeJobCount: 0 }); + + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, claimBlocked, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stalled'); + expect(body.error).toContain('blocked claiming'); + }); + + it('still reports 200 for a short poll with nothing claimed yet', () => { + const justStarted = midPoll(CLAIM_STALL_MS - 1, { activeJobCount: 0 }); + + expect(buildHealthResponse({ phase: 'ready' }, justStarted, NOW).statusCode).toBe(200); + }); + + it('does not let an in-flight poll mask a stopped worker', () => { + const stoppedMidPoll = midPoll(1_000, { running: false }); + + const { statusCode, body } = buildHealthResponse( + { phase: 'ready' }, + stoppedMidPoll, + NOW, + ); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stopped'); + }); + }); + + // ─── the gap BETWEEN polls ───────────────────────────────────────────── + // + // With no poll in flight, the loop should have rescheduled itself within + // pollIntervalMs. A long silence here means it stopped rescheduling, which is + // the genuine wedge the old bound was trying to catch. + it('reports 503 when no poll is in flight and none has completed inside the bound', () => { + const notRescheduling: WorkerHealthStatus = { + ...WORKER_HEALTH, + pollStartedAt: null, + lastPollCompletedAt: new Date(NOW - STALE_POLL_MS - 1), + }; + + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, notRescheduling, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stalled'); + expect(body.error).toContain('rescheduling'); + }); + + it('measures the gap from the completion stamp, not the start stamp', () => { + // A poll that STARTED long ago but completed a second ago is healthy. Under + // the old single-stamp reading this was the killed-mid-job case. + const longPollJustFinished: WorkerHealthStatus = { + ...WORKER_HEALTH, + pollStartedAt: null, + lastPollTime: new Date(NOW - 290_000), + lastPollCompletedAt: new Date(NOW - 1_000), + }; + + expect(buildHealthResponse({ phase: 'ready' }, longPollJustFinished, NOW).statusCode).toBe( + 200, + ); + }); + + // A half-booted worker must not be reported healthy just because the phase + // flag says ready — and the 503 must still explain itself rather than + // answering {"status":"ready","error":null}, which is the reasonless body + // this endpoint exists to eliminate. + it('reports 503 with a reason when the phase is ready but no worker exists', () => { + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, null, NOW); + + expect(statusCode).toBe(503); + expect(body.status).toBe('no-worker'); + expect(body.error).toContain('boot sequence'); + }); + + it('serves a body that survives JSON serialization', () => { + const { body } = buildHealthResponse({ phase: 'ready' }, WORKER_HEALTH, NOW); + + expect(() => JSON.stringify(body)).not.toThrow(); + expect(JSON.parse(JSON.stringify(body))).toMatchObject({ status: 'ok', running: true }); + }); +}); + +// /health is unauthenticated, so whatever lands in boot.error is published. +// Prisma's connectivity errors quote the database host, port and user; its +// schema errors arrive as a multi-line blob whose preamble carries an absolute +// container path and a source code frame. Only the object name may survive. +describe('summarizeBootError', () => { + // The shape Prisma actually throws — not a hand-built single-line message. + const realisticP2021 = Object.assign( + new Error( + 'Invalid `prisma.systemConfig.findUnique()` invocation in\n' + + '/app/packages/outpost/shared/dist/sync/config.js:34:56\n\n' + + ' 31 const existing = await db.systemConfig.findUnique({\n\n' + + 'The table `public.SystemConfig` does not exist in the current database.', + ), + { code: 'P2021' }, + ); + + it('names the missing object without leaking container paths or the code frame', () => { + const summary = summarizeBootError(realisticP2021); + + expect(summary).toContain('P2021'); + expect(summary).toContain('public.SystemConfig'); + expect(summary).not.toContain('/app/'); + expect(summary).not.toContain('findUnique'); + }); + + it('covers P2022 missing-column drift, not just P2021', () => { + const error = Object.assign( + new Error( + 'The column `public.SystemConfig.updatedAt` does not exist in the current database.', + ), + { code: 'P2022' }, + ); + + expect(summarizeBootError(error)).toContain('P2022'); + expect(summarizeBootError(error)).toContain('SystemConfig.updatedAt'); + }); + + it('redacts the host and user out of a connectivity error', () => { + const error = Object.assign( + new Error("Can't reach database server at `db.internal.railway.app:5432`"), + { code: 'P1001' }, + ); + + const summary = summarizeBootError(error); + + expect(summary).toContain('P1001'); + expect(summary).not.toContain('db.internal.railway.app'); + expect(summary).not.toContain('5432'); + }); + + // PrismaClientInitializationError carries `errorCode`, not `code` — the real + // shape observed from a live boot against an unreachable database. + it('reads errorCode as well as code', () => { + const error = Object.assign( + new Error('Timed out fetching a new connection from the pool'), + { + errorCode: 'P2024', + }, + ); + + expect(summarizeBootError(error)).toContain('P2024'); + }); + + it('falls back to the error class when no code is present at all', () => { + const error = new Error("Can't reach database server at `127.0.0.1:59999`"); + error.name = 'PrismaClientInitializationError'; + + const summary = summarizeBootError(error); + + expect(summary).toContain('PrismaClientInitializationError'); + expect(summary).not.toContain('127.0.0.1'); + }); + + it('redacts credentials out of a plain error', () => { + const summary = summarizeBootError(new Error('postgres://user:hunter2@host/db refused')); + + expect(summary).not.toContain('hunter2'); + }); + + it('handles thrown non-Error values', () => { + expect(summarizeBootError('boom')).toBeTruthy(); + expect(summarizeBootError(null)).toBeTruthy(); + expect(summarizeBootError(undefined)).toBeTruthy(); + }); + + // A plain object carrying a safe code is the one case the echo branch exists + // for; String(obj) would render "[object Object]". + it('reads .message off a non-Error object carrying a safe code', () => { + const summary = summarizeBootError({ + code: 'P2021', + message: 'The table `public.SystemConfig` does not exist in the current database.', + }); + + expect(summary).toContain('public.SystemConfig'); + expect(summary).not.toContain('[object Object]'); + }); + + it('bounds the length of anything it serves', () => { + const error = Object.assign( + new Error(`The table \`${'x'.repeat(5_000)}\` does not exist.`), + { + code: 'P2021', + }, + ); + + expect(summarizeBootError(error).length).toBeLessThanOrEqual(200); + }); +}); + +// An invalid port used to reach server.listen() as NaN, which throws +// ERR_SOCKET_BAD_PORT synchronously at module scope — killing the process before +// anything bound, the exact opaque failure the health server exists to prevent. +describe('resolvePort', () => { + it('prefers PORT, then HEALTH_PORT, then the default', () => { + expect(resolvePort({ PORT: '8080', HEALTH_PORT: '3005' })).toMatchObject({ + port: 8080, + source: 'PORT', + }); + expect(resolvePort({ HEALTH_PORT: '3005' })).toMatchObject({ + port: 3005, + source: 'HEALTH_PORT', + }); + expect(resolvePort({})).toMatchObject({ port: 3003, source: 'default' }); + }); + + // The reported trigger: `??` only falls through on null/undefined, so a + // cleared platform variable arrives as '' and parses to NaN. + it('falls back with a warning on an empty PORT rather than yielding NaN', () => { + const resolved = resolvePort({ PORT: '', HEALTH_PORT: '3005' }); + + expect(resolved.port).toBe(3005); + expect(resolved.source).toBe('HEALTH_PORT'); + }); + + it('falls back with a warning on a non-numeric or out-of-range PORT', () => { + for (const bad of ['tcp://host:5432', 'abc', '70000', '-1']) { + const resolved = resolvePort({ PORT: bad }); + + expect(resolved.port).toBe(3003); + expect(resolved.warning).toContain(bad); + expect(Number.isInteger(resolved.port)).toBe(true); + } + }); +}); + +// The bug class this resolver exists for is the same one that produced the +// SHADOW_MODE fail-open: `??` guards undefined and null, and a platform variable +// that has been CLEARED is neither — it is the empty string. `Number('')` is 0, +// not NaN, so the coercion succeeds and silently disables whatever the duration +// was guarding. +describe('resolveDurationMs', () => { + it('falls back when the variable is unset', () => { + expect(resolveDurationMs(undefined, 'X_MS', 1_000)).toEqual({ ms: 1_000, warning: null }); + }); + + it.each(['', ' '])( + 'falls back on a cleared variable (%j) instead of collapsing to 0', + (raw) => { + const { ms, warning } = resolveDurationMs(raw, 'SHUTDOWN_WATCHDOG_MS', 330_000); + + expect(ms).toBe(330_000); + // Nothing to warn about: a cleared variable is indistinguishable from an + // absent one, and both are ordinary. + expect(warning).toBeNull(); + }, + ); + + it('reads a valid value', () => { + expect(resolveDurationMs('5000', 'X_MS', 1_000).ms).toBe(5_000); + }); + + it.each(['nonsense', '0', '-1', 'NaN', 'Infinity'])( + 'falls back on the unusable value %j and says so', + (raw) => { + const { ms, warning } = resolveDurationMs(raw, 'BOOT_FAILURE_LINGER_MS', 120_000); + + expect(ms).toBe(120_000); + expect(warning).toContain('BOOT_FAILURE_LINGER_MS'); + expect(warning).toContain(raw); + }, + ); + + // A zero-length watchdog fires immediately, forcing exit(1) on every SIGTERM + // mid-drain — the concrete incident behind the '0' case above. + // setTimeout clamps anything above 2**31-1 to 1ms, so an operator reaching + // for "effectively never" gets the exact opposite: a watchdog that fires + // immediately and forces exit(1) on every SIGTERM mid-drain, or a linger + // window that ends before it publishes its reason. The lower bound alone + // does not catch this. + it.each(['2147483648', '9999999999', '1e12'])( + 'falls back on %j, which setTimeout would clamp to 1ms', + (raw) => { + const { ms, warning } = resolveDurationMs(raw, 'SHUTDOWN_WATCHDOG_MS', 330_000); + + expect(ms).toBe(330_000); + expect(warning).toContain('SHUTDOWN_WATCHDOG_MS'); + }, + ); + + it('accepts the largest value setTimeout honours', () => { + expect(resolveDurationMs('2147483647', 'X_MS', 500).ms).toBe(2_147_483_647); + }); + + it('never returns 0, whatever the input', () => { + for (const raw of ['0', '-5', '', 'x', undefined]) { + expect(resolveDurationMs(raw, 'X_MS', 500).ms).toBeGreaterThan(0); + } + }); +}); + +// A crash after boot is not a failed boot. The failed-boot log advice points at +// schema.prisma and the drift guard, so reporting a handler's stray rejection +// under 'failed' sends whoever is paged to audit migrations. +describe('buildHealthResponse — crashed vs failed', () => { + it('distinguishes a post-boot crash from a boot failure', () => { + const crashed = buildHealthResponse( + { phase: 'crashed', error: 'UNHANDLED REJECTION: Error' }, + null, + NOW, + ); + + expect(crashed.statusCode).toBe(503); + expect(crashed.body.status).toBe('crashed'); + + const failed = buildHealthResponse( + { phase: 'failed', error: 'P2021: missing database object' }, + null, + NOW, + ); + + expect(failed.body.status).toBe('failed'); + }); +}); + +// Config warnings are logged once at boot and then never mentioned again, which +// is precisely where this module exists to stop leaving diagnoses. The watchdog +// one is the sharpest: it prints at boot and bites at the next deploy. +describe('buildHealthResponse — config warnings on the probe', () => { + const warnings = ['invalid SHUTDOWN_WATCHDOG_MS="" (expected 1-2147483647 ms)']; + + it('publishes warnings on a healthy response', () => { + const { statusCode, body } = buildHealthResponse( + { phase: 'ready' }, + WORKER_HEALTH, + NOW, + warnings, + ); + + expect(statusCode).toBe(200); + expect(body.configWarnings).toEqual(warnings); + }); + + it('publishes warnings on a failed boot too', () => { + const { body } = buildHealthResponse( + { phase: 'failed', error: 'boom' }, + null, + NOW, + warnings, + ); + + expect(body.configWarnings).toEqual(warnings); + }); + + it('omits the key entirely when there is nothing to report', () => { + const { body } = buildHealthResponse({ phase: 'ready' }, WORKER_HEALTH, NOW); + + expect(body).not.toHaveProperty('configWarnings'); + }); +}); + +describe('resolvePort — values that used to pass silently', () => { + // Number.parseInt reads a numeric PREFIX, so each of these bound a port the + // operator never asked for, with warning: null. + it.each([ + ['0', 'an OS-assigned ephemeral port nothing probes'], + ['3003abc', 'a truncated parse'], + ['80.9', 'a truncated parse that needs root'], + ['1e4', 'a truncated parse of 1'], + ['70000', 'out of range'], + ['-1', 'out of range'], + ['abc', 'not a number'], + ['tcp://host:5432', 'a whole URL'], + ])('rejects PORT=%j (%s) and marks it fatal', (raw) => { + const { port, source, warning, fatal } = resolvePort({ PORT: raw }); + + expect(port).toBe(3003); + expect(source).toBe('default'); + expect(warning).toContain(raw); + // Falling back is worse than failing: the container healthcheck probes + // the value as given, so nothing would ever reach the port being served. + expect(fatal).toBe(true); + }); + + it.each(['1', '3003', '65535', ' 3005 '])('accepts the valid port %j', (raw) => { + const { port, warning, fatal } = resolvePort({ PORT: raw }); + + expect(port).toBe(Number(raw.trim())); + expect(warning).toBeNull(); + expect(fatal).toBe(false); + }); + + it('is not fatal when nothing is set', () => { + expect(resolvePort({})).toEqual({ + port: 3003, + source: 'default', + warning: null, + fatal: false, + }); + }); +}); + +describe('classifyFatalError', () => { + it('records a post-boot crash as crashed, not failed', () => { + const { next, shouldExit } = classifyFatalError( + 'UNHANDLED REJECTION', + new Error('handler blew up'), + { phase: 'ready' }, + ); + + // Distinct from 'failed' because the operator advice differs: a failed + // boot means the database does not match schema.prisma, while this means + // a handler threw. Reporting one as the other sends whoever is paged to + // audit migrations for a bug in an AI handler. + expect(next?.phase).toBe('crashed'); + expect(shouldExit).toBe(true); + }); + + it('classifies a crash during boot as crashed too', () => { + const { next } = classifyFatalError('UNCAUGHT EXCEPTION', new Error('x'), { + phase: 'starting', + }); + + expect(next?.phase).toBe('crashed'); + }); + + // A failed boot owns its own 120s exit timer. A second 5s timer scheduled + // here would win and cut the window that publishes the boot reason to 5s. + it('leaves an already-failed boot alone and schedules no second exit', () => { + const { next, shouldExit } = classifyFatalError('UNHANDLED REJECTION', new Error('y'), { + phase: 'failed', + error: 'P2021: missing database object `SystemConfig`', + }); + + expect(next).toBeNull(); + expect(shouldExit).toBe(false); + }); +}); diff --git a/apps/worker/src/health.ts b/apps/worker/src/health.ts new file mode 100644 index 00000000..ddb03de3 --- /dev/null +++ b/apps/worker/src/health.ts @@ -0,0 +1,445 @@ +/** + * Boot state, health payload construction, and port resolution — split out from + * index.ts so they are testable. + * + * index.ts is a top-level-await module with side effects on import (it binds a + * port and starts polling), so its boot behaviour cannot be exercised directly + * from a test. Everything here is pure and pinned by tests: an unbooted, failed, + * stopped or stalled worker must report 503 WITH a reason, and only a worker + * that is actually polling reports 200. + */ + +import type { WorkerHealthStatus } from '@copilotkit/outpost/queue'; + +export type BootPhase = 'starting' | 'ready' | 'failed' | 'crashed'; + +/** + * Discriminated so the failed-without-a-reason state is unrepresentable. A 503 + * carrying `error: null` is the exact signal-quality bug this module exists to + * remove; making it a type error is cheaper than remembering to assign `error` + * before `phase` on every future edit. + */ +export type BootState = + | { phase: 'starting' } + | { phase: 'ready' } + | { phase: 'failed'; error: string } + // Booted, then died. Kept distinct from 'failed' because the operator advice + // differs completely: a failed boot means the database does not match + // schema.prisma, while a crash after boot means a handler threw. Reporting + // the latter as the former sends whoever is paged to audit migrations for a + // bug in an AI handler. + | { phase: 'crashed'; error: string }; + +export interface HealthResponse { + statusCode: number; + body: Record; +} + +/** + * Misconfiguration notices to publish alongside the health verdict. + * + * A rejected PORT or a rejected duration is logged once at boot and then never + * mentioned again, which puts it exactly where this module exists to stop + * putting things: a container log nobody has reason to suspect. The watchdog case + * is the sharpest — it prints at boot but only bites at the next deploy, weeks + * later. A degraded-but-serving worker should be able to say so on the probe. + */ +function configWarnings(warnings: readonly string[]): Record { + return warnings.length > 0 ? { configWarnings: warnings } : {}; +} + +/** + * Prisma error codes whose failure names a schema object rather than a + * connection. P2021 is a missing table, P2022 a missing column — the drift class + * this endpoint exists to surface. Only the object name is echoed, never the + * message (see summarizeBootError). + */ +const SAFE_TO_ECHO_CODES = new Set(['P2021', 'P2022']); + +/** + * How long the poll loop may be idle between polls before it is wedged. + * + * This bounds the gap BETWEEN polls, never the duration of one. Worker.poll() + * reschedules every 1s when idle, so a minute of silence with no poll running + * means the loop is dead. + * + * The earlier version of this module compared `now - lastPollTime` against this + * bound, which was wrong in the dangerous direction: poll() stamps lastPollTime + * and then awaits Promise.allSettled over every job it claimed, and jobTimeouts + * allow a single job 300s. So any long job froze the stamp well past 60s and the + * probe reported a HEALTHY worker as stalled — with the container healthcheck + * killing it mid-job after 90s. "Busy" and "wedged" have to be different + * questions, which is why WorkerHealthStatus now carries pollStartedAt. + */ +export const STALE_POLL_MS = 60_000; + +/** + * How long a poll may run with NOTHING in flight before it is wedged. + * + * A poll with zero active jobs is doing one thing: claiming. That is a bounded + * 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`. + */ +export const CLAIM_STALL_MS = 60_000; + +/** Upper bound on any reason string served to an unauthenticated probe. */ +const MAX_REASON_LENGTH = 200; + +/** Whether this value carries a Prisma error code we can act on. */ +function hasPrismaCode(error: unknown): boolean { + const raw = (error ?? {}) as { code?: unknown; errorCode?: unknown }; + return typeof raw.code === 'string' || typeof raw.errorCode === 'string'; +} + +function messageOf(error: unknown): string { + if (error instanceof Error) return error.message; + if ( + typeof error === 'object' && + error !== null && + typeof (error as { message?: unknown }).message === 'string' + ) { + return (error as { message: string }).message; + } + return String(error); +} + +/** + * Reduce a boot exception to a reason that can be served on /health. + * + * /health is unauthenticated. Prisma's connectivity errors quote the database + * host, port and user (P1001 names host:port, P1000 names the user), and even + * the "safe" schema errors arrive as a multi-line blob whose preamble carries + * an absolute container path and a source code frame: + * + * Invalid `prisma.systemConfig.findUnique()` invocation in + * /app/packages/outpost/shared/dist/sync/config.js:34:56 + * 31 const existing = await db.systemConfig.findUnique({ + * The table `public.SystemConfig` does not exist in the current database. + * + * So nothing is echoed verbatim. For the schema codes the backticked object name + * is lifted out of the final line — that name is the entire diagnostic payload — + * and everything else degrades to error class plus code, with the full text left + * to the logs. + */ +export function summarizeBootError(error: unknown): string { + // Loaders wrap Prisma failures (`failed to load status map: `), + // which strands the code on the cause and would otherwise degrade the whole + // diagnosis to a bare "Error". The missing-table name IS the payload here. + const cause = (error as { cause?: unknown } | null)?.cause; + if (cause !== undefined && cause !== null && !hasPrismaCode(error) && hasPrismaCode(cause)) { + return summarizeBootError(cause); + } + + const raw = (error ?? {}) as { code?: unknown; errorCode?: unknown }; + // PrismaClientKnownRequestError carries `code`; PrismaClientInitializationError + // carries `errorCode` (frequently undefined, hence the class-name fallback). + const code = + typeof raw.code === 'string' + ? raw.code + : typeof raw.errorCode === 'string' + ? raw.errorCode + : null; + + if (code && SAFE_TO_ECHO_CODES.has(code)) { + // Matched against the diagnostic sentence itself rather than "the last + // non-empty line". Prisma does not guarantee the sentence comes last, and + // when a code frame does, `([^`]+)` lifts a backticked token straight + // out of source — publishing whatever happens to be quoted in that line + // to an unauthenticated probe. + const object = /(?:table|column|model)\s+`([^`]+)`/i.exec(messageOf(error))?.[1]; + if (object) { + return truncate( + `${code}: missing database object \`${object}\` — the database does not match schema.prisma`, + ); + } + } + + // Class name and code only. Both are stable, neither quotes the connection. + const label = [error instanceof Error ? error.name : 'Error', code].filter(Boolean).join(' '); + return truncate(`${label} — see the worker logs for the full error`); +} + +function truncate(reason: string): string { + return reason.length <= MAX_REASON_LENGTH + ? reason + : `${reason.slice(0, MAX_REASON_LENGTH - 1)}…`; +} + +/** + * Resolve the health-server port from the environment. + * + * `??` is not enough: an empty or non-numeric PORT (a cleared platform variable, + * or a reference variable that failed to resolve) parses to NaN, and + * `server.listen(NaN)` throws ERR_SOCKET_BAD_PORT synchronously at module scope + * — killing the process before anything binds, which is precisely the opaque + * "replicas never became healthy" failure this whole module exists to prevent. + * An invalid value falls back to the default and says so. + */ +export function resolvePort( + env: { PORT?: string; HEALTH_PORT?: string }, + fallback = 3003, +): { port: number; source: string; warning: string | null; fatal: boolean } { + const candidates: Array<[string, string | undefined]> = [ + ['PORT', env.PORT], + ['HEALTH_PORT', env.HEALTH_PORT], + ]; + + for (const [source, raw] of candidates) { + if (raw === undefined || raw.trim() === '') continue; + + const trimmed = raw.trim(); + // Number.parseInt is lenient in a way that matters here: it reads a + // numeric PREFIX, so "3003abc" becomes 3003, "80.9" becomes 80 and "1e4" + // becomes 1 — each a silent bind to a port the operator did not ask for. + const numeric = /^\d+$/.test(trimmed); + const parsed = numeric ? Number(trimmed) : Number.NaN; + + // 0 is the trap worth naming: server.listen(0) is valid and binds an + // OS-assigned ephemeral port, so the server comes up somewhere nothing + // probes and logs a confident "listening on port 0". + if (numeric && parsed >= 1 && parsed <= 65535) { + return { port: parsed, source, warning: null, fatal: false }; + } + + // Falling back would be worse than failing. The container healthcheck in + // apps/worker/Dockerfile probes ${PORT:-${HEALTH_PORT:-3003}}, and + // shell :- substitutes only for unset or EMPTY — so PORT="abc" leaves + // wget asking for http://127.0.0.1:abc/health while this process serves + // happily on 3003. The probe can never succeed, the container dies in + // ~90s, and /health answered 200 the whole way: the opaque "replicas + // never became healthy" signal this module exists to delete. A wrong + // port is unreportable over that port by construction, so it is fatal + // for the same reason EADDRINUSE is. + return { + port: fallback, + source: 'default', + warning: `invalid ${source}="${raw}" (expected an integer 1-65535)`, + fatal: true, + }; + } + + return { port: fallback, source: 'default', warning: null, fatal: false }; +} + +/** + * Largest delay setTimeout accepts. Anything above it is clamped to 1ms, so an + * operator reaching for "effectively never" gets the opposite: a watchdog that + * fires immediately and forces exit(1) on every SIGTERM mid-drain, or a + * boot-failure linger window that ends before it publishes its reason. The + * lower bound alone does not catch it. + */ +const MAX_TIMER_MS = 2_147_483_647; + +/** + * Resolve a millisecond duration from the environment. + * + * `Number(process.env.X ?? default)` is the trap this replaces, and it fails in + * the worst possible direction: a platform variable that exists but is empty + * defeats `??` (which only guards undefined and null), and `Number('')` is 0 — + * not NaN. So a cleared SHUTDOWN_WATCHDOG_MS silently became a 0ms watchdog that + * fires immediately, forcing exit(1) on every SIGTERM mid-drain, and a cleared + * BOOT_FAILURE_LINGER_MS disabled the linger window that is the whole point of + * staying up to explain a failed boot. Both were written one file away from a + * docblock explaining why `??` is insufficient for exactly this. + * + * Anything unusable falls back to the default and says so, rather than being + * silently coerced into a number that happens to parse. + */ +export function resolveDurationMs( + raw: string | undefined, + name: string, + fallback: number, +): { ms: number; warning: string | null } { + if (raw === undefined || raw.trim() === '') return { ms: fallback, warning: null }; + + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed <= 0 || parsed > MAX_TIMER_MS) { + return { + ms: fallback, + warning: `invalid ${name}="${raw}" (expected 1-${MAX_TIMER_MS} ms), falling back to ${fallback}`, + }; + } + + return { ms: parsed, warning: null }; +} + +/** + * Decide what a post-boot fatal error should do to the published state. + * + * Extracted from index.ts's last-resort handlers because both halves are + * load-bearing and neither is reachable from a test inside a top-level-await + * module with import-time side effects. + * + * `null` for `next` means leave the state alone: a boot that already failed owns + * its own exit timer (BOOT_FAILURE_LINGER_MS), and overwriting the reason — or + * scheduling a second, shorter exit — would cut the window that publishes it + * from 120s to 5s. A failed boot is a likely source of follow-on rejections, so + * this is the common case, not a corner. + */ +export function classifyFatalError( + kind: string, + error: unknown, + current: BootState, +): { next: BootState | null; shouldExit: boolean } { + if (current.phase === 'failed') return { next: null, shouldExit: false }; + + // 'crashed', not 'failed'. The failed-boot advice points at schema.prisma and + // the drift guard, so filing a handler's stray rejection under it sends + // whoever is paged to audit migrations for a bug in an AI handler. + return { + next: { phase: 'crashed', error: `${kind}: ${summarizeBootError(error)}` }, + shouldExit: true, + }; +} + +/** + * Build the /health response. + * + * `workerHealth` is the worker's own snapshot, or null when the worker has not + * been constructed. It is passed rather than read so this stays pure. + * + * Every non-200 answer carries a reason. The value added over simply exiting is + * that body: a probe alone explains the failure. Exiting before binding the port + * is what made a missing SystemConfig table look identical to a broken image for + * nine days. + * + * 200 requires the worker to be *polling*, not merely constructed. Worker.stop() + * sets running=false without exiting the process, and a poll loop that has + * stopped rescheduling itself leaves a worker that processes nothing while + * looking alive — the honesty gap tracked by #138. + * + * The inverse matters just as much: a worker grinding through a 300s job is + * healthy, and saying otherwise gets it killed mid-job. So a running poll answers + * 200 "busy" until it overruns what the worker's own job timeouts can account + * for, and only the gap between polls is measured against STALE_POLL_MS. + */ +export function buildHealthResponse( + boot: BootState, + workerHealth: WorkerHealthStatus | null, + now: number = Date.now(), + warnings: readonly string[] = [], +): HealthResponse { + if (boot.phase === 'failed') { + return { + statusCode: 503, + body: { status: 'failed', error: boot.error, ...configWarnings(warnings) }, + }; + } + + if (boot.phase === 'crashed') { + return { + statusCode: 503, + body: { status: 'crashed', error: boot.error, ...configWarnings(warnings) }, + }; + } + + if (boot.phase === 'starting') { + return { + statusCode: 503, + body: { + status: 'starting', + error: 'boot has not finished', + ...configWarnings(warnings), + }, + }; + } + + if (!workerHealth) { + return { + statusCode: 503, + body: { + status: 'no-worker', + error: 'boot reported ready but no worker was constructed — this is a bug in the boot sequence, not a database problem', + ...configWarnings(warnings), + }, + }; + } + + if (!workerHealth.running) { + return { + statusCode: 503, + body: { + ...workerHealth, + status: 'stopped', + error: 'worker is not running — it was stopped without the process exiting', + ...configWarnings(warnings), + }, + }; + } + + // A job that has outlived its OWN timeout means the timeout machinery failed + // — the handler is orphaned, or the untimed status write after it is hanging + // and the job's `finally` never ran. Either way the worker is not making + // progress, and it can look maximally busy while doing so: with every slot + // held, poll() returns instantly at capacity and keeps refreshing its + // completion stamp once a second. + // + // Checked before the in-flight branch, because this is exactly the state + // that would otherwise answer 200 "busy" indefinitely. + if (workerHealth.overdueJobCount > 0) { + return { + statusCode: 503, + body: { + ...workerHealth, + status: 'stalled', + ...configWarnings(warnings), + error: `${workerHealth.overdueJobCount} in-flight job(s) have outlived their own timeout — the worker is holding slots for work that is not progressing`, + }, + }; + } + + if (workerHealth.pollStartedAt) { + const pollDuration = now - workerHealth.pollStartedAt.getTime(); + + // 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. + if (workerHealth.activeJobCount === 0 && pollDuration > CLAIM_STALL_MS) { + return { + statusCode: 503, + body: { + ...workerHealth, + status: 'stalled', + error: `the poll has run for ${pollDuration}ms with no job in flight — it is blocked claiming, most likely on a hung database call`, + }, + }; + } + + return { + statusCode: 200, + body: { ...workerHealth, status: 'busy', ...configWarnings(warnings) }, + }; + } + + // No poll in flight, so the gap since the last one finished is the honest + // measure of whether the loop is still turning. + const reference = workerHealth.lastPollCompletedAt ?? workerHealth.lastPollTime; + const sincePoll = reference ? now - reference.getTime() : null; + if (sincePoll === null || sincePoll > STALE_POLL_MS) { + return { + statusCode: 503, + body: { + ...workerHealth, + status: 'stalled', + error: `no poll has been in flight and none has completed for ${sincePoll ?? 'any'}ms — the poll loop has stopped rescheduling itself`, + }, + }; + } + + // `status` last on purpose: it is the envelope's own field, and spreading the + // snapshot over it would let a future WorkerHealthStatus.status silently + // redefine what "ok" means to every probe. + return { + statusCode: 200, + body: { ...workerHealth, status: 'ok', ...configWarnings(warnings) }, + }; +} diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 1eb3cb74..1a28a9e6 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -16,6 +16,14 @@ * - JOB_CLEANUP: Periodic cleanup of old jobs and sync events * - GITHUB_REACTION_POLL: Poll GitHub reactions on AI comments (no webhook exists) * - PENDING_RESPONSE_SWEEP: Settle AI responses stranded in PENDING by a dead job + * + * BOOT ORDER: /health starts listening before any database QUERY runs, so a boot + * failure is reported rather than merely fatal. See the boot-state block below. + * Two classes still escape it, both by construction: the `prisma` import below + * constructs a PrismaClient at module scope (it throws for an ungenerated client + * or an unparseable DATABASE_URL), and a failure to bind the port itself cannot + * be reported over the port. Both are handled loudly rather than silently — see + * the health-server error handler and the last-resort handlers at the bottom. */ import http from 'node:http'; @@ -36,104 +44,370 @@ import { handlePendingResponseSweep, } from '@copilotkit/outpost/queue'; import { buildSyncEngine } from './build-sync-engine.js'; +import { + buildHealthResponse, + classifyFatalError, + resolveDurationMs, + resolvePort, + summarizeBootError, + type BootState, +} from './health.js'; -// ─── Build SyncEngine for TRACKER_SYNC handler ──────────────────────────── +// ─── Boot state ─────────────────────────────────────────────────────────── -// BOOT SEMANTICS — deliberate change. This is a top-level await that performs -// three database reads (the persisted status / priority / label mapping configs) -// before this module finishes evaluating. If the database is unreachable at boot -// the import throws, so the process exits BEFORE the health server below starts -// listening: the container crash-loops with no /health at all rather than coming -// up and reporting itself degraded. +// Fail-fast on a bad boot is still the intent: a worker running with silently +// defaulted sync mappings would write wrong statuses to Linear, so it must not +// report itself healthy. What changed is that failing is no longer SILENT. // -// Fail-fast is the intent — a worker running with silently-defaulted mappings is -// worse than one that is visibly down, since TRACKER_SYNC would then write wrong -// statuses to Linear. Railway's restart policy is the retry mechanism. Note this -// interacts with the /health honesty follow-up (#138): once /health reflects -// worker state, a degraded-but-listening mode becomes a real option and this -// decision is worth revisiting. -const syncEngine = await buildSyncEngine(); - -const handleTrackerSync = createTrackerSyncHandler(syncEngine); - -// ─── Create Worker ──────────────────────────────────────────────────────── - -const worker = new Worker({ - maxConcurrency: 10, - pollIntervalMs: 1000, - concurrencyByType: { - [JobType.AI_RESPONSE]: 4, - [JobType.ESCALATION]: 2, - [JobType.SLA_CHECK]: 1, - [JobType.ONBOARDING_DIGEST]: 1, - [JobType.ACCOUNT_SCORING]: 1, - [JobType.HUBSPOT_SYNC]: 1, - [JobType.TRACKER_SYNC]: 1, - [JobType.JOB_CLEANUP]: 1, - [JobType.GITHUB_REACTION_POLL]: 1, - [JobType.PENDING_RESPONSE_SWEEP]: 1, - }, - jobTimeouts: { - [JobType.AI_RESPONSE]: 120_000, // 2 minutes — AI pipeline is slow - [JobType.HUBSPOT_SYNC]: 300_000, // 5 minutes — full sync can be large - [JobType.ACCOUNT_SCORING]: 300_000, // 5 minutes — many accounts - }, -}); - -// ─── Register Handlers ──────────────────────────────────────────────────── +// This used to be a top-level `await buildSyncEngine()` above the health server, +// so any boot-time database problem killed the process before anything bound the +// port. Railway could only report "1/1 replicas never became healthy", which is +// indistinguishable from a broken image. That cost nine days of undiagnosed +// deploy failures when SystemConfig turned out to be missing from the production +// database: every deploy from 2026-08-07 failed with no usable signal. +// +// Now the port binds first and /health answers 503 with the reason while the boot +// is unfinished or failed, so the reason is one probe away instead of buried in +// container logs nobody had reason to suspect. +// +// A failed boot does NOT park here forever. Railway's healthcheckPath gates a NEW +// DEPLOYMENT; it does not continuously probe and restart an already-running +// service, and restartPolicyType="ALWAYS" is a restart-on-exit policy that can +// never fire on a process that never exits. Staying up indefinitely would mean a +// 20-second Postgres failover during an ordinary container restart wedges the +// worker with zero jobs processed until a human notices — strictly worse than the +// crash-loop it replaced. So the reason is published for BOOT_FAILURE_LINGER_MS +// (long enough for the deploy probe and any log scrape to read it) and then the +// process exits non-zero so the restart policy retries. Diagnosable AND +// self-healing; the two were never actually in tension. +let boot: BootState = { phase: 'starting' }; -worker.on(JobType.AI_RESPONSE, handleAiResponse); -worker.on(JobType.ESCALATION, handleEscalation); -worker.on(JobType.SLA_CHECK, handleSlaCheck); -worker.on(JobType.ONBOARDING_DIGEST, handleOnboardingDigest); -worker.on(JobType.ACCOUNT_SCORING, handleAccountScoring); -worker.on(JobType.HUBSPOT_SYNC, handleHubSpotSync); -worker.on(JobType.TRACKER_SYNC, handleTrackerSync); -worker.on(JobType.JOB_CLEANUP, handleJobCleanup); -worker.on(JobType.GITHUB_REACTION_POLL, handleGithubReactionPoll); -worker.on(JobType.PENDING_RESPONSE_SWEEP, handlePendingResponseSweep); +let worker: Worker | null = null; +let scheduler: Scheduler | null = null; -// ─── Start Scheduler ────────────────────────────────────────────────────── +// ─── Health Server ──────────────────────────────────────────────────────── -const scheduler = new Scheduler(); +// Published on /health as well as logged. A boot-time log line is exactly the +// place this module exists to stop leaving diagnoses. +const configWarningList: string[] = []; -// ─── Health Server ──────────────────────────────────────────────────────── +const { + port, + source: portSource, + warning: portWarning, + fatal: portFatal, +} = resolvePort(process.env); +if (portWarning) { + console.error(`[Worker] ${portWarning}`); + configWarningList.push(portWarning); +} -const port = parseInt(process.env.PORT ?? process.env.HEALTH_PORT ?? '3003', 10); +// An explicitly-set but unusable PORT is fatal rather than defaulted. The +// container healthcheck probes ${PORT:-${HEALTH_PORT:-3003}} and shell :- +// substitutes only for unset or EMPTY, so serving on the fallback would leave +// the probe asking for a port nothing listens on — a container that dies in +// ~90s while /health answered 200 the whole way. That is the opaque failure this +// module exists to delete, so a wrong port is fatal for the same reason +// EADDRINUSE is: it cannot be reported over the port it broke. +if (portFatal) { + console.error( + `[Worker] FATAL: refusing to start on a fallback port. ${portWarning} — ` + + `the container healthcheck probes the value as given, so nothing would ever reach /health. ` + + `Fix PORT/HEALTH_PORT, or unset it to accept the default.`, + ); + process.exit(1); +} const healthServer = http.createServer((req, res) => { - if (req.url === '/health') { - const health = worker.healthCheck(); - res.writeHead(200, { 'Content-Type': 'application/json' }); - res.end(JSON.stringify({ status: 'ok', ...health })); - } else { - res.writeHead(404); - res.end('Not Found'); + // A probe must never be able to kill the process it exists to observe: + // worker.healthCheck() and JSON.stringify both run in this callback, and an + // exception in an http listener is an uncaught exception. + try { + const path = + new URL(req.url ?? '/', 'http://localhost').pathname.replace(/\/+$/, '') || '/'; + if (path !== '/health') { + res.writeHead(404, { 'Content-Type': 'text/plain' }); + res.end('Not Found'); + return; + } + + const { statusCode, body } = buildHealthResponse( + boot, + worker ? worker.healthCheck() : null, + Date.now(), + configWarningList, + ); + res.writeHead(statusCode, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify(body)); + } catch (error) { + console.error('[Worker] /health handler threw:', error); + res.writeHead(500, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + status: 'error', + error: 'health handler failed — see the worker logs', + }), + ); } }); -// ─── Start Everything ───────────────────────────────────────────────────── +// listen() reports bind failures asynchronously through 'error'. With no listener +// that is an uncaught exception: the port never binds and the process dies to a +// bare stack trace — the same opaque signal as the original incident, arriving +// through the one step everything else now depends on. It is also the single +// failure that genuinely cannot be reported over /health, so it must be loud in +// the logs and must exit rather than linger pretending to serve. +healthServer.on('error', (error: NodeJS.ErrnoException) => { + console.error( + `[Worker] FATAL: could not bind the health server to port ${port} (${error.code ?? 'unknown'}, from ${portSource}). ` + + `Nothing can report this process's state without it. Check PORT/HEALTH_PORT and whether another process holds the port.`, + error, + ); + process.exit(1); +}); healthServer.listen(port, () => { - console.log(`[Worker] Health server listening on port ${port}`); + console.log( + `[Worker] Health server listening on port ${port} (from ${portSource}, boot: ${boot.phase})`, + ); }); -scheduler.start(); -worker.start(); +// ─── Graceful Shutdown ──────────────────────────────────────────────────── -console.log('[Worker] Worker process started'); +// Registered BEFORE the boot await, not after it. Boot is the slowest thing this +// process does, which is exactly when Railway tears a bad deploy down — and a +// SIGTERM arriving while module evaluation is still suspended would find no +// handler and kill the process outright. +// +// Note there is a SECOND registrar: Worker.start() installs its own SIGTERM / +// SIGINT handlers that call worker.stop() unawaited. Both fire. That is safe +// only because Worker.stop() early-returns on !running and this handler runs +// first, so ordering here is load-bearing — do not move this registration below +// startWorker(). +let shuttingDown = false; -// ─── Graceful Shutdown ──────────────────────────────────────────────────── +// Longer than the largest entry in jobTimeouts below (300s), because +// Worker.stop() waits for in-flight jobs to finish. A watchdog shorter than the +// drain it guards would turn every deploy that lands mid-job into a forced +// exit(1) — guarding the hang while breaking the normal path. +const { ms: SHUTDOWN_WATCHDOG_MS, warning: watchdogWarning } = resolveDurationMs( + process.env.SHUTDOWN_WATCHDOG_MS, + 'SHUTDOWN_WATCHDOG_MS', + 330_000, +); +if (watchdogWarning) { + console.error(`[Worker] ${watchdogWarning}`); + configWarningList.push(watchdogWarning); +} async function shutdown(signal: string): Promise { - console.log(`[Worker] Received ${signal}, shutting down...`); - scheduler.stop(); - await worker.stop(); - healthServer.close(); - await prisma.$disconnect(); - console.log('[Worker] Shutdown complete'); - process.exit(0); + if (shuttingDown) return; + shuttingDown = true; + console.log(`[Worker] Received ${signal} in boot phase '${boot.phase}', shutting down...`); + + // NOT unref'd. The motivating case is a $disconnect() that never settles + // after the server is closed — precisely when no other referenced handle + // remains, so an unref'd timer would let Node exit 0 (reporting a clean stop + // for a shutdown that never completed) and this line would never print. + // Every path below ends in process.exit, so a referenced timer costs nothing. + const watchdog = setTimeout(() => { + console.error( + `[Worker] Shutdown did not finish in ${SHUTDOWN_WATCHDOG_MS}ms, exiting anyway`, + ); + process.exit(1); + }, SHUTDOWN_WATCHDOG_MS); + + try { + // Close the listener FIRST. Worker.stop() blocks until in-flight jobs + // finish (up to 300s), and advertising a healthy /health for the whole + // drain window tells the platform to keep routing to a replica that has + // already committed to dying. + healthServer.close(); + scheduler?.stop(); + await worker?.stop(); + await prisma.$disconnect(); + console.log('[Worker] Shutdown complete'); + process.exit(0); + } catch (error) { + // Observed: signalled mid-boot, $disconnect() rejects while tearing down + // a pool that never filled ("Timed out fetching a new connection from the + // connection pool"). Without this the rejection is unhandled and the + // process dies to a stack trace mid-shutdown instead of reporting a + // failed stop. + console.error('[Worker] Shutdown failed:', error); + process.exit(1); + } } -process.on('SIGTERM', () => shutdown('SIGTERM')); -process.on('SIGINT', () => shutdown('SIGINT')); +// `void` because an unhandled rejection here would be the very failure the catch +// above exists to prevent. +process.on('SIGTERM', () => void shutdown('SIGTERM')); +process.on('SIGINT', () => void shutdown('SIGINT')); + +// ─── Last-Resort Handlers ───────────────────────────────────────────────── + +// The whole design rests on this process staying up to explain itself, and under +// Node's defaults a single unhandled rejection ends it with a bare stack trace — +// back to the undiagnosable behaviour. Worker.poll() is fired unawaited from a +// timer and Worker's own signal handler calls stop() unawaited, so the paths +// exist. Mark the process unhealthy so /health tells the platform to stop routing +// to it, publish the reason, and then exit so the restart policy retries rather +// than leaving a wedged replica behind. +function failFatally(kind: string, error: unknown): void { + console.error(`[Worker] ${kind}:`, error); + + const { next, shouldExit } = classifyFatalError(kind, error, boot); + if (next) boot = next; + + if (!shouldExit || shuttingDown) return; + + // Stop claiming BEFORE exiting. The scheduler's setInterval timers and the + // worker's poll loop both keep running otherwise, so every job claimed + // between here and process.exit is abandoned mid-flight and left + // status='PROCESSING' — and nothing requeues a stale lockedAt, so those jobs + // are lost rather than retried. Same hazard the boot-after-shutdown guard in + // startWorker() exists to prevent, reached by a different path. + scheduler?.stop(); + void worker?.stop().catch((stopError) => { + console.error('[Worker] Failed to stop the worker after a fatal error:', stopError); + }); + + setTimeout(() => process.exit(1), 5_000); +} + +process.on('unhandledRejection', (reason) => failFatally('UNHANDLED REJECTION', reason)); +process.on('uncaughtException', (error) => failFatally('UNCAUGHT EXCEPTION', error)); + +// ─── Boot ───────────────────────────────────────────────────────────────── + +// Everything that can throw at boot lives in here: buildSyncEngine's three +// database reads (the persisted status / priority / label mapping configs), the +// Worker construction, and the scheduler/worker start. Anything that escapes +// leaves boot.phase === 'failed' and the process ALIVE but unhealthy, so the +// reason reaches /health instead of vanishing with the process. +async function startWorker(): Promise { + const syncEngine = await buildSyncEngine(); + const handleTrackerSync = createTrackerSyncHandler(syncEngine); + + const started = new Worker({ + maxConcurrency: 10, + pollIntervalMs: 1000, + concurrencyByType: { + [JobType.AI_RESPONSE]: 4, + [JobType.ESCALATION]: 2, + [JobType.SLA_CHECK]: 1, + [JobType.ONBOARDING_DIGEST]: 1, + [JobType.ACCOUNT_SCORING]: 1, + [JobType.HUBSPOT_SYNC]: 1, + [JobType.TRACKER_SYNC]: 1, + [JobType.JOB_CLEANUP]: 1, + [JobType.GITHUB_REACTION_POLL]: 1, + [JobType.PENDING_RESPONSE_SWEEP]: 1, + }, + jobTimeouts: { + [JobType.AI_RESPONSE]: 120_000, // 2 minutes — AI pipeline is slow + [JobType.HUBSPOT_SYNC]: 300_000, // 5 minutes — full sync can be large + [JobType.ACCOUNT_SCORING]: 300_000, // 5 minutes — many accounts + }, + }); + + // ─── Register Handlers ──────────────────────────────────────────────── + started.on(JobType.AI_RESPONSE, handleAiResponse); + started.on(JobType.ESCALATION, handleEscalation); + started.on(JobType.SLA_CHECK, handleSlaCheck); + started.on(JobType.ONBOARDING_DIGEST, handleOnboardingDigest); + started.on(JobType.ACCOUNT_SCORING, handleAccountScoring); + started.on(JobType.HUBSPOT_SYNC, handleHubSpotSync); + started.on(JobType.TRACKER_SYNC, handleTrackerSync); + started.on(JobType.JOB_CLEANUP, handleJobCleanup); + started.on(JobType.GITHUB_REACTION_POLL, handleGithubReactionPoll); + started.on(JobType.PENDING_RESPONSE_SWEEP, handlePendingResponseSweep); + + // A SIGTERM can land while buildSyncEngine() is still awaiting. shutdown() + // then runs to completion against null handles and heads for process.exit, + // and without this check the boot would resume behind it: Scheduler.start() + // ticks every definition immediately (enqueueing jobs) and Worker.start() + // begins claiming them, so the exit strands freshly-claimed rows in + // PROCESSING. Nothing sequences the two promise chains, so the flag is what + // sequences them. + if (shuttingDown) { + console.log('[Worker] Boot completed after shutdown began — not starting the worker'); + return; + } + + // Published before start() so a probe landing mid-start sees the real worker, + // and so shutdown can stop it if a signal arrives during boot. + const nextScheduler = new Scheduler(); + worker = started; + scheduler = nextScheduler; + + try { + nextScheduler.start(); + started.start(); + } catch (error) { + // Scheduler.start() ticks every definition immediately and installs + // setInterval timers, so a throw between it and worker.start() would + // otherwise leave a process that reports itself failed while still + // enqueueing jobs nothing will consume. Torn down through the locals — + // the module-level handles are narrowed to null at the outer catch. + nextScheduler.stop(); + await started.stop().catch(() => {}); + worker = null; + scheduler = null; + throw error; + } +} + +// NOTHING IS RETHROWN HERE, deliberately. This is a top-level-await entry +// module: an exception escaping module evaluation rejects its evaluation +// promise, which Node reports as an uncaught exception and exits on — a +// listening HTTP server does not keep the process alive. Rethrowing would kill +// the health server before it could answer a single probe and hand Railway the +// same bare "1/1 replicas never became healthy" that hid a missing SystemConfig +// table for nine days. Staying up and answering 503 IS the fix. +// +// Fail-fast is still the intent: a worker whose sync mappings could not be read +// must never be reported healthy, because TRACKER_SYNC would write wrong +// statuses to Linear. Railway fails the deploy on the failing healthcheck and +// keeps the previous replica serving — same outcome, with a reason attached. +// How long a failed boot keeps answering 503 with its reason before exiting so +// restartPolicyType="ALWAYS" retries. Long enough for a deploy healthcheck and a +// log scrape to read it; short enough that a transient database outage recovers +// on its own rather than waiting for a human. +const { ms: BOOT_FAILURE_LINGER_MS, warning: lingerWarning } = resolveDurationMs( + process.env.BOOT_FAILURE_LINGER_MS, + 'BOOT_FAILURE_LINGER_MS', + 120_000, +); +if (lingerWarning) { + console.error(`[Worker] ${lingerWarning}`); + configWarningList.push(lingerWarning); +} + +try { + await startWorker(); + if (!shuttingDown) { + boot = { phase: 'ready' }; + console.log('[Worker] Worker process started'); + } +} catch (error) { + // startWorker() has already torn down anything it managed to start, so by + // here the process holds no timers and no poll loop — only the health server. + // + // The full error goes to the logs only — /health carries the redacted form, + // since Prisma's errors quote the database host, port, user and container paths. + boot = { phase: 'failed', error: summarizeBootError(error) }; + console.error('[Worker] BOOT FAILED:', error); + console.error( + `[Worker] /health on ${port} reports 503 ("${boot.error}") for ${BOOT_FAILURE_LINGER_MS}ms, ` + + `then this process exits 1 so Railway's restart policy retries. ` + + `A missing table or column here means the database does not match schema.prisma — ` + + `check the schema-drift guard in apps/worker/start.sh.`, + ); + setTimeout(() => { + console.error( + '[Worker] Exiting after the boot-failure linger window; restart policy takes over.', + ); + process.exit(1); + }, BOOT_FAILURE_LINGER_MS); +} diff --git a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts index 441c5f50..773c19db 100644 --- a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts +++ b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts @@ -39,13 +39,15 @@ const { Worker } = await import('../worker.js'); // ─── Helpers ──────────────────────────────────────────────────────────────── -function makeJobRow(overrides: Partial<{ - id: string; - type: string; - payload: unknown; - attempts: number; - maxAttempts: number; -}> = {}) { +function makeJobRow( + overrides: Partial<{ + id: string; + type: string; + payload: unknown; + attempts: number; + maxAttempts: number; + }> = {}, +) { return { id: overrides.id ?? 'job-1', type: overrides.type ?? JobType.AI_RESPONSE, @@ -220,11 +222,13 @@ describe('Worker per-type concurrency', () => { // Per-type claims: first call for AI_RESPONSE, second for ESCALATION mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'ai-1', type: JobType.AI_RESPONSE })]) .mockResolvedValueOnce([ - makeJobRow({ id: 'ai-1', type: JobType.AI_RESPONSE }), - ]) - .mockResolvedValueOnce([ - makeJobRow({ id: 'esc-1', type: JobType.ESCALATION, payload: { ticketId: 'tkt-esc', reason: 'test' } }), + makeJobRow({ + id: 'esc-1', + type: JobType.ESCALATION, + payload: { ticketId: 'tkt-esc', reason: 'test' }, + }), ]) .mockResolvedValue([]); @@ -274,3 +278,278 @@ describe('Worker per-type concurrency', () => { expect(callCount).toBeGreaterThanOrEqual(1); }); }); + +// ─── Poll-in-progress instrumentation ────────────────────────────────────── +// +// healthCheck() used to expose only lastPollTime, stamped at the START of poll() +// — and poll() awaits Promise.allSettled over every job it claims. So the stamp +// freezes for as long as the longest job runs, and a consumer reading it alone +// cannot tell a worker grinding through a 5-minute HubSpot sync from one wedged +// on a hung database call. apps/worker/src/health.ts answered 503 "stalled" for +// the former and the container healthcheck killed it mid-job. These fields are +// what let the two be told apart. +describe('Worker health check — poll lifecycle', () => { + let worker: InstanceType; + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(async () => { + if (worker) await worker.stop(); + vi.useRealTimers(); + }); + + it('reports no poll in flight before the worker starts', () => { + worker = new Worker({ pollIntervalMs: 100 }); + + const health: WorkerHealthStatus = worker.healthCheck(); + + expect(health.pollStartedAt).toBeNull(); + expect(health.lastPollCompletedAt).toBeNull(); + expect(health.running).toBe(false); + }); + + it('exposes an in-flight poll while a job is still running, and clears it after', async () => { + worker = new Worker({ pollIntervalMs: 100, maxConcurrency: 1 }); + + // A handler that does not settle until we let it, standing in for a long + // job. poll() cannot return while this is pending. + let release: (() => void) | undefined; + const jobRunning = new Promise((resolve) => { + release = resolve; + }); + worker.on(JobType.AI_RESPONSE, async () => { + await jobRunning; + return { success: true }; + }); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'slow-1' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + // Mid-job: the poll has not returned, so pollStartedAt is set. This is the + // observation that makes "busy" distinguishable from "stalled". + const midJob = worker.healthCheck(); + expect(midJob.pollStartedAt).toBeInstanceOf(Date); + expect(midJob.activeJobCount).toBe(1); + + release?.(); + await vi.advanceTimersByTimeAsync(0); + + const afterJob = worker.healthCheck(); + expect(afterJob.pollStartedAt).toBeNull(); + expect(afterJob.lastPollCompletedAt).toBeInstanceOf(Date); + }); + + it('clears the in-flight marker when a poll throws', async () => { + worker = new Worker({ pollIntervalMs: 100 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + + // The claim query itself fails, which poll() catches and reschedules. A + // poll that threw has still stopped running: leaving the marker set would + // report a dead loop as permanently busy. + mockPrisma.$queryRaw.mockRejectedValueOnce(new Error('connection reset')); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + expect(worker.healthCheck().pollStartedAt).toBeNull(); + expect(worker.healthCheck().lastPollCompletedAt).toBeInstanceOf(Date); + }); + + it('reports no overdue jobs on a fresh worker', () => { + worker = new Worker({ pollIntervalMs: 100 }); + + const health: WorkerHealthStatus = worker.healthCheck(); + + expect(health.overdueJobCount).toBe(0); + expect(health.lastJobSettledAt).toBeNull(); + }); + + it('stamps lastJobSettledAt when a job finishes', async () => { + worker = new Worker({ pollIntervalMs: 100 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow()]).mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + expect(worker.healthCheck().lastJobSettledAt).toBeInstanceOf(Date); + }); + + // The signal that replaced the poll-duration bound. A job past its OWN + // timeout means the timeout machinery failed — including the case where the + // untimed status write after runWithTimeout hangs, so processJob's `finally` + // never runs and the slot is held forever. + it('counts an in-flight job as overdue once it outlives its own timeout', async () => { + worker = new Worker({ + pollIntervalMs: 100, + maxConcurrency: 1, + jobTimeouts: { [JobType.AI_RESPONSE]: 1_000 }, + }); + + let release: (() => void) | undefined; + const hung = new Promise((resolve) => { + release = resolve; + }); + worker.on(JobType.AI_RESPONSE, async () => { + await hung; + return { success: true }; + }); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'hung-1' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + // Inside its timeout: busy, not overdue. + expect(worker.healthCheck().overdueJobCount).toBe(0); + + // Past timeout (1s) plus the 60s grace. Time is advanced without letting + // the handler settle, which is exactly the orphaned-handler state. + vi.setSystemTime(Date.now() + 62_000); + + expect(worker.healthCheck().overdueJobCount).toBe(1); + + release?.(); + await vi.advanceTimersByTimeAsync(0); + }); + + // Superseded poll chains must not clear the live chain's marker: a worker + // whose only running poll is wedged would otherwise show a fresh completion + // and answer 200 — a guard wrong in the reassuring direction. + it('leaves no pending poll timer after stop(), and does not resurrect one', async () => { + worker = new Worker({ pollIntervalMs: 100 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + mockPrisma.$queryRaw.mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + await worker.stop(); + + const timersAfterStop = vi.getTimerCount(); + // Advancing well past the interval must not start another poll. + await vi.advanceTimersByTimeAsync(1_000); + + expect(vi.getTimerCount()).toBeLessThanOrEqual(timersAfterStop); + expect(worker.healthCheck().running).toBe(false); + }); + + // The grace exists because runWithTimeout bounds only the handler — the + // status write after it is untimed, so a job legitimately overshoots its + // stated timeout a little on a slow database. Without the grace, ordinary + // slowness would be reported as a stall. + it('does not count a job that is past its timeout but inside the grace', async () => { + worker = new Worker({ + pollIntervalMs: 100, + maxConcurrency: 1, + jobTimeouts: { [JobType.AI_RESPONSE]: 1_000 }, + }); + + let release: (() => void) | undefined; + const hung = new Promise((resolve) => { + release = resolve; + }); + worker.on(JobType.AI_RESPONSE, async () => { + await hung; + return { success: true }; + }); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'slow-but-fine' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + // 30s past a 1s timeout, still inside the 60s grace. + vi.setSystemTime(Date.now() + 31_000); + expect(worker.healthCheck().overdueJobCount).toBe(0); + + // Past the grace. + vi.setSystemTime(Date.now() + 31_000); + expect(worker.healthCheck().overdueJobCount).toBe(1); + + release?.(); + await vi.advanceTimersByTimeAsync(0); + }); + + // stop() clears pollTimer, but an in-flight poll resumes after its await and + // would otherwise install a fresh timer behind it — a live handle outliving + // `await worker.stop()`, still claiming jobs. + it('claims nothing more after stop(), however long time advances', async () => { + worker = new Worker({ pollIntervalMs: 100 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + mockPrisma.$queryRaw.mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(250); + await worker.stop(); + + const claimsAtStop = mockPrisma.$queryRaw.mock.calls.length; + await vi.advanceTimersByTimeAsync(5_000); + + expect(mockPrisma.$queryRaw.mock.calls.length).toBe(claimsAtStop); + }); + + // Covers the drain: stop() is called with a job in flight, and the poll + // awaiting that job resumes after stop() has already cleared the timer. + // + // Note on what this does NOT pin: reschedule()'s `!this.running` guard is + // defense-in-depth with the same check at the top of poll(), and its effect + // is not observable from here — after processing, nextPollDelay is 0, so the + // re-armed timer fires on the next flush and poll() returns immediately + // either way. Removing the guard leaves this test green. It is kept because a + // handle queued behind a resolved stop() is worth preventing at the source, + // not because a test distinguishes it. + it('finishes a post-stop poll without leaving the worker running', async () => { + worker = new Worker({ pollIntervalMs: 50, maxConcurrency: 1 }); + + let release: (() => void) | undefined; + const jobRunning = new Promise((resolve) => { + release = resolve; + }); + worker.on(JobType.AI_RESPONSE, async () => { + await jobRunning; + return { success: true }; + }); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'draining' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + // A poll is genuinely in flight, holding the job. + expect(worker.healthCheck().pollStartedAt).toBeInstanceOf(Date); + + // Signal shutdown, then let the job settle so the drain can complete. + const stopping = worker.stop(); + release?.(); + await stopping; + + // Let the resumed poll finish unwinding: completePoll() and reschedule() + // run after stop() has already resolved. + await vi.advanceTimersByTimeAsync(0); + expect(worker.healthCheck().running).toBe(false); + expect(worker.healthCheck().pollStartedAt).toBeNull(); + }); + + // NOTE on the at-capacity branch in poll(): it is not reachable while a poll + // awaits its own jobs, because the poll cannot return to schedule the next + // one until every slot it filled has drained. It is left calling + // completePoll() for correctness rather than because a test can reach it. +}); diff --git a/packages/outpost/queue/src/types.ts b/packages/outpost/queue/src/types.ts index e8503981..e8fa4e5f 100644 --- a/packages/outpost/queue/src/types.ts +++ b/packages/outpost/queue/src/types.ts @@ -172,7 +172,30 @@ export interface WorkerHealthStatus { running: boolean; activeJobCount: number; activeJobsByType: Record; + /** When the most recent poll STARTED. Frozen for the duration of that poll. */ lastPollTime: Date | null; + /** + * When the in-flight poll began, or null when no poll is running. + * + * poll() awaits every job it claims, so `lastPollTime` stops advancing for + * as long as the longest job runs. Reading it alone makes a busy worker + * indistinguishable from a wedged one. Consumers deciding liveness must ask + * whether a poll is in progress before judging staleness. + */ + pollStartedAt: Date | null; + /** When the last poll returned. Only meaningful while `pollStartedAt` is null. */ + lastPollCompletedAt: Date | null; + /** + * 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. + */ + overdueJobCount: number; + /** When a job last settled, whatever the outcome. Null before the first one. */ + lastJobSettledAt: Date | null; registeredHandlers: string[]; upSince: Date | null; } diff --git a/packages/outpost/queue/src/worker.ts b/packages/outpost/queue/src/worker.ts index 6c7738ad..9e0e6cbe 100644 --- a/packages/outpost/queue/src/worker.ts +++ b/packages/outpost/queue/src/worker.ts @@ -37,6 +37,26 @@ export class Worker { /** Track active job counts per type for per-type concurrency enforcement */ private activeJobsByType = new Map(); private lastPollTime: Date | null = null; + /** + * When the in-flight poll began, or null when no poll is running. + * + * `lastPollTime` alone cannot distinguish "busy" from "wedged": poll() + * stamps it and then awaits Promise.allSettled over every claimed job, so a + * 300s job freezes the stamp for 300s on a perfectly healthy worker. A + * health check that reads only the stamp reports such a worker stalled and + * the container probe kills it mid-job. These two fields separate the + * question "is a poll running right now" from "how long since one finished". + */ + private pollStartedAt: Date | null = null; + /** Start time and own timeout of every in-flight job, keyed by job id. */ + private activeJobStarts = new Map< + string, + { type: string; startedAt: Date; timeoutMs: number } + >(); + /** When a job last reached its `finally`, whatever the outcome. */ + private lastJobSettledAt: Date | null = null; + /** When the last poll returned. Only meaningful while pollStartedAt is null. */ + private lastPollCompletedAt: Date | null = null; private upSince: Date | null = null; private shutdownResolve: (() => void) | null = null; private signalHandlers: { signal: string; handler: () => void }[] = []; @@ -45,7 +65,9 @@ export class Worker { this.pollIntervalMs = options?.pollIntervalMs ?? 1000; this.batchSize = options?.batchSize ?? 10; this.maxConcurrency = options?.maxConcurrency ?? 5; - this.concurrencyByType = (options?.concurrencyByType ?? {}) as Partial>; + this.concurrencyByType = (options?.concurrencyByType ?? {}) as Partial< + Record + >; this.jobTimeouts = options?.jobTimeouts ?? {}; this.defaultTimeoutMs = options?.defaultTimeoutMs ?? 30_000; } @@ -80,6 +102,11 @@ export class Worker { this.shuttingDown = true; this.running = false; + // A stopped worker has no poll in flight, whatever the poll that is + // still unwinding thinks. Health checks read `running` first, but leaving + // a marker set here would make the snapshot self-contradictory. + this.pollStartedAt = null; + if (this.pollTimer) { clearTimeout(this.pollTimer); this.pollTimer = null; @@ -89,7 +116,9 @@ export class Worker { // Wait for active jobs to finish if (this.activeJobs.size > 0) { - console.log(`[Queue Worker] Waiting for ${this.activeJobs.size} active jobs to complete...`); + console.log( + `[Queue Worker] Waiting for ${this.activeJobs.size} active jobs to complete...`, + ); await new Promise((resolve) => { this.shutdownResolve = resolve; // Check immediately in case jobs finished between the check and setting the resolver @@ -113,11 +142,51 @@ export class Worker { activeJobCount: this.activeJobs.size, activeJobsByType: Object.fromEntries(this.activeJobsByType), lastPollTime: this.lastPollTime, + pollStartedAt: this.pollStartedAt, + lastPollCompletedAt: this.lastPollCompletedAt, + overdueJobCount: this.overdueJobCount(), + lastJobSettledAt: this.lastJobSettledAt, registeredHandlers: Array.from(this.handlers.keys()), upSince: this.upSince, }; } + /** + * How far past its own timeout an in-flight job may run before it counts as + * overdue. + * + * runWithTimeout bounds only the handler; the status write that follows it is + * untimed (see processJob), so a job legitimately overshoots its timeout by a + * little on a slow database. It does not overshoot by a minute. + */ + private static readonly JOB_OVERRUN_GRACE_MS = 60_000; + + /** + * In-flight jobs that have outlived their own timeout plus the grace. + * + * 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. + * + * Asking whether any single job has outlived its own timeout needs no + * scheduling arithmetic, so it cannot drift out of step with how poll() + * batches. A non-zero count means the timeout machinery itself failed — + * which also catches the case where a job's untimed status write hangs, its + * `finally` never runs, and the poll spins at capacity looking healthy. + */ + private overdueJobCount(now: number = Date.now()): number { + let overdue = 0; + for (const { startedAt, timeoutMs } of this.activeJobStarts.values()) { + if (now - startedAt.getTime() > timeoutMs + Worker.JOB_OVERRUN_GRACE_MS) overdue++; + } + return overdue; + } + private registerSignalHandlers(): void { const handler = () => { console.log('[Queue Worker] Received shutdown signal'); @@ -139,13 +208,15 @@ export class Worker { private async poll(): Promise { if (!this.running) return; + this.pollStartedAt = new Date(); try { - this.lastPollTime = new Date(); + this.lastPollTime = this.pollStartedAt; const availableSlots = this.maxConcurrency - this.activeJobs.size; if (availableSlots <= 0) { // At capacity, wait and retry - this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs); + this.completePoll(); + this.reschedule(this.pollIntervalMs); return; } @@ -162,13 +233,40 @@ export class Worker { // If we processed jobs, poll immediately for more const nextPollDelay = processedCount > 0 ? 0 : this.pollIntervalMs; - this.pollTimer = setTimeout(() => this.poll(), nextPollDelay); + this.completePoll(); + this.reschedule(nextPollDelay); } catch (error) { console.error('[Queue Worker] Poll error:', error); - this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs); + this.completePoll(); + this.reschedule(this.pollIntervalMs); } } + /** + * Mark the in-flight poll finished. Called on every exit path out of poll() + * — including the error path, because a poll that threw has still stopped + * running, and leaving pollStartedAt set would report a dead loop as busy + * forever. + */ + private completePoll(): void { + this.pollStartedAt = null; + this.lastPollCompletedAt = new Date(); + } + + /** + * Re-arm the poll loop, unless the worker has stopped. + * + * shutdown() calls `await worker.stop()` with a job potentially in flight, + * and the poll awaiting that job resumes AFTER stop() has cleared the timer. + * Without this check it installs a fresh timer behind the stop, so + * `await worker.stop()` returns while a live handle is still queued to claim + * more work. + */ + private reschedule(delayMs: number): void { + if (!this.running) return; + this.pollTimer = setTimeout(() => this.poll(), delayMs); + } + /** * Claim jobs respecting per-type concurrency limits. * For each registered job type that has available capacity, claim up to @@ -287,7 +385,15 @@ export class Worker { `; // Process jobs concurrently (each tracked in activeJobs) - const promises = jobs.map((job: { id: string; type: string; payload: unknown; attempts: number; maxAttempts: number }) => this.processJob(job)); + const promises = jobs.map( + (job: { + id: string; + type: string; + payload: unknown; + attempts: number; + maxAttempts: number; + }) => this.processJob(job), + ); await Promise.allSettled(promises); return jobs.length; @@ -301,10 +407,12 @@ export class Worker { maxAttempts: number; }): Promise { this.activeJobs.add(job.id); - this.activeJobsByType.set( - job.type, - (this.activeJobsByType.get(job.type) ?? 0) + 1, - ); + this.activeJobsByType.set(job.type, (this.activeJobsByType.get(job.type) ?? 0) + 1); + this.activeJobStarts.set(job.id, { + type: job.type, + startedAt: new Date(), + timeoutMs: this.jobTimeouts[job.type as JobType] ?? this.defaultTimeoutMs, + }); try { const handler = this.handlers.get(job.type); @@ -349,7 +457,12 @@ export class Worker { }, }); } else { - await this.handleFailure(job.id, attempt, job.maxAttempts, result.error ?? 'Unknown error'); + await this.handleFailure( + job.id, + attempt, + job.maxAttempts, + result.error ?? 'Unknown error', + ); } } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); @@ -357,6 +470,8 @@ export class Worker { } } finally { this.activeJobs.delete(job.id); + this.activeJobStarts.delete(job.id); + this.lastJobSettledAt = new Date(); const currentCount = this.activeJobsByType.get(job.type) ?? 1; if (currentCount <= 1) { this.activeJobsByType.delete(job.type); @@ -374,7 +489,10 @@ export class Worker { private async runWithTimeout(promise: Promise, timeoutMs: number): Promise { let timer: ReturnType; const timeout = new Promise((_resolve, reject) => { - timer = setTimeout(() => reject(new Error(`Job timed out after ${timeoutMs}ms`)), timeoutMs); + timer = setTimeout( + () => reject(new Error(`Job timed out after ${timeoutMs}ms`)), + timeoutMs, + ); }); try { From e13af18a0fef4cdcc61f80ce67ca54c7640c28d6 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: Sun, 13 Sep 2026 22:45:10 -0400 Subject: [PATCH 2/3] fix(worker): see a failing poll, and stop calling a healthy worker stalled Review of the liveness model found two ways it still lies, one in each direction. Both are fixed here, with the tests that would have caught them. A POLL THAT THROWS IS NOT A POLL THAT FOUND NOTHING completePoll() runs on the error path -- deliberately, so a dead loop is not reported as busy forever -- and that stamps lastPollCompletedAt exactly as success does. Nothing recorded that the poll threw. So a worker whose every claim query fails was indistinguishable from an idle one and answered 200 forever. This is reachable and likely: buildSyncEngine reads only SystemConfig, so a drifted Job table, rotated credentials or a refused connection all let boot SUCCEED and leave the worker throwing once a second with every timing signal looking perfect. Fast-failing is the more common Postgres failure by far, and bounding only the HUNG call left it wide open. Worker now publishes pollFailingSince and consecutivePollFailures, and /health answers 503 once the failures outlast STALE_POLL_MS. A window rather than a count, so a failover blip does not flap the probe. THE CLAIM BOUND MEASURED THE WHOLE POLL claimJobsByType awaits each type's batch sequentially, so after a 118s batch of AI_RESPONSE jobs settles, the nine remaining claim queries each run with activeJobCount === 0 against a pollDuration that already carries those 118s. The worker answered 503 "blocked claiming" having just done 118 seconds of successful work -- the reverted defect, moved rather than removed. The clock now runs from the later of "this poll started" and "a job last settled", which is the honest reading of how long a poll has been doing nothing but claim. This also gives lastJobSettledAt its first consumer; it was published and never read. SMALLER, ALL FROM THE SAME REVIEW - resolvePort validates the RAW value. It trimmed, and the container healthcheck probes ${PORT:-...} verbatim -- so PORT=" 3000" made wget request an invalid URL every time while this process bound 3000 and answered 200 to anything that reached it. Trimming made the one input class that produces that failure non-fatal. - resolveDurationMs takes digits only. Number() accepts hex and exponent forms, so SHUTDOWN_WATCHDOG_MS=0x10 was a 16ms watchdog forcing exit(1) on every SIGTERM mid-drain. Same coercion-leniency class as the Number('') === 0 trap the function was written to close; resolvePort already guarded it. - summarizeBootError scopes its match to the diagnostic sentence. Against the whole blob it returns the FIRST backticked token, which on a message carrying a code frame is a source token published to an unauthenticated probe. - The /health handler serializes before committing headers. With writeHead first, res.headersSent is already true and the catch threw ERR_HTTP_HEADERS_SENT -- an uncaught exception in an http listener, a probe killing the process it exists to observe. - Both stalled branches carry configWarnings. Six of eight did; the two that did not were the ones an operator reads while the worker is stalled. - BootPhase deleted. Unreferenced, and the undiscriminated projection of BootState -- anything typed as it can express 'failed' with no reason, the state the union exists to forbid. TESTS 84 in apps/worker (was 70), 21 in the queue's worker suite (was 16). The gap that mattered most had no test at all: removing activeJobStarts.delete(job.id) left every suite green, while entries accumulate so every job the worker has ever run eventually crosses its timeout plus the grace -- overdue climbs without bound and, since overdue is checked first, a healthy worker answers 503 permanently. Covered now on both the success and failure paths. The bounds are pinned as literals too. Fixtures computed as CONSTANT +/- 1 re-derive from whatever the constant says, so CLAIM_STALL_MS = 600_000 shipped green -- the same sizing error as the first attempt, invisible to the suite. Mutation-checked: dropping the poll-failure branch fails 1; reverting the claim clock to whole-poll fails 1; restoring the port trim fails 3; lenient Number() fails 4; removing activeJobStarts.delete fails 2; either bound at 600_000 fails 1-2. STILL NOT COVERED, and worth saying plainly rather than implying otherwise: index.ts's wiring is verified by hand-run only -- failFatally's ordering, startWorker's teardown path, and the boot-after-shutdown guard. The pure decision functions are extracted and tested; the effects they drive are not. --- apps/worker/src/__tests__/health.test.ts | 144 +++++++++++++++++- apps/worker/src/health.ts | 71 +++++++-- apps/worker/src/index.ts | 11 +- .../src/__tests__/worker-concurrency.test.ts | 134 ++++++++++++++++ packages/outpost/queue/src/types.ts | 11 ++ packages/outpost/queue/src/worker.ts | 23 +++ 6 files changed, 381 insertions(+), 13 deletions(-) diff --git a/apps/worker/src/__tests__/health.test.ts b/apps/worker/src/__tests__/health.test.ts index a860838c..59eb1747 100644 --- a/apps/worker/src/__tests__/health.test.ts +++ b/apps/worker/src/__tests__/health.test.ts @@ -24,10 +24,27 @@ const WORKER_HEALTH: WorkerHealthStatus = { lastPollCompletedAt: new Date(NOW - 1_000), overdueJobCount: 0, lastJobSettledAt: new Date(NOW - 1_000), + pollFailingSince: null, + consecutivePollFailures: 0, registeredHandlers: ['AI_RESPONSE', 'TRACKER_SYNC'], upSince: new Date(NOW - 3_600_000), }; +// The fixtures below are computed as `CONSTANT ± 1`, so they re-derive from +// whatever the constant says and can never disagree with it. These pin the +// literals, because the sizing is the thing that was wrong twice: 60s against a +// 300s job timeout is what got the first attempt reverted, and a bound ten times +// too lenient would ship green under fixtures alone. +describe('the bounds themselves', () => { + it('sizes the between-polls bound for a loop that reschedules every second', () => { + expect(STALE_POLL_MS).toBe(60_000); + }); + + it('sizes the claim bound for a claim query, not for a poll', () => { + expect(CLAIM_STALL_MS).toBe(60_000); + }); +}); + describe('buildHealthResponse', () => { it('reports 200 with the worker snapshot once boot is ready and the worker is polling', () => { const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, WORKER_HEALTH, NOW); @@ -207,7 +224,12 @@ describe('buildHealthResponse', () => { // A poll with nothing claimed is doing one bounded thing. Long silence // there is a blocked claim query, and no job can be blamed for it. it('reports 503 when a poll runs long with no job in flight', () => { - const claimBlocked = midPoll(CLAIM_STALL_MS + 1, { activeJobCount: 0 }); + const claimBlocked = midPoll(CLAIM_STALL_MS + 1, { + activeJobCount: 0, + // Nothing has settled inside this poll, so the whole poll really + // has been spent claiming. + lastJobSettledAt: new Date(NOW - CLAIM_STALL_MS - 10_000), + }); const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, claimBlocked, NOW); @@ -489,6 +511,20 @@ describe('resolveDurationMs', () => { }, ); + // Number() accepts hex and exponent forms, so `0x10` is a 16ms watchdog that + // forces exit(1) on every SIGTERM mid-drain and `3e2` is 300ms — the same + // coercion-leniency class as the Number('') === 0 trap this function exists + // to close. resolvePort guards it with a digits-only test; so does this. + it.each(['0x10', '3e2', '1_000', '+500', '5.5'])( + 'falls back on the non-decimal form %j', + (raw) => { + const { ms, warning } = resolveDurationMs(raw, 'SHUTDOWN_WATCHDOG_MS', 330_000); + + expect(ms).toBe(330_000); + expect(warning).toContain('SHUTDOWN_WATCHDOG_MS'); + }, + ); + it('accepts the largest value setTimeout honours', () => { expect(resolveDurationMs('2147483647', 'X_MS', 500).ms).toBe(2_147_483_647); }); @@ -583,14 +619,25 @@ describe('resolvePort — values that used to pass silently', () => { expect(fatal).toBe(true); }); - it.each(['1', '3003', '65535', ' 3005 '])('accepts the valid port %j', (raw) => { + it.each(['1', '3003', '65535'])('accepts the valid port %j', (raw) => { const { port, warning, fatal } = resolvePort({ PORT: raw }); - expect(port).toBe(Number(raw.trim())); + expect(port).toBe(Number(raw)); expect(warning).toBeNull(); expect(fatal).toBe(false); }); + // The container healthcheck probes ${PORT:-...} verbatim, so a padded value + // makes wget request an invalid URL every time while this process binds the + // trimmed port and answers 200 to anything that reaches it — the exact + // unprobeable-by-construction failure the fatal branch exists to prevent. + it.each([' 3000', '3000 ', ' 3000 '])('treats the padded port %j as fatal', (raw) => { + const { fatal, warning } = resolvePort({ PORT: raw }); + + expect(fatal).toBe(true); + expect(warning).toContain(raw); + }); + it('is not fatal when nothing is set', () => { expect(resolvePort({})).toEqual({ port: 3003, @@ -637,3 +684,94 @@ describe('classifyFatalError', () => { expect(shouldExit).toBe(false); }); }); + +// A poll that THROWS is not a poll that found nothing. `completePoll()` runs on +// the error path too — deliberately, so a dead loop is not reported as busy +// forever — which stamps `lastPollCompletedAt` and makes a failing poll +// indistinguishable from an idle one. +// +// The failure this closes: the schema drifts and `Job` is missing, or +// credentials rotate, or Postgres refuses connections. buildSyncEngine reads +// only SystemConfig, so boot SUCCEEDS and the worker settles into throwing once +// a second forever. Every timing signal looks perfect and the probe answered 200 +// the whole way. Fast-failing is the more common Postgres failure by far, and it +// was the one hole left open by bounding only the HUNG call. +describe('a poll that keeps failing is not healthy', () => { + const failing = (sinceMs: number, failures: number): WorkerHealthStatus => ({ + ...WORKER_HEALTH, + activeJobCount: 0, + pollStartedAt: null, + // The catch path stamps these exactly as a successful poll would. + lastPollTime: new Date(NOW - 500), + lastPollCompletedAt: new Date(NOW - 500), + pollFailingSince: new Date(NOW - sinceMs), + consecutivePollFailures: failures, + }); + + it('reports 503 once polls have been failing longer than the stale bound', () => { + const { statusCode, body } = buildHealthResponse( + { phase: 'ready' }, + failing(STALE_POLL_MS + 1, 61), + NOW, + ); + + expect(statusCode).toBe(503); + expect(body.status).toBe('stalled'); + expect(String(body.error)).toMatch(/fail/i); + }); + + // A single blip during a failover must not flap the probe, which is why this + // is a window rather than a count. + it('stays 200 for a brief run of failures inside the window', () => { + expect(buildHealthResponse({ phase: 'ready' }, failing(5_000, 5), NOW).statusCode).toBe( + 200, + ); + }); + + it('is healthy again once a poll succeeds and clears the window', () => { + const recovered: WorkerHealthStatus = { + ...failing(STALE_POLL_MS + 1, 61), + pollFailingSince: null, + consecutivePollFailures: 0, + }; + + expect(buildHealthResponse({ phase: 'ready' }, recovered, NOW).statusCode).toBe(200); + }); +}); + +// 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. +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 = { + ...WORKER_HEALTH, + 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. + lastJobSettledAt: new Date(NOW - 200), + }; + + const { statusCode, body } = buildHealthResponse({ phase: 'ready' }, afterLongBatch, NOW); + + expect(statusCode).toBe(200); + expect(body.status).toBe('busy'); + }); + + it('reports 503 once nothing has settled for longer than the bound', () => { + const nothingSettling: WorkerHealthStatus = { + ...WORKER_HEALTH, + activeJobCount: 0, + pollStartedAt: new Date(NOW - 300_000), + lastPollTime: new Date(NOW - 300_000), + lastJobSettledAt: new Date(NOW - CLAIM_STALL_MS - 1), + }; + + expect(buildHealthResponse({ phase: 'ready' }, nothingSettling, NOW).statusCode).toBe(503); + }); +}); diff --git a/apps/worker/src/health.ts b/apps/worker/src/health.ts index ddb03de3..33a10998 100644 --- a/apps/worker/src/health.ts +++ b/apps/worker/src/health.ts @@ -11,8 +11,6 @@ import type { WorkerHealthStatus } from '@copilotkit/outpost/queue'; -export type BootPhase = 'starting' | 'ready' | 'failed' | 'crashed'; - /** * Discriminated so the failed-without-a-reason state is unrepresentable. A 503 * carrying `error: null` is the exact signal-quality bug this module exists to @@ -155,7 +153,16 @@ export function summarizeBootError(error: unknown): string { // when a code frame does, `([^`]+)` lifts a backticked token straight // out of source — publishing whatever happens to be quoted in that line // to an unauthenticated probe. - const object = /(?:table|column|model)\s+`([^`]+)`/i.exec(messageOf(error))?.[1]; + // Scoped to the diagnostic sentence. Matching the whole blob returns the + // FIRST backticked token, which on a message carrying a code frame is a + // source token rather than the object name — published verbatim to an + // unauthenticated probe. + const sentence = messageOf(error) + .split('\n') + .find((line) => /does not exist/i.test(line)); + const object = sentence + ? /(?:table|column|model)\s+`([^`]+)`/i.exec(sentence)?.[1] + : undefined; if (object) { return truncate( `${code}: missing database object \`${object}\` — the database does not match schema.prisma`, @@ -196,7 +203,13 @@ export function resolvePort( for (const [source, raw] of candidates) { if (raw === undefined || raw.trim() === '') continue; - const trimmed = raw.trim(); + // Validated against the RAW value, not a trimmed one. The container + // healthcheck probes `${PORT:-...}` verbatim, so `PORT=" 3000"` makes + // wget request `http://127.0.0.1: 3000/health` — an invalid URL that + // fails every time — while this process binds 3000 and answers 200 to + // anything that reaches it. Trimming here would make the one input class + // that actually produces that failure non-fatal. + const trimmed = raw; // Number.parseInt is lenient in a way that matters here: it reads a // numeric PREFIX, so "3003abc" becomes 3003, "80.9" becomes 80 and "1e4" // becomes 1 — each a silent bind to a port the operator did not ask for. @@ -261,7 +274,12 @@ export function resolveDurationMs( ): { ms: number; warning: string | null } { if (raw === undefined || raw.trim() === '') return { ms: fallback, warning: null }; - const parsed = Number(raw); + // Digits only, for the reason resolvePort gives twelve lines up: `Number()` + // accepts hex and exponent forms, so `SHUTDOWN_WATCHDOG_MS=0x10` is a 16ms + // watchdog that forces exit(1) on every SIGTERM mid-drain, and `3e2` is + // 300ms. Same coercion-leniency class as the `Number('') === 0` trap this + // function was written to close. + const parsed = /^\d+$/.test(raw.trim()) ? Number(raw.trim()) : Number.NaN; if (!Number.isFinite(parsed) || parsed <= 0 || parsed > MAX_TIMER_MS) { return { ms: fallback, @@ -397,19 +415,53 @@ export function buildHealthResponse( }; } - if (workerHealth.pollStartedAt) { - const pollDuration = now - workerHealth.pollStartedAt.getTime(); + // Running, and every poll is throwing. Checked here with the overdue branch + // because both answer the same question — the worker is up and doing no work + // — and because every timing signal below looks perfect in this state: the + // error path stamps `lastPollCompletedAt` exactly as success does. + // + // A window, not a count: a Postgres failover produces a burst of failures and + // then recovers, and flapping the probe through that is worse than waiting. + if (workerHealth.pollFailingSince) { + const failingFor = now - workerHealth.pollFailingSince.getTime(); + if (failingFor > STALE_POLL_MS) { + return { + statusCode: 503, + body: { + ...workerHealth, + status: 'stalled', + ...configWarnings(warnings), + error: `every poll has failed for ${failingFor}ms (${workerHealth.consecutivePollFailures} in a row) — the worker is running but claiming nothing; see the worker logs for the underlying error`, + }, + }; + } + } + 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. - if (workerHealth.activeJobCount === 0 && pollDuration > CLAIM_STALL_MS) { + // 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. + const claimingSince = Math.max( + workerHealth.pollStartedAt.getTime(), + workerHealth.lastJobSettledAt?.getTime() ?? 0, + ); + const claimingFor = now - claimingSince; + if (workerHealth.activeJobCount === 0 && claimingFor > CLAIM_STALL_MS) { return { statusCode: 503, body: { ...workerHealth, status: 'stalled', - error: `the poll has run for ${pollDuration}ms with no job in flight — it is blocked claiming, most likely on a hung database call`, + ...configWarnings(warnings), + error: `no job has been in flight for ${claimingFor}ms of this poll — it is blocked claiming, most likely on a hung database call`, }, }; } @@ -430,6 +482,7 @@ export function buildHealthResponse( body: { ...workerHealth, status: 'stalled', + ...configWarnings(warnings), error: `no poll has been in flight and none has completed for ${sincePoll ?? 'any'}ms — the poll loop has stopped rescheduling itself`, }, }; diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 1a28a9e6..90468329 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -137,10 +137,19 @@ const healthServer = http.createServer((req, res) => { Date.now(), configWarningList, ); + // Serialized BEFORE the headers are committed. With writeHead first, + // res.headersSent is already true, so the catch below would throw + // ERR_HTTP_HEADERS_SENT — an uncaught exception inside an http listener, + // which is a probe killing the process it exists to observe. + const payload = JSON.stringify(body); res.writeHead(statusCode, { 'Content-Type': 'application/json' }); - res.end(JSON.stringify(body)); + res.end(payload); } catch (error) { console.error('[Worker] /health handler threw:', error); + if (res.headersSent) { + res.end(); + return; + } res.writeHead(500, { 'Content-Type': 'application/json' }); res.end( JSON.stringify({ diff --git a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts index 773c19db..05334f8a 100644 --- a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts +++ b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts @@ -553,3 +553,137 @@ describe('Worker health check — poll lifecycle', () => { // one until every slot it filled has drained. It is left calling // completePoll() for correctness rather than because a test can reach it. }); + +// A poll that THROWS is not a poll that found nothing. completePoll() runs on +// the error path too, stamping lastPollCompletedAt exactly as success does, so +// without a failure record a worker whose every claim query fails — drifted +// schema, rotated credentials, refused connections — is indistinguishable from +// an idle one and answers 200 forever. +describe('Worker health check — poll failures', () => { + let worker: InstanceType; + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(async () => { + if (worker) await worker.stop(); + vi.useRealTimers(); + }); + + it('records nothing while polls succeed', async () => { + worker = new Worker({ pollIntervalMs: 50 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + mockPrisma.$queryRaw.mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(120); + + const health: WorkerHealthStatus = worker.healthCheck(); + expect(health.pollFailingSince).toBeNull(); + expect(health.consecutivePollFailures).toBe(0); + }); + + it('opens a failure window when the claim query throws', async () => { + worker = new Worker({ pollIntervalMs: 50 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + mockPrisma.$queryRaw.mockRejectedValue(new Error('relation "Job" does not exist')); + + worker.start(); + await vi.advanceTimersByTimeAsync(120); + + const health = worker.healthCheck(); + expect(health.pollFailingSince).toBeInstanceOf(Date); + expect(health.consecutivePollFailures).toBeGreaterThan(1); + // The tell: the error path stamps this exactly as success would, which is + // why it cannot be the liveness signal on its own. + expect(health.lastPollCompletedAt).toBeInstanceOf(Date); + }); + + it('closes the window as soon as a poll returns normally', async () => { + worker = new Worker({ pollIntervalMs: 50 }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + mockPrisma.$queryRaw + .mockRejectedValueOnce(new Error('connection refused')) + .mockRejectedValueOnce(new Error('connection refused')) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(60); + expect(worker.healthCheck().pollFailingSince).toBeInstanceOf(Date); + + await vi.advanceTimersByTimeAsync(200); + + const recovered = worker.healthCheck(); + expect(recovered.pollFailingSince).toBeNull(); + expect(recovered.consecutivePollFailures).toBe(0); + }); +}); + +// `overdueJobCount` reads `activeJobStarts`, and nothing else prunes that map. +// If a completed job's entry is left behind, every job the worker has EVER run +// eventually crosses its timeout plus the grace, the count climbs without bound, +// and — because overdue is checked before every timing branch — a healthy worker +// answers 503 permanently and the container probe kills it. That is the exact +// revert-worthy failure this endpoint exists to avoid, reached from the other +// direction, plus an unbounded Map. +describe('completed jobs leave no overdue residue', () => { + let worker: InstanceType; + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(async () => { + if (worker) await worker.stop(); + vi.useRealTimers(); + }); + + it('reports no overdue jobs long after finished work would have aged out', async () => { + worker = new Worker({ + pollIntervalMs: 50, + maxConcurrency: 2, + jobTimeouts: { [JobType.AI_RESPONSE]: 1_000 }, + }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: true })); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'done-1' }), makeJobRow({ id: 'done-2' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + expect(worker.healthCheck().activeJobCount).toBe(0); + + // Well past the 1s timeout plus the 60s grace. If the entries survived + // their jobs, both would now be counted overdue. + vi.setSystemTime(Date.now() + 120_000); + + expect(worker.healthCheck().overdueJobCount).toBe(0); + }); + + // The failure path frees the slot too, so it must prune the same way. + it('leaves no residue when a job fails', async () => { + worker = new Worker({ + pollIntervalMs: 50, + maxConcurrency: 1, + jobTimeouts: { [JobType.AI_RESPONSE]: 1_000 }, + }); + worker.on(JobType.AI_RESPONSE, async () => ({ success: false, error: 'nope' })); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockResolvedValueOnce([makeJobRow({ id: 'failed-1' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(0); + vi.setSystemTime(Date.now() + 120_000); + + expect(worker.healthCheck().overdueJobCount).toBe(0); + }); +}); diff --git a/packages/outpost/queue/src/types.ts b/packages/outpost/queue/src/types.ts index e8fa4e5f..140ffe55 100644 --- a/packages/outpost/queue/src/types.ts +++ b/packages/outpost/queue/src/types.ts @@ -196,6 +196,17 @@ export interface WorkerHealthStatus { overdueJobCount: number; /** When a job last settled, whatever the outcome. Null before the first one. */ lastJobSettledAt: Date | null; + /** + * When the current unbroken run of poll failures began, or null if the last + * poll returned normally. + * + * The error path stamps `lastPollCompletedAt` just as success does, so + * without this a worker failing every claim query — drifted schema, rotated + * credentials, refused connections — looks exactly like an idle one. + */ + pollFailingSince: Date | null; + /** Polls that have thrown in a row. 0 once one returns normally. */ + consecutivePollFailures: number; registeredHandlers: string[]; upSince: Date | null; } diff --git a/packages/outpost/queue/src/worker.ts b/packages/outpost/queue/src/worker.ts index 9e0e6cbe..4b148e98 100644 --- a/packages/outpost/queue/src/worker.ts +++ b/packages/outpost/queue/src/worker.ts @@ -55,6 +55,21 @@ export class Worker { >(); /** When a job last reached its `finally`, whatever the outcome. */ private lastJobSettledAt: Date | null = null; + /** + * When the current unbroken run of poll failures began, or null if the last + * poll returned normally. + * + * A poll that THROWS is not a poll that found nothing, but `completePoll()` + * runs on the error path too — deliberately, so a dead loop is not reported + * as busy forever — and that stamps `lastPollCompletedAt` exactly as success + * would. Without this, a worker whose every claim query fails is + * indistinguishable from an idle one, and answers 200 forever. + * + * A window rather than a count, so one blip during a Postgres failover does + * not flap the probe. + */ + private pollFailingSince: Date | null = null; + private consecutivePollFailures = 0; /** When the last poll returned. Only meaningful while pollStartedAt is null. */ private lastPollCompletedAt: Date | null = null; private upSince: Date | null = null; @@ -146,6 +161,8 @@ export class Worker { lastPollCompletedAt: this.lastPollCompletedAt, overdueJobCount: this.overdueJobCount(), lastJobSettledAt: this.lastJobSettledAt, + pollFailingSince: this.pollFailingSince, + consecutivePollFailures: this.consecutivePollFailures, registeredHandlers: Array.from(this.handlers.keys()), upSince: this.upSince, }; @@ -233,10 +250,16 @@ export class Worker { // If we processed jobs, poll immediately for more const nextPollDelay = processedCount > 0 ? 0 : this.pollIntervalMs; + // A poll that returned normally clears the failure window, however + // long it had been running. + this.pollFailingSince = null; + this.consecutivePollFailures = 0; this.completePoll(); this.reschedule(nextPollDelay); } catch (error) { console.error('[Queue Worker] Poll error:', error); + this.consecutivePollFailures++; + this.pollFailingSince ??= new Date(); this.completePoll(); this.reschedule(this.pollIntervalMs); } From 651694a0176b374cfff1a848f80219925ddd29bb 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: Mon, 14 Sep 2026 09:11:46 -0400 Subject: [PATCH 3/3] fix(worker): give a hung boot a deadline, and stop a recovering worker reading dead Review found the one boot outcome nothing watched, plus two narrower windows where the endpoint still answers wrongly. A HUNG BOOT HAD NO DEADLINE BOOT_FAILURE_LINGER_MS arms inside the catch, so it only ever watched a boot that THREW. A boot that HANGS never reaches it: startWorker() awaits buildSyncEngine()'s three database reads and Prisma applies no query timeout, so a Postgres that accepts the connection and then stops answering -- a failover, a saturated pool, a partition that drops packets without resetting -- leaves that await pending forever. The process sits at 503 `starting` processing nothing, and restartPolicyType="ALWAYS" cannot fire because nothing exits. That is precisely the indefinite wedge this module's own header argues is unacceptable ("strictly worse than the crash-loop it replaced"), reached through the one path the linger timer does not watch. BOOT_DEADLINE_MS (180s, well above a cold boot's three reads) now arms before the await and is cleared on both the success and failure paths. On expiry it publishes that the boot is hung rather than failed, and exits 1 immediately -- unlike a thrown boot there is no new reason to linger over. A RECOVERING WORKER COULD READ AS DEAD The failure counters cleared at the end of the poll body, while pollStartedAt is set at the top. A poll that recovers and then spends 300s processing what it claimed reported "every poll has failed for Nms" for that whole window, about a poll in the middle of succeeding. Cleared at the CLAIM now -- the moment the database proves it is answering. Narrower than the version that got reverted and conservative in direction, but this file's own docblock sets the bar: that class of mistake let a container healthcheck kill a healthy worker mid-job. A SUCCESSFUL BOOT OVERWROTE A CRASH VERDICT `if (!shuttingDown) boot = ready` did not check whether failFatally had already set `crashed` and armed exit(1) while boot was finishing. For up to five seconds /health answered 200 on a process committed to dying, and 200 is the one answer that makes a load balancer send work. ALSO shutdown() closing the health server before the drain is deliberate -- it inverts main's order so the probe fails immediately on SIGTERM rather than advertising a replica that has committed to dying for the length of the drain. It read as a reordering nobody noticed, so the reason is written down now, including what it costs: /health is unreachable during the drain, which is when an operator most wants it. TESTS The recovery window is pinned on the claimJobsByType path, which is the one production takes -- apps/worker/src/index.ts names all ten types, so hasPerTypeLimits is always true. The first version of that test used the other path and the mutation survived it. Mutation-checked: moving the clear back to the end of the poll body fails 1; inverting the poll-failure comparison fails 2. NOT CLAIMED ANY MORE: #181. worker.stop() sets running=false, but claimJobsByType's loop over typesWithCapacity does not re-check it, so an in-flight poll goes on claiming jobs for the remaining types after stop() has returned. This PR improves the loss profile and does not close that issue; the body said otherwise and no longer does. --- apps/worker/src/index.ts | 69 ++++++++++++++++++- .../src/__tests__/worker-concurrency.test.ts | 61 ++++++++++++++++ packages/outpost/queue/src/worker.ts | 20 ++++-- 3 files changed, 144 insertions(+), 6 deletions(-) diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 90468329..8095a5d6 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -82,6 +82,16 @@ import { // self-healing; the two were never actually in tension. let boot: BootState = { phase: 'starting' }; +/** + * Read `boot` without control-flow narrowing. + * + * TypeScript narrows the module-level `boot` to its initializer and cannot see + * that failFatally reassigns it from a process-level handler while an await is + * pending, so a direct `boot.phase === 'crashed'` reads as an impossible + * comparison. + */ +const currentBoot = (): BootState => boot; + let worker: Worker | null = null; let scheduler: Scheduler | null = null; @@ -231,6 +241,14 @@ async function shutdown(signal: string): Promise { // finish (up to 300s), and advertising a healthy /health for the whole // drain window tells the platform to keep routing to a replica that has // already committed to dying. + // Closed FIRST, which inverts main's order deliberately. worker.stop() + // blocks until in-flight jobs finish — up to the watchdog — and answering + // 200 for that whole window tells the platform to keep routing to a + // replica that has already committed to dying. The cost is that /health + // is unreachable during the drain, which is a real loss: the drain is + // when an operator most wants to ask what the worker is doing. The logs + // carry that instead, and routing work to a dying replica is the worse + // of the two. healthServer.close(); scheduler?.stop(); await worker?.stop(); @@ -393,13 +411,60 @@ if (lingerWarning) { configWarningList.push(lingerWarning); } +// How long boot may sit in `starting` before it is treated as failed. +// +// BOOT_FAILURE_LINGER_MS arms inside the catch, so it only ever watches a boot +// that THREW. A boot that HANGS never reaches it: buildSyncEngine() does three +// database reads and Prisma applies no query timeout, so a Postgres that accepts +// the connection and then stops answering — a failover, a saturated pool, a +// partition that drops packets without resetting — leaves that await pending +// forever. The process then sits at 503 `starting` processing nothing, and +// restartPolicyType="ALWAYS" cannot fire because nothing exits. +// +// That is the indefinite wedge the header above argues is unacceptable, reached +// through the one path the linger timer does not watch. Sized well above a cold +// boot's three reads so a slow-but-fine start is never cut short. +const { ms: BOOT_DEADLINE_MS, warning: deadlineWarning } = resolveDurationMs( + process.env.BOOT_DEADLINE_MS, + 'BOOT_DEADLINE_MS', + 180_000, +); +if (deadlineWarning) { + console.error(`[Worker] ${deadlineWarning}`); + configWarningList.push(deadlineWarning); +} + +const bootDeadline = setTimeout(() => { + if (currentBoot().phase !== 'starting' || shuttingDown) return; + boot = { + phase: 'failed', + error: `boot did not finish within ${BOOT_DEADLINE_MS}ms — it is hung rather than failed, most likely on a database read that never returned`, + }; + console.error(`[Worker] BOOT HUNG: ${boot.error}`); + console.error( + '[Worker] Exiting 1 so the restart policy retries; staying up would wedge this replica at zero jobs.', + ); + // Exits immediately rather than lingering: unlike a thrown boot there is no + // error to publish that a reader has not already had the whole window to see. + process.exit(1); +}, BOOT_DEADLINE_MS); + try { await startWorker(); + clearTimeout(bootDeadline); if (!shuttingDown) { - boot = { phase: 'ready' }; - console.log('[Worker] Worker process started'); + // Not if the process has already committed to dying. failFatally may have + // set `crashed` and armed exit(1) while boot was still finishing, and 200 + // is the one answer that makes a load balancer send work. + if (currentBoot().phase === 'crashed') { + console.error('[Worker] Boot finished after a fatal error; not reporting ready.'); + } else { + boot = { phase: 'ready' }; + console.log('[Worker] Worker process started'); + } } } catch (error) { + clearTimeout(bootDeadline); // startWorker() has already torn down anything it managed to start, so by // here the process holds no timers and no poll loop — only the health server. // diff --git a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts index 05334f8a..7c115e23 100644 --- a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts +++ b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts @@ -687,3 +687,64 @@ describe('completed jobs leave no overdue residue', () => { expect(worker.healthCheck().overdueJobCount).toBe(0); }); }); + +// A poll that recovers then spends a long time processing what it claimed must +// not still read as failing. The counters are cleared when the CLAIM returns — +// the moment the database proves it is answering — not at the end of the poll +// body, which can be minutes later. +describe('a recovering worker stops reporting failure at the claim, not at the end of the poll', () => { + let worker: InstanceType; + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(async () => { + if (worker) await worker.stop(); + vi.useRealTimers(); + }); + + it('clears the failure window while the recovered poll is still processing', async () => { + // concurrencyByType populated so hasPerTypeLimits is true and the poll + // takes the claimJobsByType path — the one production uses, since + // apps/worker/src/index.ts names all ten types. + worker = new Worker({ + pollIntervalMs: 50, + maxConcurrency: 1, + concurrencyByType: { [JobType.AI_RESPONSE]: 1 }, + }); + + let release: (() => void) | undefined; + const stillWorking = new Promise((resolve) => { + release = resolve; + }); + worker.on(JobType.AI_RESPONSE, async () => { + await stillWorking; + return { success: true }; + }); + + mockPrismaJob.update.mockResolvedValue({}); + mockPrisma.$queryRaw + .mockRejectedValueOnce(new Error('connection refused')) + .mockRejectedValueOnce(new Error('connection refused')) + .mockResolvedValueOnce([makeJobRow({ id: 'recovered-1' })]) + .mockResolvedValue([]); + + worker.start(); + await vi.advanceTimersByTimeAsync(60); + expect(worker.healthCheck().pollFailingSince).toBeInstanceOf(Date); + + // The claim has now succeeded and the job is in flight. The poll body has + // NOT finished — it is awaiting the handler. + await vi.advanceTimersByTimeAsync(60); + + const midRecovery = worker.healthCheck(); + expect(midRecovery.activeJobCount).toBe(1); + expect(midRecovery.pollFailingSince).toBeNull(); + expect(midRecovery.consecutivePollFailures).toBe(0); + + release?.(); + await vi.advanceTimersByTimeAsync(0); + }); +}); diff --git a/packages/outpost/queue/src/worker.ts b/packages/outpost/queue/src/worker.ts index 4b148e98..c5a72600 100644 --- a/packages/outpost/queue/src/worker.ts +++ b/packages/outpost/queue/src/worker.ts @@ -250,10 +250,9 @@ export class Worker { // If we processed jobs, poll immediately for more const nextPollDelay = processedCount > 0 ? 0 : this.pollIntervalMs; - // A poll that returned normally clears the failure window, however - // long it had been running. - this.pollFailingSince = null; - this.consecutivePollFailures = 0; + // Belt and braces: a poll that claimed nothing at all still proves + // the database answered. + this.recordPollSuccess(); this.completePoll(); this.reschedule(nextPollDelay); } catch (error) { @@ -271,6 +270,12 @@ export class Worker { * running, and leaving pollStartedAt set would report a dead loop as busy * forever. */ + /** The database answered a claim, so any open failure window is closed. */ + private recordPollSuccess(): void { + this.pollFailingSince = null; + this.consecutivePollFailures = 0; + } + private completePoll(): void { this.pollStartedAt = null; this.lastPollCompletedAt = new Date(); @@ -331,6 +336,12 @@ export class Worker { const limit = Math.min(available, remainingGlobalSlots, this.batchSize); const jobs = await this.claimJobsForType(type, limit); + // The claim came back, so the database is answering. Cleared here + // rather than at the end of the poll body: a poll that recovers then + // spends 300s processing what it claimed would otherwise report + // "every poll has failed" for that whole window, about a poll in the + // middle of succeeding. + this.recordPollSuccess(); if (jobs.length > 0) { const promises = jobs.map((job) => this.processJob(job)); @@ -406,6 +417,7 @@ export class Worker { ) RETURNING id, type, payload, attempts, "maxAttempts" `; + this.recordPollSuccess(); // Process jobs concurrently (each tracked in activeJobs) const promises = jobs.map(