resolve shutdown race

docker/dev/docker-compose.yaml

@@ -30,6 +30,7 @@ services:
     environment:
       QUEUE_MODE: "worker"
       HEALTH_PORT: "52346"
+    stop_grace_period: 30s
     healthcheck:
       test:
         [

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<void>;
 	close(): Promise<void>;
+	address(): ReturnType<http.Server["address"]>;
 }
 
 const isLive = (health: WorkerHealth): boolean => {
@@ -84,9 +87,28 @@ export class HealthServer implements IHealthServer {
 	};
 
 	listen = () => {
-		return new Promise<void>((resolve) => {
-			const port = Number(this.port) || 52346;
+		return new Promise<void>((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<void>((resolve) => {
 			this.server.close(() => resolve());

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<void> = 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);
 	};

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<void>((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<void>((resolve) => blocker.close(() => resolve()));
+			}
+		});
+	});
 });