diff --git a/apps/worker/src/__tests__/health.test.ts b/apps/worker/src/__tests__/health.test.ts new file mode 100644 index 00000000..59eb1747 --- /dev/null +++ b/apps/worker/src/__tests__/health.test.ts @@ -0,0 +1,777 @@ +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), + 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); + + 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, + // 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); + + 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'); + }, + ); + + // 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); + }); + + 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'])('accepts the valid port %j', (raw) => { + const { port, warning, fatal } = resolvePort({ PORT: raw }); + + 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, + 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); + }); +}); + +// 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 new file mode 100644 index 00000000..33a10998 --- /dev/null +++ b/apps/worker/src/health.ts @@ -0,0 +1,498 @@ +/** + * 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'; + +/** + * 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. + // 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`, + ); + } + } + + // 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; + + // 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. + 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 }; + + // 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, + 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`, + }, + }; + } + + // 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. + // 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', + ...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`, + }, + }; + } + + 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', + ...configWarnings(warnings), + 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..8095a5d6 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,444 @@ 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 - }, -}); +// 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' }; -// ─── Register Handlers ──────────────────────────────────────────────────── +/** + * 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; -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, + ); + // 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(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({ + 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. + // 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(); + 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); + } +} + +// `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('SIGTERM', () => shutdown('SIGTERM')); -process.on('SIGINT', () => shutdown('SIGINT')); +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); +} + +// 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) { + // 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. + // + // 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..7c115e23 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,473 @@ 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. +}); + +// 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); + }); +}); + +// 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/types.ts b/packages/outpost/queue/src/types.ts index e8503981..140ffe55 100644 --- a/packages/outpost/queue/src/types.ts +++ b/packages/outpost/queue/src/types.ts @@ -172,7 +172,41 @@ 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; + /** + * 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 6c7738ad..c5a72600 100644 --- a/packages/outpost/queue/src/worker.ts +++ b/packages/outpost/queue/src/worker.ts @@ -37,6 +37,41 @@ 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 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; private shutdownResolve: (() => void) | null = null; private signalHandlers: { signal: string; handler: () => void }[] = []; @@ -45,7 +80,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 +117,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 +131,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 +157,53 @@ 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, + pollFailingSince: this.pollFailingSince, + consecutivePollFailures: this.consecutivePollFailures, 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 +225,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 +250,51 @@ export class Worker { // If we processed jobs, poll immediately for more const nextPollDelay = processedCount > 0 ? 0 : this.pollIntervalMs; - this.pollTimer = setTimeout(() => this.poll(), nextPollDelay); + // Belt and braces: a poll that claimed nothing at all still proves + // the database answered. + this.recordPollSuccess(); + this.completePoll(); + this.reschedule(nextPollDelay); } catch (error) { console.error('[Queue Worker] Poll error:', error); - this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs); + this.consecutivePollFailures++; + this.pollFailingSince ??= new Date(); + 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. + */ + /** 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(); + } + + /** + * 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 @@ -210,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)); @@ -285,9 +417,18 @@ export class Worker { ) RETURNING id, type, payload, attempts, "maxAttempts" `; + this.recordPollSuccess(); // 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 +442,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 +492,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 +505,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 +524,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 {