diff --git a/docker/dev/docker-compose.yaml b/docker/dev/docker-compose.yaml index c5889ff35..09c12fe18 100755 --- a/docker/dev/docker-compose.yaml +++ b/docker/dev/docker-compose.yaml @@ -30,6 +30,7 @@ services: environment: QUEUE_MODE: "worker" HEALTH_PORT: "52346" + stop_grace_period: 30s healthcheck: test: [ diff --git a/server/src/worker/worker.health-server.ts b/server/src/worker/worker.health-server.ts index b68579b80..19bd22754 100644 --- a/server/src/worker/worker.health-server.ts +++ b/server/src/worker/worker.health-server.ts @@ -3,6 +3,8 @@ import { IQueueWorker, WorkerHealth } from "@/worker/worker.interface.js"; import { ILogger } from "@/utils/logger.js"; const SERVICE_NAME = "HealthServer"; +const DEFAULT_HEALTH_PORT = 52346; +const MAX_PORT_ATTEMPTS = 20; const LIVENESS_STALE_MS = 30_000; // 30s ~6x POLL_MAX_MS so we don't restart workers const CONTENT_TYPE_METRICS = "text/plain; version=0.0.4; charset=utf-8"; const gauge = (name: string, help: string, value: number): string => `# HELP ${name} ${help}\n# TYPE ${name} gauge\n${name} ${value}\n`; @@ -10,6 +12,7 @@ const gauge = (name: string, help: string, value: number): string => `# HELP ${n export interface IHealthServer { listen(): Promise; close(): Promise; + address(): ReturnType; } const isLive = (health: WorkerHealth): boolean => { @@ -84,9 +87,28 @@ export class HealthServer implements IHealthServer { }; listen = () => { - return new Promise((resolve) => { - const port = Number(this.port) || 52346; + return new Promise((resolve, reject) => { + let port = Number(this.port) || DEFAULT_HEALTH_PORT; + const lastPort = port + MAX_PORT_ATTEMPTS; + + const onError = (error: NodeJS.ErrnoException) => { + if (error.code === "EADDRINUSE" && port < lastPort) { + this.logger.warn({ + message: `Health port ${port} in use, retrying on ${port + 1}`, + service: SERVICE_NAME, + method: "listen", + }); + port++; + this.server.listen(port); + return; + } + this.server.removeListener("error", onError); + reject(error); + }; + + this.server.on("error", onError); this.server.listen(port, () => { + this.server.removeListener("error", onError); this.logger.info({ message: `Health server listening on ${port}`, service: SERVICE_NAME, @@ -97,6 +119,8 @@ export class HealthServer implements IHealthServer { }); }; + address = () => this.server.address(); + close = () => { return new Promise((resolve) => { this.server.close(() => resolve()); diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts index 0aa2e7a4b..458db1d08 100644 --- a/server/src/worker/worker.job-scheduler.ts +++ b/server/src/worker/worker.job-scheduler.ts @@ -29,6 +29,7 @@ export class JobScheduler implements IJobScheduler { get serviceName() { return JobScheduler.SERVICE_NAME; } + private heartbeatInFlight: Promise = Promise.resolve(); protected processesJobs = false; // scheduler-only by default; DBQueueWorker flips this on when it runs job loops protected stopped = false; @@ -205,7 +206,15 @@ export class JobScheduler implements IJobScheduler { // Register worker protected heartbeat = async () => { try { - await this.queueWorkersRepository.upsert(this.workerId, this.queueMode, this.processesJobs); + if (this.stopped) return; // Don't re-register if shutting down + this.heartbeatInFlight = this.queueWorkersRepository.upsert(this.workerId, this.queueMode, this.processesJobs).catch((error: unknown) => { + this.logger.warn({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: "heartbeat", + }); + }); + await this.heartbeatInFlight; } catch (error: unknown) { this.logger.warn({ message: error instanceof Error ? error.message : String(error), @@ -227,6 +236,7 @@ export class JobScheduler implements IJobScheduler { clearTimeout(timer); } this.timers.clear(); // Clear the map out + await this.heartbeatInFlight; // Wait for in flight jobs // Remove workers from registry await this.queueWorkersRepository.deleteById(this.workerId); }; diff --git a/server/test/unit/services/healthServer.test.ts b/server/test/unit/services/healthServer.test.ts index 1ca62d4bb..ccf9d535c 100644 --- a/server/test/unit/services/healthServer.test.ts +++ b/server/test/unit/services/healthServer.test.ts @@ -172,4 +172,28 @@ describe("HealthServer", () => { expect((await get(port, "/metrics")).status).toBe(500); }); }); + + describe("listen", () => { + it("walks to the next port when the configured one is in use", async () => { + const base = await freePort(); + // Hold the base port so the health server's first bind hits EADDRINUSE. + const blocker = net.createServer(); + await new Promise((resolve) => blocker.listen(base, resolve)); + + try { + const worker = { getHealth: () => makeHealth(), countDueBacklog: async () => 0, countAliveWorkers: async () => 0 } as any; + const server = new HealthServer(createMockLogger(), String(base), worker); + await server.listen(); + active = server; + + // base is still held by the blocker, so the health server retried onto a higher port and is reachable there. + const addr = server.address(); + const bound = typeof addr === "object" && addr ? addr.port : 0; + expect(bound).toBeGreaterThan(base); + expect((await get(bound, "/livez")).status).toBe(200); + } finally { + await new Promise((resolve) => blocker.close(() => resolve())); + } + }); + }); });