← back to Homesonspec
TK-10240: homesonspec worker resilience — pg-boss error-ride-out + graceful SIGINT/SIGTERM + health canary
44bdb168c19ad81642511f836f59df061739f5fd · 2026-08-08 08:44:27 -0700 · Steve
- queue.ts: boss.on('error') logs+records instead of crashing (pool reconnects lazily); no more process.exit on a transient DB blip
- installGracefulShutdown(): SIGINT/SIGTERM -> stopQueue() drain, injectable/testable, per-signal no-double-register guard
- health canary (computeQueueHealth + startQueueHealthCanary) + /health 503-on-wedged endpoint in worker: detects the 'wedged-but-online' failure mode the ticket flagged
- honest stopQueue comment (boss.stop waits then FAILs WIP per retry policy, not silent drain)
- 8 vitest tests (shutdown state machine + health verdict), shared+workers typecheck clean
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Files touched
M apps/workers/src/index.tsA packages/shared/src/queue.test.tsM packages/shared/src/queue.ts
Diff
commit 44bdb168c19ad81642511f836f59df061739f5fd
Author: Steve <steve@designerwallcoverings.com>
Date: Sat Aug 8 08:44:27 2026 -0700
TK-10240: homesonspec worker resilience — pg-boss error-ride-out + graceful SIGINT/SIGTERM + health canary
- queue.ts: boss.on('error') logs+records instead of crashing (pool reconnects lazily); no more process.exit on a transient DB blip
- installGracefulShutdown(): SIGINT/SIGTERM -> stopQueue() drain, injectable/testable, per-signal no-double-register guard
- health canary (computeQueueHealth + startQueueHealthCanary) + /health 503-on-wedged endpoint in worker: detects the 'wedged-but-online' failure mode the ticket flagged
- honest stopQueue comment (boss.stop waits then FAILs WIP per retry policy, not silent drain)
- 8 vitest tests (shutdown state machine + health verdict), shared+workers typecheck clean
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
apps/workers/src/index.ts | 43 +++++++++++-
packages/shared/src/queue.test.ts | 73 ++++++++++++++++++++
packages/shared/src/queue.ts | 141 ++++++++++++++++++++++++++++++++++++++
3 files changed, 255 insertions(+), 2 deletions(-)
diff --git a/apps/workers/src/index.ts b/apps/workers/src/index.ts
index 13e51382..cf81245f 100644
--- a/apps/workers/src/index.ts
+++ b/apps/workers/src/index.ts
@@ -1,4 +1,12 @@
-import { work } from "@homesonspec/shared";
+import http from "node:http";
+import {
+ work,
+ stopQueue,
+ installGracefulShutdown,
+ startQueueHealthCanary,
+ computeQueueHealth,
+ getQueueHealthState,
+} from "@homesonspec/shared";
import { meridianHomesAdapter } from "@homesonspec/collector-meridian-homes";
import { runPipeline } from "./pipeline";
@@ -8,15 +16,46 @@ import { runPipeline } from "./pipeline";
* refresh jobs per source-registry intervals lands with live sources.
*/
const ADAPTERS = { [meridianHomesAdapter.key]: meridianHomesAdapter };
+const JOB_NAME = "pipeline.run";
+const HEALTH_PORT = Number(process.env.WORKER_HEALTH_PORT ?? 9976);
async function main() {
- await work<{ adapterKey: string }>("pipeline.run", async ({ adapterKey }) => {
+ await work<{ adapterKey: string }>(JOB_NAME, async ({ adapterKey }) => {
const adapter = ADAPTERS[adapterKey];
if (!adapter) throw new Error(`unknown adapter ${adapterKey}`);
const summary = await runPipeline(adapter);
console.log(`[pipeline.run] ${adapterKey}:`, JSON.stringify(summary));
});
console.log("workers: subscribed to pipeline.run");
+
+ // Liveness probe: distinguishes a healthy-idle worker from a wedged-but-online
+ // one (pg-boss no longer crashes on a DB error, so pm2 would otherwise report
+ // "online" through an outage). 503 when the queue has errored recently without
+ // a canary recovery — a pm2/monitoring health-check can act on it.
+ const healthServer = http.createServer((req, res) => {
+ if (req.url === "/health") {
+ const verdict = computeQueueHealth(getQueueHealthState(), Date.now());
+ res.writeHead(verdict.healthy ? 200 : 503, { "content-type": "application/json" });
+ res.end(JSON.stringify(verdict));
+ return;
+ }
+ res.writeHead(404);
+ res.end();
+ });
+ healthServer.listen(HEALTH_PORT, () => console.log(`workers: health endpoint on :${HEALTH_PORT}/health`));
+
+ const canary = startQueueHealthCanary({ queueName: JOB_NAME });
+
+ // pm2 reload/restart -> SIGINT/SIGTERM: stop the canary + health server and
+ // drain the queue (see stopQueue) instead of dropping in-flight leases.
+ installGracefulShutdown({
+ stop: async () => {
+ clearInterval(canary);
+ healthServer.close();
+ await stopQueue();
+ },
+ });
+ console.log("workers: graceful shutdown wired (SIGINT/SIGTERM)");
}
main().catch((error) => {
diff --git a/packages/shared/src/queue.test.ts b/packages/shared/src/queue.test.ts
new file mode 100644
index 00000000..d463127f
--- /dev/null
+++ b/packages/shared/src/queue.test.ts
@@ -0,0 +1,73 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { installGracefulShutdown, computeQueueHealth } from "./queue";
+
+describe("installGracefulShutdown", () => {
+ it("stops the queue then exits 0 on a termination signal", async () => {
+ const stop = vi.fn().mockResolvedValue(undefined);
+ const exit = vi.fn();
+ const handler = installGracefulShutdown({ stop, exit, log: () => {}, signals: [] });
+
+ await handler("SIGTERM");
+
+ expect(stop).toHaveBeenCalledTimes(1);
+ expect(exit).toHaveBeenCalledWith(0);
+ });
+
+ it("is idempotent: a second signal during shutdown does not stop twice", async () => {
+ let resolveStop: () => void = () => {};
+ const stop = vi.fn(() => new Promise<void>((r) => (resolveStop = r)));
+ const exit = vi.fn();
+ const handler = installGracefulShutdown({ stop, exit, log: () => {}, signals: [] });
+
+ const first = handler("SIGINT");
+ await handler("SIGINT"); // arrives while first is still in flight -> ignored
+ resolveStop();
+ await first;
+
+ expect(stop).toHaveBeenCalledTimes(1);
+ expect(exit).toHaveBeenCalledTimes(1);
+ expect(exit).toHaveBeenCalledWith(0);
+ });
+
+ it("exits 1 (not a hang or crash) when the graceful stop throws", async () => {
+ const stop = vi.fn().mockRejectedValue(new Error("boss.stop failed"));
+ const exit = vi.fn();
+ const handler = installGracefulShutdown({ stop, exit, log: () => {}, signals: [] });
+
+ await handler("SIGTERM");
+
+ expect(stop).toHaveBeenCalledTimes(1);
+ expect(exit).toHaveBeenCalledWith(1);
+ });
+
+ it("does not stack process listeners when wired twice for the same signal", () => {
+ const before = process.listenerCount("SIGUSR2");
+ const args = { stop: vi.fn().mockResolvedValue(undefined), exit: vi.fn(), log: () => {}, signals: ["SIGUSR2" as NodeJS.Signals] };
+ installGracefulShutdown(args);
+ installGracefulShutdown(args); // second call must be a no-op for registration
+ expect(process.listenerCount("SIGUSR2")).toBe(before + 1);
+ process.removeAllListeners("SIGUSR2");
+ });
+});
+
+describe("computeQueueHealth", () => {
+ const now = 1_000_000;
+
+ it("is healthy when there has never been an error", () => {
+ expect(computeQueueHealth({ lastErrorAt: null, lastHealthyAt: null }, now).healthy).toBe(true);
+ });
+
+ it("is UNHEALTHY when a recent error has not been cleared by a canary", () => {
+ const v = computeQueueHealth({ lastErrorAt: now - 5_000, lastHealthyAt: null }, now);
+ expect(v.healthy).toBe(false);
+ expect(v.reason).toMatch(/not recovered/);
+ });
+
+ it("is healthy again once a canary check succeeds after the error", () => {
+ expect(computeQueueHealth({ lastErrorAt: now - 5_000, lastHealthyAt: now - 1_000 }, now).healthy).toBe(true);
+ });
+
+ it("is healthy again once the error ages past the window", () => {
+ expect(computeQueueHealth({ lastErrorAt: now - 45_000, lastHealthyAt: null }, now).healthy).toBe(true);
+ });
+});
diff --git a/packages/shared/src/queue.ts b/packages/shared/src/queue.ts
index ca6761ca..86a1abd6 100644
--- a/packages/shared/src/queue.ts
+++ b/packages/shared/src/queue.ts
@@ -7,11 +7,78 @@ import PgBoss from "pg-boss";
let boss: PgBoss | null = null;
+// --- Queue health tracking (detects the "wedged-but-online" failure mode) ---
+// After TK-10240 the worker no longer crashes on a transient DB error (see the
+// getQueue 'error' handler): pg-boss/pg-pool lazily re-establish connections, so
+// a momentary blip is survived. The tradeoff is that a LONGER outage leaves the
+// worker "online" to pm2 while it silently fails every poll cycle. This state +
+// computeQueueHealth() + startQueueHealthCanary() give a health probe something
+// real to read so a wedged worker is detectable (the canary the ticket asked
+// for), rather than trading a visible crash-loop for an invisible wedge.
+interface QueueHealthState {
+ lastErrorAt: number | null;
+ lastErrorMessage: string | null;
+ lastHealthyAt: number | null;
+ startedAt: number;
+}
+
+const health: QueueHealthState = {
+ lastErrorAt: null,
+ lastErrorMessage: null,
+ lastHealthyAt: null,
+ startedAt: Date.now(),
+};
+
+export function recordQueueError(err: unknown): void {
+ health.lastErrorAt = Date.now();
+ health.lastErrorMessage = err instanceof Error ? err.message : String(err);
+}
+
+export function recordQueueHealthy(): void {
+ health.lastHealthyAt = Date.now();
+ health.lastErrorAt = null;
+ health.lastErrorMessage = null;
+}
+
+export function getQueueHealthState(): QueueHealthState {
+ return { ...health };
+}
+
+/**
+ * Pure health verdict from a state snapshot. Unhealthy iff a pg-boss error
+ * occurred within `unhealthyAfterMs` and no successful canary check has cleared
+ * it since (i.e. the worker is online but not recovering).
+ */
+export function computeQueueHealth(
+ state: Pick<QueueHealthState, "lastErrorAt" | "lastHealthyAt">,
+ now: number,
+ unhealthyAfterMs = 30_000,
+): { healthy: boolean; reason: string } {
+ if (state.lastErrorAt !== null && now - state.lastErrorAt < unhealthyAfterMs) {
+ if (state.lastHealthyAt === null || state.lastHealthyAt < state.lastErrorAt) {
+ return {
+ healthy: false,
+ reason: `pg-boss error ${now - state.lastErrorAt}ms ago, not recovered by a canary check`,
+ };
+ }
+ }
+ return { healthy: true, reason: "ok" };
+}
+
export async function getQueue(): Promise<PgBoss> {
if (!boss) {
const url = process.env.DATABASE_URL;
if (!url) throw new Error("DATABASE_URL is required for the job queue");
boss = new PgBoss(url);
+ // Ride out transient DB blips: pg-boss (via pg-pool) removes a dropped idle
+ // client and lazily creates a fresh one on the next query. Without an
+ // 'error' listener Node treats the emitted 'error' event as fatal (uncaught
+ // -> process crash) — the exact process.exit-on-blip TK-10240 fixes. Record
+ // it for the health probe and continue instead of dying.
+ boss.on("error", (err) => {
+ recordQueueError(err);
+ console.error("[queue] pg-boss error (non-fatal; pool reconnects lazily):", err);
+ });
await boss.start();
}
return boss;
@@ -34,9 +101,83 @@ export async function work<T extends object>(
});
}
+/**
+ * Periodically issue a trivial round-trip (getQueueSize) so the worker can tell
+ * "idle" apart from "wedged": a success marks the queue healthy (clearing a
+ * prior error), a failure records it. Pair with computeQueueHealth() behind a
+ * /health probe. Timer is unref'd so it never keeps the process alive on its own.
+ */
+export function startQueueHealthCanary(opts: { queueName: string; intervalMs?: number }): NodeJS.Timeout {
+ const intervalMs = opts.intervalMs ?? 60_000;
+ const timer = setInterval(async () => {
+ try {
+ const queue = await getQueue();
+ await queue.getQueueSize(opts.queueName);
+ recordQueueHealthy();
+ } catch (err) {
+ recordQueueError(err);
+ }
+ }, intervalMs);
+ timer.unref?.();
+ return timer;
+}
+
export async function stopQueue(): Promise<void> {
if (boss) {
+ // boss.stop() stops polling and waits up to its configured timeout (default
+ // ~30s) for in-flight jobs to finish; anything still active when the timeout
+ // elapses is FAILED ('pg-boss shut down while active') and retried per the
+ // queue's retry policy — NOT silently dropped. Either way this is a clean
+ // shutdown vs. a SIGKILL that drops the lease (redelivered only after TTL).
await boss.stop();
boss = null;
}
}
+
+// Guard so repeated installGracefulShutdown() calls don't stack process
+// listeners (the shared package is imported by every worker process).
+const registeredSignals = new Set<NodeJS.Signals>();
+
+/**
+ * Wire process-termination signals to a graceful queue stop. pm2 reload/restart
+ * sends SIGINT/SIGTERM; without this the process is killed while pg-boss still
+ * holds in-flight job leases (see stopQueue for the exact drain semantics).
+ *
+ * Dependencies are injectable so the handler is unit-testable without a live DB
+ * or actually terminating the test process. Returns the handler for tests.
+ * Idempotent: a second signal while shutdown is in flight is ignored, and a
+ * given OS signal is only ever wired once.
+ */
+export function installGracefulShutdown(opts: {
+ stop?: () => Promise<void>;
+ exit?: (code: number) => void;
+ log?: (...args: unknown[]) => void;
+ signals?: NodeJS.Signals[];
+} = {}): (signal: string) => Promise<void> {
+ const stop = opts.stop ?? stopQueue;
+ const exit = opts.exit ?? ((code: number) => process.exit(code));
+ const log = opts.log ?? ((...args: unknown[]) => console.log("[queue]", ...args));
+ const signals = opts.signals ?? (["SIGINT", "SIGTERM"] as NodeJS.Signals[]);
+
+ let shuttingDown = false;
+ const handler = async (signal: string): Promise<void> => {
+ if (shuttingDown) return;
+ shuttingDown = true;
+ log(`received ${signal}, stopping queue gracefully...`);
+ try {
+ await stop();
+ log("queue stopped, exiting cleanly");
+ exit(0);
+ } catch (err) {
+ log("error during graceful shutdown:", err);
+ exit(1);
+ }
+ };
+
+ for (const sig of signals) {
+ if (registeredSignals.has(sig)) continue;
+ registeredSignals.add(sig);
+ process.on(sig, () => void handler(sig));
+ }
+ return handler;
+}
← ccf602c5 Add 5 web-rendered 1320x2868 device-spec App Store screensho
·
back to Homesonspec
·
test(workers): real-Postgres integration test for TK-10240 p 2831eee7 →