diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 40a208aa2..dd210b3e2 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -119,8 +119,10 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses findRecent: jest.fn().mockResolvedValue([]), }; const logger = createMockLogger(); + const isDbConnected = jest.fn(() => 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(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().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", () => { diff --git a/server/test/unit/services/healthServer.test.ts b/server/test/unit/services/healthServer.test.ts new file mode 100644 index 000000000..208cd0902 --- /dev/null +++ b/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 => + 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 => ({ + 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); + }); + }); +}); diff --git a/server/test/unit/services/jobScheduler.test.ts b/server/test/unit/services/jobScheduler.test.ts index 9fd2ee0db..dceb8760c 100644 --- a/server/test/unit/services/jobScheduler.test.ts +++ b/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(); });