From 8579edd56ea182e61b1d2d41d1b4d8a923eff301 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 11:58:32 -0700 Subject: [PATCH 01/16] add a shutdown method to buffer service --- server/src/service/bufferService.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/server/src/service/bufferService.ts b/server/src/service/bufferService.ts index 989ebf936..d47a6472e 100755 --- a/server/src/service/bufferService.ts +++ b/server/src/service/bufferService.ts @@ -13,6 +13,7 @@ export interface IBufferService { scheduleNextFlush(): void; flushBuffer(): Promise; flushGeoBuffer(): Promise; + shutdown(): Promise; } export class BufferService implements IBufferService { @@ -152,4 +153,13 @@ export class BufferService implements IBufferService { this.geoBuffer = []; } } + + async shutdown() { + if (this.bufferTimer) { + clearTimeout(this.bufferTimer); + this.bufferTimer = null; + } + await this.flushBuffer(); + await this.flushGeoBuffer(); + } } From 860b2d621470ba206bdd74c4d2f0bb89d7d6fb5e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 12:16:37 -0700 Subject: [PATCH 02/16] add a getHealth method to interface, stub out implementation --- server/src/worker/worker.db-queue.ts | 46 +++++++++++++++++++++++++++ server/src/worker/worker.interface.ts | 12 +++++++ 2 files changed, 58 insertions(+) diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index ec7d1cf94..5376b9f26 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -15,6 +15,7 @@ import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.rep import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; import { JobScheduler } from "@/worker/worker.job-scheduler.js"; +import { IBufferService } from "@/service/bufferService.js"; const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const POLL_MAX_MS = 5000; // back off poll interval to 5s if no jobs @@ -22,6 +23,9 @@ const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive i const HEARTBEAT_MS = WORKER_STALE_MS / 3; // Worker can miss two beats without being considered stale const LOCK_RENEW_MS = LOCK_MS / 3; // renew an in-flight job's lock at 1/3 the lease, so it survives two missed renewals const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; +const DRAIN_TIMEOUT_MS = 25_000; +const DRAIN_POLL_MS = 200; + const CONCURRENCY: Record = { check: 500, "geo-check": 20, @@ -46,6 +50,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { private monitorsRepository: IMonitorsRepository, private checksRepository: IChecksRepository, private checkService: ICheckService, + private bufferService: IBufferService, private checkProducer: ICheckProducer, private checkEvaluator: ICheckEvaluator, private geoCheckPipeline: ICheckPipeline, @@ -69,6 +74,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { monitorsRepository: IMonitorsRepository, checksRepository: IChecksRepository, checkService: ICheckService, + bufferService: IBufferService, checkProducer: ICheckProducer, checkEvaluator: ICheckEvaluator, geoCheckPipeline: ICheckPipeline, @@ -85,6 +91,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, @@ -99,6 +106,8 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { return instance; } + private getInFlightCount = () => Object.values(this.inFlight).reduce((sum, n) => sum + n, 0); + private toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({ id: jobId(type, null), type, @@ -287,4 +296,41 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { const ok = await this.init(); return { success: ok }; }; + + override drain = async () => { + this.stopped = true; + const cutoff = Date.now() + DRAIN_TIMEOUT_MS; + while (this.getInFlightCount() > 0 && Date.now() < cutoff) { + await new Promise((resolve) => setTimeout(resolve, DRAIN_POLL_MS)); // Wait for DRAIN_POLL_MS for jobs to finish + } + + const remainingJobs = this.getInFlightCount(); + if (remainingJobs > 0) { + this.logger.warn({ + message: `Draining timed out with ${remainingJobs} in-flight. Locks will expire and jobs will be reclaimed`, + service: SERVICE_NAME, + method: "drain", + }); + } else { + this.logger.info({ + message: `${this.workerId} drained`, + service: SERVICE_NAME, + method: "drain", + }); + } + + await this.bufferService.shutdown(); // Flush buffers + }; + + getHealth = () => { + 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 + }; + }; } diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 522ed3204..6f7c57d07 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -4,6 +4,7 @@ import { JobType } from "@/domain/jobs/job.type.js"; import { MonitorPayloadMap, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js"; import { MonitorActionDecision } from "@/worker/worker.helper.js"; import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; +import type { QueueMode } from "@/domain/app-settings/app-settings.type.js"; export type MonitorEvaluation = { monitor: Monitor; @@ -56,6 +57,15 @@ export type WorkerJobsPage = { count: number; }; +export type WorkerHealth = { + workerId: string; + mode: QueueMode; + dbConnected: boolean; // mongoose connection readyState === 1 + initComplete: boolean; // init() finished, loops armed + draining: boolean; // SIGTERM received, no longer claiming + lastTickAt: number | null; // newest tick across all loops (epoch ms) + inFlight: number; // total in-flight jobs across types +}; export interface IJobScheduler { readonly serviceName: string; addJob(monitorId: string, monitor: Monitor): Promise; @@ -67,10 +77,12 @@ export interface IJobScheduler { getMetrics(): Promise; getJobs(pagination: WorkerJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; + drain(): Promise; shutdown(): Promise; } export interface IQueueWorker extends IJobScheduler { readonly serviceName: string; init(): Promise; + getHealth(): WorkerHealth; } From 5fab2d0a30056cbade8aabf55c495275a5d76ee9 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 12:17:02 -0700 Subject: [PATCH 03/16] inject buffer service, implement drain --- server/src/config/services.worker.ts | 1 + server/src/shutdown.ts | 1 + server/src/worker/worker.job-scheduler.ts | 4 ++++ 3 files changed, 6 insertions(+) diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index 93233cd37..e3996d09c 100644 --- a/server/src/config/services.worker.ts +++ b/server/src/config/services.worker.ts @@ -136,6 +136,7 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 625547e85..bf77a77c9 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -24,6 +24,7 @@ export const initShutdownListener = (server: Server | null, services: ShutdownTa try { server?.close(); + await services.worker.drain(); await services.worker.shutdown(); await services.db.disconnect(); logger.info({ message: "Graceful shutdown complete" }); diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts index 4e05e4c38..2f3f40950 100644 --- a/server/src/worker/worker.job-scheduler.ts +++ b/server/src/worker/worker.job-scheduler.ts @@ -154,6 +154,10 @@ export class JobScheduler implements IJobScheduler { return { success: true }; }; + drain = async () => { + this.stopped = true; + }; + shutdown = async () => { this.stopped = true; // Stop all timers From 7a82c76d5c44da2b94310e2ac687e913ac676e2f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 12:20:21 -0700 Subject: [PATCH 04/16] update tests --- .../test/unit/services/bufferService.test.ts | 28 +++++++ .../test/unit/services/dbQueueWorker.test.ts | 80 +++++++++++++++++++ 2 files changed, 108 insertions(+) diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index 7c344c086..d994e5f24 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/server/test/unit/services/bufferService.test.ts @@ -482,4 +482,32 @@ describe("BufferService", () => { ); }); }); + + // ── shutdown ───────────────────────────────────────────────────────────── + + describe("shutdown", () => { + it("stops the flush timer and flushes both buffers", async () => { + const { service, checkService, geoChecksService } = createService(); + service.addToBuffer(makeCheck()); + service.addGeoCheckToBuffer(makeGeoCheck()); + + await service.shutdown(); + + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + expect(geoChecksService.createGeoChecks).toHaveBeenCalledTimes(1); + + // Timer is cleared: advancing past the flush interval triggers no further flush. + await jest.advanceTimersByTimeAsync(60 * 1000); + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + }); + + it("is safe to call with empty buffers", async () => { + const { service, checkService, geoChecksService } = createService(); + + await service.shutdown(); + + expect(checkService.createChecks).not.toHaveBeenCalled(); + expect(geoChecksService.createGeoChecks).not.toHaveBeenCalled(); + }); + }); }); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 66ca6753d..40a208aa2 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -20,6 +20,8 @@ import { createMockLogger } from "../../helpers/createMockLogger.ts"; const POLL_MS = 250; const LOCK_RENEW_MS = LOCK_MS / 3; +const DRAIN_TIMEOUT_MS = 25_000; // mirror of DBQueueWorker DRAIN_TIMEOUT_MS +const DRAIN_POLL_MS = 200; // mirror of DBQueueWorker DRAIN_POLL_MS const deferred = () => { let resolve!: (value: T) => void; @@ -91,6 +93,14 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses toStatusResponse: jest.fn().mockReturnValue({ status: "up" }), toLastEvaluatedAt: jest.fn().mockReturnValue(12345), }; + const bufferService = { + addToBuffer: jest.fn(), + addGeoCheckToBuffer: jest.fn(), + scheduleNextFlush: jest.fn(), + flushBuffer: jest.fn().mockResolvedValue(undefined), + flushGeoBuffer: jest.fn().mockResolvedValue(undefined), + shutdown: jest.fn().mockResolvedValue(undefined), + }; const checkProducer = { produce: jest.fn().mockResolvedValue({ status: { status: "up" }, check: { id: "c1" } }), }; @@ -115,6 +125,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, @@ -131,6 +142,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses mocks.monitorsRepository as any, mocks.checksRepository as any, mocks.checkService as any, + mocks.bufferService as any, mocks.checkProducer as any, mocks.checkEvaluator as any, mocks.geoCheckPipeline as any, @@ -233,6 +245,7 @@ describe("DBQueueWorker", () => { mocks.monitorsRepository as any, mocks.checksRepository as any, mocks.checkService as any, + mocks.bufferService as any, mocks.checkProducer as any, mocks.checkEvaluator as any, mocks.geoCheckPipeline as any, @@ -619,6 +632,73 @@ describe("DBQueueWorker", () => { }); }); + // ── drain ───────────────────────────────────────────────────────────────────── + + describe("drain", () => { + // Hand the "check" loop exactly one job, then nothing. + const claimOnce = (job: Job) => { + let handed = false; + return jest.fn(async (type: string) => { + if (type === job.type && !handed) { + handed = true; + return [job]; + } + return []; + }); + }; + + it("stops the loops and flushes the buffer when nothing is in flight", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + const claimsBefore = mocks.jobsRepository.claimDueBatch.mock.calls.length; + + await worker.drain(); + + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + expect(mocks.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("drained") })); + // Loops must not claim again after draining. + await jest.advanceTimersByTimeAsync(POLL_MS * 4); + expect(mocks.jobsRepository.claimDueBatch.mock.calls.length).toBe(claimsBefore); + }); + + it("waits for an in-flight job to finish before flushing", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const slowProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker, mocks } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: slowProducer }, + }); + + // Job is parked on produce(); drain must not flush while it is still in flight. + const draining = worker.drain(); + await jest.advanceTimersByTimeAsync(DRAIN_POLL_MS * 3); + expect(mocks.bufferService.shutdown).not.toHaveBeenCalled(); + + // Let the job finish; drain observes inFlight == 0, then flushes. + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); + await jest.advanceTimersByTimeAsync(DRAIN_POLL_MS); + await draining; + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + }); + + it("times out after DRAIN_TIMEOUT_MS, warns, and still flushes", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const stuckProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker, mocks } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: stuckProducer }, + }); + + const draining = worker.drain(); + await jest.advanceTimersByTimeAsync(DRAIN_TIMEOUT_MS + DRAIN_POLL_MS); + await draining; + + expect(mocks.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("timed out") })); + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); // release the stuck job for cleanup + }); + }); + // ── flushQueues ─────────────────────────────────────────────────────────────── describe("flushQueues", () => { From fd536541d7b90ff738387fd5381d3db1bd39c67e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 09:16:07 -0700 Subject: [PATCH 05/16] add test for job scheduler drain --- .../test/unit/services/jobScheduler.test.ts | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 server/test/unit/services/jobScheduler.test.ts diff --git a/server/test/unit/services/jobScheduler.test.ts b/server/test/unit/services/jobScheduler.test.ts new file mode 100644 index 000000000..9fd2ee0db --- /dev/null +++ b/server/test/unit/services/jobScheduler.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it, jest } from "@jest/globals"; +import { JobScheduler } from "../../../src/worker/worker.job-scheduler.ts"; +import { createMockLogger } from "../../helpers/createMockLogger.ts"; + +// ── Notes ────────────────────────────────────────────────────────────────────── +// +// JobScheduler is the scheduler-only primary base class: it owns the queue +// registry/heartbeat but processes no jobs and has no buffer. drain() is therefore +// a no-op beyond setting the stop flag — it must NOT tear down (clear timers or +// deregister the worker); that stays in shutdown(). The real draining (wait for +// in-flight, flush buffer) lives in DBQueueWorker and is covered separately. + +const createScheduler = () => { + const jobsRepository = {} as any; + const queueWorkersRepository = { + deleteById: jest.fn().mockResolvedValue(undefined), + }; + const logger = createMockLogger(); + const scheduler = new JobScheduler(jobsRepository, queueWorkersRepository as any, "worker-1"); + return { scheduler, mocks: { jobsRepository, queueWorkersRepository, logger } }; +}; + +describe("JobScheduler", () => { + describe("drain", () => { + it("sets the stop flag without tearing down (no deregister)", async () => { + const { scheduler, mocks } = createScheduler(); + + await scheduler.drain(); + + expect((scheduler as any).stopped).toBe(true); + // drain must not do shutdown's teardown work + expect(mocks.queueWorkersRepository.deleteById).not.toHaveBeenCalled(); + }); + + it("stops wake() from arming a tick once drained", async () => { + const { scheduler } = createScheduler(); + const tick = jest.fn(); + (scheduler as any).tickFns.set("check", tick); + + await scheduler.drain(); + scheduler.wake("check"); + + expect((scheduler as any).timers.has("check")).toBe(false); + }); + }); +}); From b3d9c5b94c019a08cc8fa742ce3c9faf3eaf7320 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 10:36:31 -0700 Subject: [PATCH 06/16] add health server and metrics to db queue worker --- server/src/worker/worker.db-queue.ts | 20 ++++-- server/src/worker/worker.health-server.ts | 74 +++++++++++++++++++++++ server/src/worker/worker.job-scheduler.ts | 2 + 3 files changed, 91 insertions(+), 5 deletions(-) create mode 100644 server/src/worker/worker.health-server.ts 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 () => { From c9f0254aeda0f4404b6c01120e3acdd4150e7b18 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 10:36:53 -0700 Subject: [PATCH 07/16] start health server, update shutdown, update env config --- server/src/config/envValidation.ts | 2 ++ server/src/config/services.worker.ts | 2 ++ server/src/index.ts | 5 ++++- server/src/shutdown.ts | 13 ++++++++----- 4 files changed, 16 insertions(+), 6 deletions(-) diff --git a/server/src/config/envValidation.ts b/server/src/config/envValidation.ts index bdfacf906..05471e690 100644 --- a/server/src/config/envValidation.ts +++ b/server/src/config/envValidation.ts @@ -6,6 +6,8 @@ import { ILogger } from "@/utils/logger.js"; const envSchema = z.object({ // Server Configuration PORT: z.string().default("52345"), + // Port for worker health checks + HEALTH_PORT: z.string().default("52346"), NODE_ENV: z.enum(["development", "production", "test"]).default("development"), LOG_LEVEL: z.enum(["error", "warn", "info", "debug"]).default("debug"), diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index e3996d09c..44997af5f 100644 --- a/server/src/config/services.worker.ts +++ b/server/src/config/services.worker.ts @@ -9,6 +9,7 @@ import jmespath from "jmespath"; import * as grpc from "@grpc/grpc-js"; import * as protoLoader from "@grpc/proto-loader"; import WebSocket from "ws"; +import mongoose from "mongoose"; import { EnvConfig } from "@/domain/app-settings/app-settings.service.js"; import { SharedServices } from "@/config/services.shared.js"; @@ -132,6 +133,7 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig const worker = await DBQueueWorker.create( logger, + () => mongoose.connection.readyState === 1, jobsRepository, monitorsRepository, checksRepository, diff --git a/server/src/index.ts b/server/src/index.ts index 3b0ab6ba4..0c0a6233e 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -14,6 +14,7 @@ import Logger, { ILogger } from "@/utils/logger.js"; import { SettingsService } from "@/domain/app-settings/app-settings.service.js"; import { JobScheduler } from "@/worker/worker.job-scheduler.js"; import { IJobScheduler } from "@/worker/worker.interface.js"; +import { HealthServer } from "@/worker/worker.health-server.js"; const SERVICE_NAME = "Server"; let logger: ILogger; @@ -51,8 +52,10 @@ const startApp = async () => { // *********************** if (queueMode === "worker") { const { worker } = await buildWorker(shared, envSettings); + const healthServer = new HealthServer(logger, env.HEALTH_PORT, worker); + await healthServer.listen(); logger.info({ message: "Worker instance started. API will not be started", service: SERVICE_NAME }); - initShutdownListener(null, { worker, db: shared.db, logger }); + initShutdownListener(null, { worker, db: shared.db, logger, healthServer }); return; } diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index bf77a77c9..a25d96c0e 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -3,15 +3,17 @@ import type { ILogger } from "@/utils/logger.js"; import type { Server } from "http"; import { Mongoose } from "mongoose"; import { IJobScheduler } from "@/worker/worker.interface.js"; +import { IHealthServer } from "@/worker/worker.health-server.js"; type ShutdownTargets = { worker: IJobScheduler; db: IDb; logger: ILogger; + healthServer?: IHealthServer; }; -export const initShutdownListener = (server: Server | null, services: ShutdownTargets) => { +export const initShutdownListener = (server: Server | null, shutdownTargets: ShutdownTargets) => { const SERVICE_NAME = "Server"; - const logger = services.logger; + const logger = shutdownTargets.logger; let isShuttingDown = false; @@ -24,9 +26,10 @@ export const initShutdownListener = (server: Server | null, services: ShutdownTa try { server?.close(); - await services.worker.drain(); - await services.worker.shutdown(); - await services.db.disconnect(); + await shutdownTargets.worker.drain(); + await shutdownTargets.worker.shutdown(); + await shutdownTargets.healthServer?.close(); + await shutdownTargets.db.disconnect(); logger.info({ message: "Graceful shutdown complete" }); process.exit(0); } catch (error) { From e036c6077e1a86711196c07a4a2c75f02ba6f522 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 10:37:07 -0700 Subject: [PATCH 08/16] add tests --- .../test/unit/services/dbQueueWorker.test.ts | 69 +++++++++ .../test/unit/services/healthServer.test.ts | 136 ++++++++++++++++++ .../test/unit/services/jobScheduler.test.ts | 3 +- 3 files changed, 207 insertions(+), 1 deletion(-) create mode 100644 server/test/unit/services/healthServer.test.ts 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(); }); From 48ca64ec6ff5070902b0dca4e6d6e189f2f5d556 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:02:18 -0700 Subject: [PATCH 09/16] add observability repo method --- server/src/domain/jobs/job.repository.interface.ts | 1 + server/src/domain/jobs/job.repository.mongo.ts | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index 190323eed..d253523fa 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -72,4 +72,5 @@ export interface IJobsRepository { // ******************** findPage(pagination: JobPageQuery): Promise; findAll(): Promise; + countDueBacklog(now: number): Promise; // Scaling signal } diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index 89d8dfd04..60bd4cd77 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -301,6 +301,15 @@ class MongoJobsRepository implements IJobsRepository { const docs = await JobModel.find().lean(); return docs.map(this.toEntity); }; + + countDueBacklog = async (now: number) => { + return JobModel.countDocuments({ + type: "check", + isActive: true, + nextScheduledAt: { $lte: now }, + $or: [{ lockedUntil: null }, { lockedUntil: { $lt: now } }], // free or expired lock + }); + }; } export default MongoJobsRepository; From f43bf9ab3d9a6914730c56d0808a1029c78f25f5 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:02:38 -0700 Subject: [PATCH 10/16] add observability to worker and health server --- server/src/worker/worker.db-queue.ts | 13 +++++++++ server/src/worker/worker.health-server.ts | 33 ++++++++++++++++++++++- server/src/worker/worker.interface.ts | 2 ++ 3 files changed, 47 insertions(+), 1 deletion(-) diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index e0f168eb1..b7f971fd7 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -332,6 +332,10 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { await this.bufferService.shutdown(); // Flush buffers }; + // ************************ + // Observability + // ************************ + getHealth = () => { return { workerId: this.workerId, @@ -343,4 +347,13 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { inFlight: this.getInFlightCount(), }; }; + + countDueBacklog = () => { + return this.jobsRepository.countDueBacklog(Date.now()); + }; + + countAliveWorkers = async () => { + const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + return workers.length; + }; } diff --git a/server/src/worker/worker.health-server.ts b/server/src/worker/worker.health-server.ts index 55727f734..74aa9b786 100644 --- a/server/src/worker/worker.health-server.ts +++ b/server/src/worker/worker.health-server.ts @@ -4,6 +4,8 @@ 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 +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`; export interface IHealthServer { listen(): Promise; @@ -37,8 +39,37 @@ export class HealthServer implements IHealthServer { this.server = http.createServer(this.handle); } - private handle = (req: http.IncomingMessage, res: http.ServerResponse) => { + private metrics = async (): Promise => { + const health = this.worker.getHealth(); + const [dueBacklog, aliveWorkers] = await Promise.all([this.worker.countDueBacklog(), this.worker.countAliveWorkers()]); + return ( + gauge("checkmate_worker_jobs_in_flight", "Jobs currently being processed by this worker", health.inFlight) + + gauge("checkmate_worker_due_backlog", "Due check jobs not currently locked (cluster-wide)", dueBacklog) + + gauge("checkmate_worker_alive_total", "Queue members seen within the TTL window (cluster-wide)", aliveWorkers) + + gauge("checkmate_worker_draining", "1 if this worker is draining, else 0", health.draining ? 1 : 0) + ); + }; + + private handle = async (req: http.IncomingMessage, res: http.ServerResponse) => { const path = (req.url ?? "").split("?")[0]; // Strip query params + + // Metrics + if (path === "/metrics") { + try { + const body = await this.metrics(); + res.writeHead(200, { "Content-Type": CONTENT_TYPE_METRICS }); + res.end(body); + } catch (error: unknown) { + this.logger.warn({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: "metrics", + }); + res.writeHead(500).end(); + } + return; + } + if (path !== "/livez" && path !== "/readyz") { res.writeHead(404).end(); return; diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 6f7c57d07..d418e0263 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -85,4 +85,6 @@ export interface IQueueWorker extends IJobScheduler { readonly serviceName: string; init(): Promise; getHealth(): WorkerHealth; + countDueBacklog(): Promise; + countAliveWorkers(): Promise; } From b21ef60cdab78d097e2ece7c0b342e6f6b27b83e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:03:04 -0700 Subject: [PATCH 11/16] add a worker to the dev docker compose --- docker/dev/docker-compose.yaml | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/docker/dev/docker-compose.yaml b/docker/dev/docker-compose.yaml index a2c66c027..c5889ff35 100755 --- a/docker/dev/docker-compose.yaml +++ b/docker/dev/docker-compose.yaml @@ -20,9 +20,34 @@ services: - server.env depends_on: - mongodb + worker: + image: uptime_server:latest + restart: always + ports: + - "52346:52346" + env_file: + - server.env + environment: + QUEUE_MODE: "worker" + HEALTH_PORT: "52346" + healthcheck: + test: + [ + "CMD", + "node", + "-e", + "require('http').get('http://localhost:52346/livez',r=>process.exit(r.statusCode===200?0:1)).on('error',()=>process.exit(1))", + ] + interval: 15s + timeout: 3s + retries: 3 + depends_on: + - mongodb mongodb: image: uptime_mongo:latest restart: always + ports: + - "27017:27017" command: ["mongod", "--quiet", "--bind_ip_all"] volumes: - ./mongo/data:/data/db From ef353dcb360237f84d1072f5ab58a97c3e4eb1da Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:47:22 -0700 Subject: [PATCH 12/16] bugfix, move reconcile to job scheduler --- .../queue-workers/queue-worker.model.ts | 1 + .../queue-worker.repository.interface.ts | 2 +- .../queue-worker.repository.mongo.ts | 5 +- .../domain/queue-workers/queue-worker.type.ts | 5 +- server/src/index.ts | 3 +- server/src/worker/worker.db-queue.ts | 77 ++++--------------- server/src/worker/worker.health-server.ts | 2 +- server/src/worker/worker.interface.ts | 2 +- server/src/worker/worker.job-scheduler.ts | 62 ++++++++++++++- 9 files changed, 88 insertions(+), 71 deletions(-) diff --git a/server/src/domain/queue-workers/queue-worker.model.ts b/server/src/domain/queue-workers/queue-worker.model.ts index d1a7aa86b..0dd90b5c5 100644 --- a/server/src/domain/queue-workers/queue-worker.model.ts +++ b/server/src/domain/queue-workers/queue-worker.model.ts @@ -10,6 +10,7 @@ const QueueWorkerSchema = new Schema( { _id: { type: String, required: true }, mode: { type: String, enum: QueueModes, required: true }, + processesJobs: { type: Boolean, required: true }, lastSeenAt: { type: Date, default: Date.now, expires: WORKER_TTL_SECONDS }, }, { timestamps: true } diff --git a/server/src/domain/queue-workers/queue-worker.repository.interface.ts b/server/src/domain/queue-workers/queue-worker.repository.interface.ts index e4522c916..4da6f736e 100644 --- a/server/src/domain/queue-workers/queue-worker.repository.interface.ts +++ b/server/src/domain/queue-workers/queue-worker.repository.interface.ts @@ -2,7 +2,7 @@ import type { QueueMode } from "@/domain/app-settings/app-settings.type.js"; import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; export interface IQueueWorkersRepository { - upsert(workerId: string, mode: QueueMode): Promise; + upsert(workerId: string, mode: QueueMode, processesJobs: boolean): Promise; findRecent(maxAgeMs: number): Promise; deleteById(workerId: string): Promise; } diff --git a/server/src/domain/queue-workers/queue-worker.repository.mongo.ts b/server/src/domain/queue-workers/queue-worker.repository.mongo.ts index 35565bd03..da62d9850 100644 --- a/server/src/domain/queue-workers/queue-worker.repository.mongo.ts +++ b/server/src/domain/queue-workers/queue-worker.repository.mongo.ts @@ -11,12 +11,13 @@ class MongoQueueWorkersRepository implements IQueueWorkersRepository { return { workerId: doc._id, mode: doc.mode, + processesJobs: doc.processesJobs, lastSeenAt: doc.lastSeenAt.getTime(), }; }; - upsert = async (workerId: string, mode: QueueMode): Promise => { - await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true }); + upsert = async (workerId: string, mode: QueueMode, processesJobs: boolean): Promise => { + await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, processesJobs, lastSeenAt: new Date() } }, { upsert: true }); }; findRecent = async (maxAgeMs: number): Promise => { diff --git a/server/src/domain/queue-workers/queue-worker.type.ts b/server/src/domain/queue-workers/queue-worker.type.ts index a83c831ad..b750f77d7 100644 --- a/server/src/domain/queue-workers/queue-worker.type.ts +++ b/server/src/domain/queue-workers/queue-worker.type.ts @@ -1,11 +1,9 @@ import { type QueueMode } from "@/domain/app-settings/app-settings.type.js"; -// Heartbeat presence record for a scheduler instance. _id is the scheduler's -// workerId (hostname:pid:uuid). The TTL index on lastSeenAt garbage-collects -// records for instances that have stopped heartbeating. export interface QueueWorkerDocument { _id: string; mode: QueueMode; + processesJobs: boolean; lastSeenAt: Date; createdAt: Date; updatedAt: Date; @@ -14,5 +12,6 @@ export interface QueueWorkerDocument { export type QueueWorker = { workerId: string; // hostname:pid:uuid mode: QueueMode; + processesJobs: boolean; lastSeenAt: number; // epoch ms of the last heartbeat }; diff --git a/server/src/index.ts b/server/src/index.ts index 0c0a6233e..8614c12ac 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -75,7 +75,8 @@ const startApp = async () => { // Primary node does not process jobs, only need scheduler // *********************** else { - scheduler = new JobScheduler(shared.jobsRepository, shared.queueWorkersRepository, shared.workerId); + scheduler = new JobScheduler(shared.jobsRepository, shared.queueWorkersRepository, shared.monitorsRepository, queueMode, logger, shared.workerId); + await scheduler.init(); // register in the queue_workers registry + reconcile (seed the jobs collection) } const services = buildApi(shared, scheduler); diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index b7f971fd7..8c644eacc 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -1,6 +1,5 @@ -import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { IQueueWorker } from "@/worker/worker.interface.js"; -import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js"; +import { type Job, type JobType, LOCK_MS } from "@/domain/jobs/job.type.js"; import { ILogger } from "@/utils/logger.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; @@ -20,9 +19,7 @@ const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const POLL_MAX_MS = 5000; // back off poll interval to 5s if no jobs const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window -const HEARTBEAT_MS = WORKER_STALE_MS / 3; // Worker can miss two beats without being considered stale const LOCK_RENEW_MS = LOCK_MS / 3; // renew an in-flight job's lock at 1/3 the lease, so it survives two missed renewals -const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; const DRAIN_TIMEOUT_MS = 25_000; const DRAIN_POLL_MS = 200; @@ -48,11 +45,13 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { "cleanup-retention": 0, }; + private processesJobs: boolean = false; + constructor( - private logger: ILogger, + logger: ILogger, // forwarded to super private isDbConnected: () => boolean, - jobsRepository: IJobsRepository, // ← no modifier: forwarded to super - private monitorsRepository: IMonitorsRepository, + jobsRepository: IJobsRepository, // forwarded + monitorsRepository: IMonitorsRepository, // forwarded private checksRepository: IChecksRepository, private checkService: ICheckService, private bufferService: IBufferService, @@ -61,12 +60,13 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { private geoCheckPipeline: ICheckPipeline, private dispatcher: IReactorDispatcher, private helper: IWorkerHelper, - queueWorkersRepository: IQueueWorkersRepository, // ← no modifier: forwarded - private queueMode: QueueMode, - private queuePrimaryProcesses: boolean, + queueWorkersRepository: IQueueWorkersRepository, // forwarded + queueMode: QueueMode, // forwarded + queuePrimaryProcesses: boolean, // consumed immediately workerId: string ) { - super(jobsRepository, queueWorkersRepository, workerId); + super(jobsRepository, queueWorkersRepository, monitorsRepository, queueMode, logger, workerId); + this.processesJobs = queuePrimaryProcesses === true || queueMode === "worker"; } get serviceName() { @@ -115,15 +115,6 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { private getInFlightCount = () => Object.values(this.inFlight).reduce((sum, n) => sum + n, 0); - private toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({ - id: jobId(type, null), - type, - refId: null, - isActive: true, - nextScheduledAt: now, - intervalMs: CLEANUP_INTERVAL_MS, - }); - // ******************** // Stage 1: This can eventually be offloaded to a separate service // ******************** @@ -253,52 +244,16 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { // Wake a worker loop after adding a job that's due immediately. // Otherwise the loop could be delayed by POLL_MAX_MS if it is already fully backed off - // Seed queue - private reconcile = async () => { - const now = Date.now(); - const monitors = (await this.monitorsRepository.findAll()) ?? []; - for (const monitor of monitors) { - await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now)); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { - await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now)); - } - } - // Both cleanup jobs run immediately on every startup, then once a day - await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-orphaned", now)); - await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-retention", now)); - }; - - // Register worker - private heartbeat = async () => { - try { - await this.queueWorkersRepository.upsert(this.workerId, this.queueMode); - } catch (error: unknown) { - this.logger.warn({ - message: error instanceof Error ? error.message : String(error), - service: SERVICE_NAME, - method: "heartbeat", - }); - } - }; - - init = async () => { - this.stopped = false; - - await this.heartbeat(); - this.timers.set("heartbeat", setInterval(this.heartbeat, HEARTBEAT_MS)); - - if (this.queueMode === "primary") { - await this.reconcile(); - } - - if (this.queuePrimaryProcesses === true || this.queueMode === "worker") { + override async init() { + await super.init(); // heartbeat + (primary?) reconcile + if (this.processesJobs) { for (const type of Object.keys(this.inFlight) as JobType[]) { this.startLoop(type); } } this.initComplete = true; return true; - }; + } override flushQueues = async () => { await this.shutdown(); @@ -354,6 +309,6 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { countAliveWorkers = async () => { const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); - return workers.length; + return workers.filter((worker) => worker.processesJobs).length; }; } diff --git a/server/src/worker/worker.health-server.ts b/server/src/worker/worker.health-server.ts index 74aa9b786..b68579b80 100644 --- a/server/src/worker/worker.health-server.ts +++ b/server/src/worker/worker.health-server.ts @@ -45,7 +45,7 @@ export class HealthServer implements IHealthServer { return ( gauge("checkmate_worker_jobs_in_flight", "Jobs currently being processed by this worker", health.inFlight) + gauge("checkmate_worker_due_backlog", "Due check jobs not currently locked (cluster-wide)", dueBacklog) + - gauge("checkmate_worker_alive_total", "Queue members seen within the TTL window (cluster-wide)", aliveWorkers) + + gauge("checkmate_worker_alive_total", "Job-processing nodes seen within the TTL window (cluster-wide)", aliveWorkers) + gauge("checkmate_worker_draining", "1 if this worker is draining, else 0", health.draining ? 1 : 0) ); }; diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index d418e0263..98120fa47 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -77,13 +77,13 @@ export interface IJobScheduler { getMetrics(): Promise; getJobs(pagination: WorkerJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; + init(): Promise; drain(): Promise; shutdown(): Promise; } export interface IQueueWorker extends IJobScheduler { readonly serviceName: string; - init(): Promise; getHealth(): WorkerHealth; countDueBacklog(): Promise; countAliveWorkers(): Promise; diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts index a431143ac..0aa2e7a4b 100644 --- a/server/src/worker/worker.job-scheduler.ts +++ b/server/src/worker/worker.job-scheduler.ts @@ -1,20 +1,28 @@ import { IJobScheduler } from "@/worker/worker.interface.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js"; import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; import { WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; +import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; +import { ILogger } from "@/utils/logger.js"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const SERVICE_NAME = "JobScheduler"; const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window +const HEARTBEAT_MS = WORKER_STALE_MS / 3; // Worker can miss two beats without being considered stale +const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; export class JobScheduler implements IJobScheduler { static SERVICE_NAME = SERVICE_NAME; constructor( protected jobsRepository: IJobsRepository, protected queueWorkersRepository: IQueueWorkersRepository, + protected monitorsRepository: IMonitorsRepository, + protected queueMode: QueueMode, + protected logger: ILogger, protected readonly workerId: string // same id MongoJobsRepository stamps into lockedBy ) {} @@ -22,6 +30,7 @@ export class JobScheduler implements IJobScheduler { return JobScheduler.SERVICE_NAME; } + protected processesJobs = false; // scheduler-only by default; DBQueueWorker flips this on when it runs job loops protected stopped = false; protected draining = false; protected ticking: Partial> = {}; // Whether a loop is mid tick @@ -53,6 +62,15 @@ export class JobScheduler implements IJobScheduler { intervalMs: monitor.geoCheckInterval!, }); + protected toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({ + id: jobId(type, null), + type, + refId: null, + isActive: true, + nextScheduledAt: now, + intervalMs: CLEANUP_INTERVAL_MS, + }); + private toSummary = (job: Job): WorkerJobSummary => ({ monitorId: job.refId ?? job.id, monitorType: job.type, @@ -112,7 +130,8 @@ export class JobScheduler implements IJobScheduler { getMetrics = async (): Promise => { const rows = await this.jobsRepository.findAll(); - const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + const recent = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + const workers = recent.filter((worker) => worker.processesJobs); const now = Date.now(); return rows.reduce( (acc, job) => { @@ -155,6 +174,47 @@ export class JobScheduler implements IJobScheduler { return { success: true }; }; + // init is the scheduler's startup: register in the registry + (if primary) seed the queue. + // Subclasses override to ALSO start processing, calling super.init() first. + // Declared as a prototype method (not an arrow field) so subclasses can reach it via super.init(). + async init(): Promise { + this.stopped = false; + await this.heartbeat(); + this.timers.set("heartbeat", setInterval(this.heartbeat, HEARTBEAT_MS)); + if (this.queueMode === "primary") { + await this.reconcile(); + } + return true; + } + + // Seed queue + protected reconcile = async () => { + const now = Date.now(); + const monitors = (await this.monitorsRepository.findAll()) ?? []; + for (const monitor of monitors) { + await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now)); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now)); + } + } + // Both cleanup jobs run immediately on every startup, then once a day + await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-orphaned", now)); + await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-retention", now)); + }; + + // Register worker + protected heartbeat = async () => { + try { + await 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", + }); + } + }; + drain = async () => { this.stopped = true; this.draining = true; From 6e1c26b987aa40dbe3e8c2cf476b7ec40e5f87a3 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:55:28 -0700 Subject: [PATCH 13/16] remove shadowing var --- server/src/worker/worker.db-queue.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 8c644eacc..cdcb2de44 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -45,8 +45,6 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { "cleanup-retention": 0, }; - private processesJobs: boolean = false; - constructor( logger: ILogger, // forwarded to super private isDbConnected: () => boolean, From 9d969d67ca8d0d91643a5ae955b4e5ebe2c5431e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:55:40 -0700 Subject: [PATCH 14/16] udpate tests --- .../test/integration/jobsRepository.test.ts | 36 ++++++ .../integration/schedulerReconcile.test.ts | 107 ++++++++++++++++ .../test/unit/services/dbQueueWorker.test.ts | 35 +++++- .../test/unit/services/healthServer.test.ts | 47 ++++++- .../test/unit/services/jobScheduler.test.ts | 116 ++++++++++++++++-- 5 files changed, 324 insertions(+), 17 deletions(-) create mode 100644 server/test/integration/schedulerReconcile.test.ts diff --git a/server/test/integration/jobsRepository.test.ts b/server/test/integration/jobsRepository.test.ts index 02b0ecc5d..78ac17b64 100644 --- a/server/test/integration/jobsRepository.test.ts +++ b/server/test/integration/jobsRepository.test.ts @@ -497,4 +497,40 @@ describe("MongoJobsRepository", () => { expect(await repo.findAll()).toHaveLength(2); }); }); + + // ── countDueBacklog (scaling signal) ───────────────────────────────────────── + describe("countDueBacklog", () => { + it("counts a due, active, unlocked check job", async () => { + await seedJob(); + expect(await repo.countDueBacklog(NOW)).toBe(1); + }); + + it("counts a check job whose lock has expired (reclaimable work)", async () => { + await seedJob({ lockedBy: "dead-worker", lockedUntil: NOW - 1 }); + expect(await repo.countDueBacklog(NOW)).toBe(1); + }); + + it("excludes not-yet-due, inactive, live-locked, and non-check rows", async () => { + await seedJob({ _id: "check:due", refId: "due" }); // counts + await seedJob({ _id: "check:future", refId: "future", nextScheduledAt: NOW + 1 }); // not due + await seedJob({ _id: "check:paused", refId: "paused", isActive: false }); // inactive + await seedJob({ _id: "check:locked", refId: "locked", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // live lock + await seedJob({ _id: "geo:mon", refId: "mon", type: "geo-check" }); // wrong type + await seedJob({ _id: "evaluate:mon", refId: "mon", type: "evaluate" }); // wrong type + + expect(await repo.countDueBacklog(NOW)).toBe(1); // only check:due + }); + + it("counts every qualifying check row across monitors", async () => { + await seedJob({ _id: "check:a", refId: "a" }); + await seedJob({ _id: "check:b", refId: "b" }); + await seedJob({ _id: "check:c", refId: "c", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // excluded + expect(await repo.countDueBacklog(NOW)).toBe(2); + }); + + it("returns 0 when nothing is due", async () => { + await seedJob({ nextScheduledAt: NOW + 10_000 }); + expect(await repo.countDueBacklog(NOW)).toBe(0); + }); + }); }); diff --git a/server/test/integration/schedulerReconcile.test.ts b/server/test/integration/schedulerReconcile.test.ts new file mode 100644 index 000000000..8b1a48f79 --- /dev/null +++ b/server/test/integration/schedulerReconcile.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it, beforeAll, afterAll, beforeEach } from "@jest/globals"; +import mongoose from "mongoose"; +import { MongoMemoryServer } from "mongodb-memory-server"; +import { JobScheduler } from "../../src/worker/worker.job-scheduler.ts"; +import MongoJobsRepository from "../../src/domain/jobs/job.repository.mongo.ts"; +import MongoQueueWorkersRepository from "../../src/domain/queue-workers/queue-worker.repository.mongo.ts"; +import MongoMonitorsRepository from "../../src/domain/monitors/monitor.repository.mongo.ts"; +import JobModel, { type JobDocument } from "../../src/domain/jobs/job.model.ts"; +import { MonitorModel } from "../../src/domain/monitors/monitor.model.ts"; +import { QueueWorkerModel } from "../../src/domain/queue-workers/queue-worker.model.ts"; +import { createMockLogger } from "../helpers/createMockLogger.ts"; + +// ── Real-Mongo harness ───────────────────────────────────────────────────────── +// The bug: a scheduler-only primary (queueMode=primary, queuePrimaryProcesses=false) +// never seeded the jobs collection, so a fresh start with existing monitors left the +// workers idle. The fix moved reconcile/heartbeat onto the base JobScheduler.init(). +// This proves init() against a live mongod actually writes the expected job rows. + +let mongod: MongoMemoryServer; + +beforeAll(async () => { + mongod = await MongoMemoryServer.create(); + await mongoose.connect(mongod.getUri()); + await JobModel.init(); +}, 120_000); + +afterAll(async () => { + await mongoose.disconnect(); + await mongod.stop(); +}); + +beforeEach(async () => { + await JobModel.deleteMany({}); + await MonitorModel.collection.deleteMany({}); + await QueueWorkerModel.deleteMany({}); +}); + +const WORKER_ID = "scheduler-primary-1"; + +const seedMonitor = (overrides: Record = {}) => + MonitorModel.create({ + userId: new mongoose.Types.ObjectId(), + teamId: new mongoose.Types.ObjectId(), + name: "mon", + type: "http", + url: "https://example.com", + interval: 60_000, + isActive: true, + ...overrides, + }); + +const makeScheduler = () => + new JobScheduler( + new MongoJobsRepository(WORKER_ID), + new MongoQueueWorkersRepository(), + new MongoMonitorsRepository(), + "primary", + createMockLogger(), + WORKER_ID + ); + +const idsOf = async () => (await JobModel.find().lean()).map((j) => j._id); + +describe("JobScheduler.init (scheduler-only primary reconcile)", () => { + it("seeds a check row per monitor plus the two cleanup rows", async () => { + const m1 = await seedMonitor(); + const m2 = await seedMonitor({ name: "mon-2" }); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); // stop the heartbeat interval + + const ids = await idsOf(); + expect(ids).toContain(`check:${m1.id}`); + expect(ids).toContain(`check:${m2.id}`); + expect(ids).toContain("cleanup-orphaned"); + expect(ids).toContain("cleanup-retention"); + }); + + it("seeds a geo-check row only for geo-enabled monitors", async () => { + const geo = await seedMonitor({ name: "geo", geoCheckEnabled: true }); + const plain = await seedMonitor({ name: "plain", geoCheckEnabled: false }); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); + + const ids = await idsOf(); + expect(ids).toContain(`geo-check:${geo.id}`); + expect(ids).not.toContain(`geo-check:${plain.id}`); + }); + + it("registers the primary in the queue_workers registry", async () => { + await seedMonitor(); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); + + // shutdown deregisters, so assert mid-flight: re-init then read before shutdown. + const scheduler2 = makeScheduler(); + await scheduler2.init(); + const row = await QueueWorkerModel.findById(WORKER_ID).lean(); + expect(row?.mode).toBe("primary"); + await scheduler2.shutdown(); + }); +}); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index dd210b3e2..cabf90e31 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -79,6 +79,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses updateScheduleById: jest.fn().mockResolvedValue(true), findAll: jest.fn().mockResolvedValue([]), findPage: jest.fn().mockResolvedValue({ jobs: [], count: 0 }), + countDueBacklog: jest.fn().mockResolvedValue(0), }; const monitorsRepository = { findByIds: jest.fn().mockResolvedValue([makeMonitor()]), @@ -186,7 +187,7 @@ describe("DBQueueWorker", () => { describe("init", () => { it("registers the worker in the heartbeat registry on startup", async () => { const { mocks } = await start({ queueMode: "worker" }); - expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker"); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true); }); it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => { @@ -263,7 +264,7 @@ describe("DBQueueWorker", () => { active = worker; await jest.advanceTimersByTimeAsync(1); - expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker"); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true); }); }); @@ -768,6 +769,29 @@ describe("DBQueueWorker", () => { }); }); + // ── metrics accessors ─────────────────────────────────────────────────────────── + + describe("metrics accessors", () => { + it("countDueBacklog delegates to the repo with the current time", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + mocks.jobsRepository.countDueBacklog.mockResolvedValue(7); + + expect(await worker.countDueBacklog()).toBe(7); + expect(mocks.jobsRepository.countDueBacklog).toHaveBeenCalledWith(expect.any(Number)); + }); + + it("countAliveWorkers counts only recently-seen nodes that process jobs", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + mocks.queueWorkersRepository.findRecent.mockResolvedValue([ + { processesJobs: true }, + { processesJobs: true }, + { processesJobs: false }, // scheduler-only primary: alive but not a processor + ]); + + expect(await worker.countAliveWorkers()).toBe(2); + }); + }); + // ── flushQueues ─────────────────────────────────────────────────────────────── describe("flushQueues", () => { @@ -799,7 +823,10 @@ describe("DBQueueWorker", () => { const { worker, mocks } = createWorker(); active = worker; mocks.jobsRepository.findAll.mockResolvedValue(rows); - mocks.queueWorkersRepository.findRecent.mockResolvedValue([{ workerId: "w" }]); + mocks.queueWorkersRepository.findRecent.mockResolvedValue([ + { workerId: "w", processesJobs: true }, + { workerId: "scheduler", processesJobs: false }, // scheduler-only primary excluded from metrics + ]); const metrics = await worker.getMetrics(); @@ -810,7 +837,7 @@ describe("DBQueueWorker", () => { expect(metrics.failingJobs).toBe(1); expect(metrics.jobsWithFailures).toHaveLength(1); expect(metrics.jobsWithFailures[0]).toEqual(expect.objectContaining({ failCount: 3, failReason: "boom" })); - expect(metrics.workers).toEqual([{ workerId: "w" }]); + expect(metrics.workers).toEqual([{ workerId: "w", processesJobs: true }]); }); }); diff --git a/server/test/unit/services/healthServer.test.ts b/server/test/unit/services/healthServer.test.ts index 208cd0902..1ca62d4bb 100644 --- a/server/test/unit/services/healthServer.test.ts +++ b/server/test/unit/services/healthServer.test.ts @@ -26,13 +26,13 @@ const freePort = (): Promise => }); }); -const get = (port: number, path: string): Promise<{ status: number; body: string }> => +const get = (port: number, path: string): Promise<{ status: number; body: string; contentType: 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 })); + res.on("end", () => resolve({ status: res.statusCode ?? 0, body, contentType: String(res.headers["content-type"] ?? "") })); }) .on("error", reject); }); @@ -58,9 +58,16 @@ describe("HealthServer", () => { active = null; }); - const setup = async (health: WorkerHealth) => { + // The metrics endpoint also calls countDueBacklog()/countAliveWorkers(); default them + // to 0 and let tests override (including to a rejection, to exercise the 500 path). + const setup = async (health: WorkerHealth, workerOverrides: Record = {}) => { const port = await freePort(); - const worker = { getHealth: () => health } as any; + const worker = { + getHealth: () => health, + countDueBacklog: async () => 0, + countAliveWorkers: async () => 0, + ...workerOverrides, + } as any; const server = new HealthServer(createMockLogger(), String(port), worker); await server.listen(); active = server; @@ -133,4 +140,36 @@ describe("HealthServer", () => { expect((await get(port, "/livez?probe=1")).status).toBe(200); }); }); + + describe("/metrics", () => { + it("200 with prometheus content-type and all four gauges", async () => { + const { port } = await setup(makeHealth({ inFlight: 3, draining: false }), { + countDueBacklog: async () => 5, + countAliveWorkers: async () => 2, + }); + const res = await get(port, "/metrics"); + + expect(res.status).toBe(200); + expect(res.contentType).toMatch(/text\/plain/); + expect(res.body).toContain("# TYPE checkmate_worker_due_backlog gauge"); + expect(res.body).toContain("checkmate_worker_jobs_in_flight 3"); + expect(res.body).toContain("checkmate_worker_due_backlog 5"); + expect(res.body).toContain("checkmate_worker_alive_total 2"); + expect(res.body).toContain("checkmate_worker_draining 0"); + }); + + it("reports draining as 1 when the worker is draining", async () => { + const { port } = await setup(makeHealth({ draining: true })); + expect((await get(port, "/metrics")).body).toContain("checkmate_worker_draining 1"); + }); + + it("500 when a metrics query fails", async () => { + const { port } = await setup(makeHealth(), { + countDueBacklog: async () => { + throw new Error("db down"); + }, + }); + expect((await get(port, "/metrics")).status).toBe(500); + }); + }); }); diff --git a/server/test/unit/services/jobScheduler.test.ts b/server/test/unit/services/jobScheduler.test.ts index dceb8760c..90d7a2b99 100644 --- a/server/test/unit/services/jobScheduler.test.ts +++ b/server/test/unit/services/jobScheduler.test.ts @@ -1,26 +1,124 @@ -import { describe, expect, it, jest } from "@jest/globals"; +import { describe, expect, it, jest, beforeEach, afterEach } from "@jest/globals"; import { JobScheduler } from "../../../src/worker/worker.job-scheduler.ts"; +import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; +import type { QueueMode } from "../../../src/domain/app-settings/app-settings.type.ts"; import { createMockLogger } from "../../helpers/createMockLogger.ts"; // ── Notes ────────────────────────────────────────────────────────────────────── // // JobScheduler is the scheduler-only primary base class: it owns the queue -// registry/heartbeat but processes no jobs and has no buffer. drain() is therefore -// a no-op beyond setting the stop flag — it must NOT tear down (clear timers or -// deregister the worker); that stays in shutdown(). The real draining (wait for -// in-flight, flush buffer) lives in DBQueueWorker and is covered separately. +// registry/heartbeat AND, on a primary, the reconcile that seeds the jobs +// collection — scheduling is its job whether or not it also processes. It runs no +// polling loops and has no buffer, so drain() is a no-op beyond setting the stop +// flag — it must NOT tear down (clear timers or deregister the worker); that stays +// in shutdown(). The processing tier (poll loops, buffer flush) lives in +// DBQueueWorker and is covered separately. -const createScheduler = () => { - const jobsRepository = {} as any; +const makeMonitor = (overrides?: Partial): Monitor => + ({ + id: "m1", + teamId: "team", + type: "http", + interval: 60000, + status: "up", + isActive: true, + geoCheckEnabled: false, + geoCheckInterval: 300000, + lastEvaluatedAt: 0, + ...overrides, + }) as Monitor; + +const createScheduler = (overrides?: { queueMode?: QueueMode; mocks?: Record }) => { + const jobsRepository = { + upsertJob: jest.fn().mockResolvedValue(true), + upsertCleanupJob: jest.fn().mockResolvedValue(true), + }; const queueWorkersRepository = { + upsert: jest.fn().mockResolvedValue(undefined), deleteById: jest.fn().mockResolvedValue(undefined), }; + const monitorsRepository = { + findAll: jest.fn().mockResolvedValue([]), + }; const logger = createMockLogger(); - const scheduler = new JobScheduler(jobsRepository, queueWorkersRepository as any, "worker-1"); - return { scheduler, mocks: { jobsRepository, queueWorkersRepository, logger } }; + const mocks = { jobsRepository, queueWorkersRepository, monitorsRepository, logger, ...overrides?.mocks }; + const scheduler = new JobScheduler( + mocks.jobsRepository as any, + mocks.queueWorkersRepository as any, + mocks.monitorsRepository as any, + overrides?.queueMode ?? "primary", + mocks.logger as any, + "worker-1" + ); + return { scheduler, mocks }; }; describe("JobScheduler", () => { + describe("init", () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(async () => { + jest.clearAllTimers(); + jest.useRealTimers(); + }); + + it("registers the worker in the heartbeat registry on startup", async () => { + const { scheduler, mocks } = createScheduler({ queueMode: "primary" }); + await scheduler.init(); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "primary", false); + }); + + it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "primary", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor()]) } }, + }); + await scheduler.init(); + + const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type); + expect(seededTypes).toContain("check"); + const cleanupTypes = mocks.jobsRepository.upsertCleanupJob.mock.calls.map((c: any[]) => c[0].type); + expect(cleanupTypes).toContain("cleanup-orphaned"); + expect(cleanupTypes).toContain("cleanup-retention"); + }); + + it("primary mode also seeds a geo-check row for geo-enabled monitors", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "primary", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor({ geoCheckEnabled: true })]) } }, + }); + await scheduler.init(); + + const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type); + expect(seededTypes).toContain("geo-check"); + }); + + it("worker mode heartbeats but does not reconcile (no monitor reads, no seeds)", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "worker", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor()]) } }, + }); + await scheduler.init(); + + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", false); + expect(mocks.monitorsRepository.findAll).not.toHaveBeenCalled(); + expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled(); + expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled(); + }); + + it("arms a heartbeat interval timer that shutdown() clears and deregisters", async () => { + const { scheduler, mocks } = createScheduler({ queueMode: "primary" }); + await scheduler.init(); + expect((scheduler as any).timers.has("heartbeat")).toBe(true); + + await scheduler.shutdown(); + expect((scheduler as any).timers.has("heartbeat")).toBe(false); + expect(mocks.queueWorkersRepository.deleteById).toHaveBeenCalledWith("worker-1"); + }); + }); + describe("drain", () => { it("sets the stop and draining flags without tearing down (no deregister)", async () => { const { scheduler, mocks } = createScheduler(); From d2d0fb788e3a85ef3107fe4931d94546edcd337f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 12:29:21 -0700 Subject: [PATCH 15/16] resolve shutdown race --- docker/dev/docker-compose.yaml | 1 + server/src/worker/worker.health-server.ts | 28 +++++++++++++++++-- server/src/worker/worker.job-scheduler.ts | 12 +++++++- .../test/unit/services/healthServer.test.ts | 24 ++++++++++++++++ 4 files changed, 62 insertions(+), 3 deletions(-) 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())); + } + }); + }); }); From 3f5707721f2272ef3e6095e509b09b3b53f8b0c0 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 29 Jun 2026 09:12:42 -0700 Subject: [PATCH 16/16] fix logging on server startup --- server/src/index.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/index.ts b/server/src/index.ts index 8614c12ac..44536ce8b 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -37,7 +37,7 @@ const startApp = async () => { }); logger.info({ - message: `Process: ${queuePrimaryProcesses}`, + message: `Process jobs: ${queueMode === "worker" ? true : queuePrimaryProcesses}`, service: SERVICE_NAME, method: "startApp", });