diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 5376b9f26..e0f168eb1 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -36,6 +36,10 @@ const CONCURRENCY: Record = { export class DBQueueWorker extends JobScheduler implements IQueueWorker { static SERVICE_NAME = SERVICE_NAME; + + private initComplete: boolean = false; + private lastTickAt: number | null = null; + private inFlight: Record = { check: 0, "geo-check": 0, @@ -46,6 +50,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { constructor( private logger: ILogger, + private isDbConnected: () => boolean, jobsRepository: IJobsRepository, // ← no modifier: forwarded to super private monitorsRepository: IMonitorsRepository, private checksRepository: IChecksRepository, @@ -70,6 +75,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { static async create( logger: ILogger, + isDbConnected: () => boolean, jobsRepository: IJobsRepository, monitorsRepository: IMonitorsRepository, checksRepository: IChecksRepository, @@ -87,6 +93,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { ): Promise { const instance = new DBQueueWorker( logger, + isDbConnected, jobsRepository, monitorsRepository, checksRepository, @@ -213,6 +220,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { private startLoop = (type: JobType) => { this.pollMs[type] = POLL_MS; const tick = async () => { + this.lastTickAt = Date.now(); if (this.stopped) return; this.ticking[type] = true; try { @@ -288,6 +296,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { this.startLoop(type); } } + this.initComplete = true; return true; }; @@ -298,6 +307,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { }; override drain = async () => { + this.draining = true; this.stopped = true; const cutoff = Date.now() + DRAIN_TIMEOUT_MS; while (this.getInFlightCount() > 0 && Date.now() < cutoff) { @@ -326,11 +336,11 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { return { workerId: this.workerId, mode: this.queueMode, - dbConnected: true, // TODO - initComplete: true, // TODO - draining: false, // TODO - lastTickAt: null, // TODO - inFlight: Number.MAX_SAFE_INTEGER, // TODO + dbConnected: this.isDbConnected(), + initComplete: this.initComplete, + draining: this.draining, + lastTickAt: this.lastTickAt, + inFlight: this.getInFlightCount(), }; }; } diff --git a/server/src/worker/worker.health-server.ts b/server/src/worker/worker.health-server.ts new file mode 100644 index 000000000..55727f734 --- /dev/null +++ b/server/src/worker/worker.health-server.ts @@ -0,0 +1,74 @@ +import http from "http"; +import { IQueueWorker, WorkerHealth } from "@/worker/worker.interface.js"; +import { ILogger } from "@/utils/logger.js"; + +const SERVICE_NAME = "HealthServer"; +const LIVENESS_STALE_MS = 30_000; // 30s ~6x POLL_MAX_MS so we don't restart workers + +export interface IHealthServer { + listen(): Promise; + close(): Promise; +} + +const isLive = (health: WorkerHealth): boolean => { + if (!health.initComplete) return false; // Not init yet + if (health.draining) return true; // Draining workers are still healthy + + if (health.lastTickAt === null) return false; // Never ticked + + if (Date.now() - health.lastTickAt >= LIVENESS_STALE_MS) return false; // Stale if last tick exceed LIVENESS_STALE_MS + + return true; +}; + +const isReady = (health: WorkerHealth): boolean => { + if (health.draining) return false; // Draining workers are not ready + return health.initComplete && health.dbConnected; +}; + +export class HealthServer implements IHealthServer { + private server: http.Server; + + constructor( + private logger: ILogger, + private port: string, + private worker: IQueueWorker + ) { + this.server = http.createServer(this.handle); + } + + private handle = (req: http.IncomingMessage, res: http.ServerResponse) => { + const path = (req.url ?? "").split("?")[0]; // Strip query params + if (path !== "/livez" && path !== "/readyz") { + res.writeHead(404).end(); + return; + } + + // Get health + const health = this.worker.getHealth(); + const ok = path === "/livez" ? isLive(health) : isReady(health); + const code = ok ? 200 : 503; + res.writeHead(code, { "Content-Type": "application/json" }); + res.end(JSON.stringify(health)); + }; + + listen = () => { + return new Promise((resolve) => { + const port = Number(this.port) || 52346; + this.server.listen(port, () => { + this.logger.info({ + message: `Health server listening on ${port}`, + service: SERVICE_NAME, + method: "listen", + }); + resolve(); + }); + }); + }; + + 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 2f3f40950..a431143ac 100644 --- a/server/src/worker/worker.job-scheduler.ts +++ b/server/src/worker/worker.job-scheduler.ts @@ -23,6 +23,7 @@ export class JobScheduler implements IJobScheduler { } protected stopped = false; + protected draining = false; protected ticking: Partial> = {}; // Whether a loop is mid tick protected tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake protected timers = new Map(); @@ -156,6 +157,7 @@ export class JobScheduler implements IJobScheduler { drain = async () => { this.stopped = true; + this.draining = true; }; shutdown = async () => {