add tests

server/test/unit/services/dbQueueWorker.test.ts

@@ -119,8 +119,10 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 		findRecent: jest.fn<any>().mockResolvedValue([]),
 	};
 	const logger = createMockLogger();
+	const isDbConnected = jest.fn<any>(() => true);
 
 	const mocks = {
+		isDbConnected,
 		jobsRepository,
 		monitorsRepository,
 		checksRepository,
@@ -138,6 +140,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 
 	const worker = new DBQueueWorker(
 		mocks.logger as any,
+		mocks.isDbConnected as any,
 		mocks.jobsRepository as any,
 		mocks.monitorsRepository as any,
 		mocks.checksRepository as any,
@@ -241,6 +244,7 @@ describe("DBQueueWorker", () => {
 			const { mocks } = createWorker(); // reuse the mock set; the un-init'd instance is discarded
 			const worker = await DBQueueWorker.create(
 				mocks.logger as any,
+				mocks.isDbConnected as any,
 				mocks.jobsRepository as any,
 				mocks.monitorsRepository as any,
 				mocks.checksRepository as any,
@@ -699,6 +703,71 @@ describe("DBQueueWorker", () => {
 		});
 	});
 
+	// ── getHealth ───────────────────────────────────────────────────────────────────
+
+	describe("getHealth", () => {
+		// Hand the "check" loop exactly one job, then nothing — keeps it parked in-flight.
+		const claimOneCheck = (job: Job) => {
+			let handed = false;
+			return jest.fn<any>(async (type: string) => {
+				if (type === job.type && !handed) {
+					handed = true;
+					return [job];
+				}
+				return [];
+			});
+		};
+
+		it("reports un-init state before init(): not complete, not draining, never ticked", () => {
+			const { worker } = createWorker({ queueMode: "worker" });
+			active = worker; // afterEach shutdown() is safe on an un-init'd worker
+			expect(worker.getHealth()).toMatchObject({
+				workerId: "worker-1",
+				mode: "worker",
+				initComplete: false,
+				draining: false,
+				lastTickAt: null,
+				inFlight: 0,
+			});
+		});
+
+		it("flips initComplete and stamps lastTickAt once the loops run", async () => {
+			const { worker } = await start({ queueMode: "worker" });
+			const health = worker.getHealth();
+			expect(health.initComplete).toBe(true);
+			expect(typeof health.lastTickAt).toBe("number");
+		});
+
+		it("reflects the injected dbConnected probe", async () => {
+			const { worker, mocks } = await start({ queueMode: "worker" });
+			expect(worker.getHealth().dbConnected).toBe(true);
+			mocks.isDbConnected.mockReturnValue(false);
+			expect(worker.getHealth().dbConnected).toBe(false);
+		});
+
+		it("sets draining true after drain()", async () => {
+			const { worker } = await start({ queueMode: "worker" });
+			expect(worker.getHealth().draining).toBe(false);
+			await worker.drain();
+			expect(worker.getHealth().draining).toBe(true);
+		});
+
+		it("counts in-flight jobs while they run and releases on completion", async () => {
+			const job = makeJob({ type: "check" });
+			const gate = deferred<{ status: unknown; check: unknown }>();
+			const slowProducer = { produce: jest.fn<any>().mockReturnValue(gate.promise) };
+			const { worker } = await start({
+				mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOneCheck(job) }, checkProducer: slowProducer },
+			});
+
+			expect(worker.getHealth().inFlight).toBe(1); // parked on produce()
+
+			gate.resolve({ status: { status: "up" }, check: { id: "c1" } });
+			await jest.advanceTimersByTimeAsync(1);
+			expect(worker.getHealth().inFlight).toBe(0);
+		});
+	});
+
 	// ── flushQueues ───────────────────────────────────────────────────────────────
 
 	describe("flushQueues", () => {

server/test/unit/services/healthServer.test.ts

@@ -0,0 +1,136 @@
+import { afterEach, describe, expect, it } from "@jest/globals";
+import http from "http";
+import net from "net";
+import { HealthServer } from "../../../src/worker/worker.health-server.ts";
+import type { WorkerHealth } from "../../../src/worker/worker.interface.ts";
+import { createMockLogger } from "../../helpers/createMockLogger.ts";
+
+// ── Notes ──────────────────────────────────────────────────────────────────────
+//
+// HealthServer is a dumb bare-http listener over the worker's getHealth() snapshot:
+// it maps the snapshot to 200/503 per endpoint and serves the snapshot as the body.
+// We drive it end-to-end with real sockets and a stub worker, so REAL timers are used
+// (Date.now() drives the /livez staleness check) — no fake timers in this file.
+//
+// listen() binds `Number(port) || 52346`, so "0" would resolve to 52346, not an
+// ephemeral port. We discover a free port first and pass it as a concrete string.
+
+const freePort = (): Promise<number> =>
+	new Promise((resolve, reject) => {
+		const srv = net.createServer();
+		srv.once("error", reject);
+		srv.listen(0, () => {
+			const addr = srv.address();
+			const port = typeof addr === "object" && addr ? addr.port : 0;
+			srv.close(() => resolve(port));
+		});
+	});
+
+const get = (port: number, path: string): Promise<{ status: number; body: string }> =>
+	new Promise((resolve, reject) => {
+		http
+			.get({ host: "127.0.0.1", port, path }, (res) => {
+				let body = "";
+				res.on("data", (chunk) => (body += chunk));
+				res.on("end", () => resolve({ status: res.statusCode ?? 0, body }));
+			})
+			.on("error", reject);
+	});
+
+// Fresh, fully-healthy snapshot; override per test. lastTickAt is stamped at call time
+// so it is "recent" relative to the request that follows a moment later.
+const makeHealth = (over?: Partial<WorkerHealth>): WorkerHealth => ({
+	workerId: "worker-1",
+	mode: "worker",
+	dbConnected: true,
+	initComplete: true,
+	draining: false,
+	lastTickAt: Date.now(),
+	inFlight: 0,
+	...over,
+});
+
+describe("HealthServer", () => {
+	let active: HealthServer | null = null;
+
+	afterEach(async () => {
+		if (active) await active.close();
+		active = null;
+	});
+
+	const setup = async (health: WorkerHealth) => {
+		const port = await freePort();
+		const worker = { getHealth: () => health } as any;
+		const server = new HealthServer(createMockLogger(), String(port), worker);
+		await server.listen();
+		active = server;
+		return { port };
+	};
+
+	describe("/livez", () => {
+		it("200 when init complete and a tick is recent", async () => {
+			const { port } = await setup(makeHealth());
+			expect((await get(port, "/livez")).status).toBe(200);
+		});
+
+		it("503 before init completes", async () => {
+			const { port } = await setup(makeHealth({ initComplete: false }));
+			expect((await get(port, "/livez")).status).toBe(503);
+		});
+
+		it("503 when it has never ticked", async () => {
+			const { port } = await setup(makeHealth({ lastTickAt: null }));
+			expect((await get(port, "/livez")).status).toBe(503);
+		});
+
+		it("503 when the last tick is stale and not draining", async () => {
+			const { port } = await setup(makeHealth({ lastTickAt: Date.now() - 31_000 }));
+			expect((await get(port, "/livez")).status).toBe(503);
+		});
+
+		it("200 when draining even if ticks are stale (drain must read as live)", async () => {
+			const { port } = await setup(makeHealth({ draining: true, lastTickAt: Date.now() - 60_000 }));
+			expect((await get(port, "/livez")).status).toBe(200);
+		});
+	});
+
+	describe("/readyz", () => {
+		it("200 when init complete, db connected, and not draining", async () => {
+			const { port } = await setup(makeHealth());
+			expect((await get(port, "/readyz")).status).toBe(200);
+		});
+
+		it("503 while draining", async () => {
+			const { port } = await setup(makeHealth({ draining: true }));
+			expect((await get(port, "/readyz")).status).toBe(503);
+		});
+
+		it("503 when the db is disconnected", async () => {
+			const { port } = await setup(makeHealth({ dbConnected: false }));
+			expect((await get(port, "/readyz")).status).toBe(503);
+		});
+
+		it("503 before init completes", async () => {
+			const { port } = await setup(makeHealth({ initComplete: false }));
+			expect((await get(port, "/readyz")).status).toBe(503);
+		});
+	});
+
+	describe("routing & body", () => {
+		it("serves the WorkerHealth snapshot as the JSON body", async () => {
+			const { port } = await setup(makeHealth({ inFlight: 3 }));
+			const res = await get(port, "/readyz");
+			expect(JSON.parse(res.body)).toMatchObject({ workerId: "worker-1", mode: "worker", inFlight: 3 });
+		});
+
+		it("404s unknown paths", async () => {
+			const { port } = await setup(makeHealth());
+			expect((await get(port, "/nope")).status).toBe(404);
+		});
+
+		it("ignores query params when matching the route", async () => {
+			const { port } = await setup(makeHealth());
+			expect((await get(port, "/livez?probe=1")).status).toBe(200);
+		});
+	});
+});

server/test/unit/services/jobScheduler.test.ts

@@ -22,12 +22,13 @@ const createScheduler = () => {
 
 describe("JobScheduler", () => {
 	describe("drain", () => {
-		it("sets the stop flag without tearing down (no deregister)", async () => {
+		it("sets the stop and draining flags without tearing down (no deregister)", async () => {
 			const { scheduler, mocks } = createScheduler();
 
 			await scheduler.drain();
 
 			expect((scheduler as any).stopped).toBe(true);
+			expect((scheduler as any).draining).toBe(true);
 			// drain must not do shutdown's teardown work
 			expect(mocks.queueWorkersRepository.deleteById).not.toHaveBeenCalled();
 		});