Merge pull request #3742 from bluewave-labs/feat/worker-health

Feat/worker health

docker/dev/docker-compose.yaml

@@ -20,9 +20,35 @@ 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"
+    stop_grace_period: 30s
+    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

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"),
 

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,10 +133,12 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig
 
 	const worker = await DBQueueWorker.create(
 		logger,
+		() => mongoose.connection.readyState === 1,
 		jobsRepository,
 		monitorsRepository,
 		checksRepository,
 		checkService,
+		bufferService,
 		checkProducer,
 		checkEvaluator,
 		geoCheckPipeline,

server/src/domain/jobs/job.repository.interface.ts

@@ -72,4 +72,5 @@ export interface IJobsRepository {
 	// ********************
 	findPage(pagination: JobPageQuery): Promise<JobPage>;
 	findAll(): Promise<Job[]>;
+	countDueBacklog(now: number): Promise<number>; // Scaling signal
 }

server/src/domain/jobs/job.repository.mongo.ts

@@ -301,6 +301,15 @@ class MongoJobsRepository implements IJobsRepository {
 		const docs = await JobModel.find().lean<JobDocument[]>();
 		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;

server/src/domain/queue-workers/queue-worker.model.ts

@@ -10,6 +10,7 @@ const QueueWorkerSchema = new Schema<QueueWorkerDocument>(
 	{
 		_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 }

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<void>;
+	upsert(workerId: string, mode: QueueMode, processesJobs: boolean): Promise<void>;
 	findRecent(maxAgeMs: number): Promise<QueueWorker[]>;
 	deleteById(workerId: string): Promise<void>;
 }

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<void> => {
-		await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true });
+	upsert = async (workerId: string, mode: QueueMode, processesJobs: boolean): Promise<void> => {
+		await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, processesJobs, lastSeenAt: new Date() } }, { upsert: true });
 	};
 
 	findRecent = async (maxAgeMs: number): Promise<QueueWorker[]> => {

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
 };

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;
 
@@ -36,7 +37,7 @@ const startApp = async () => {
 	});
 
 	logger.info({
-		message: `Process: ${queuePrimaryProcesses}`,
+		message: `Process jobs: ${queueMode === "worker" ? true : queuePrimaryProcesses}`,
 		service: SERVICE_NAME,
 		method: "startApp",
 	});
@@ -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;
 	}
 
@@ -72,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);

server/src/service/bufferService.ts

@@ -13,6 +13,7 @@ export interface IBufferService {
 	scheduleNextFlush(): void;
 	flushBuffer(): Promise<void>;
 	flushGeoBuffer(): Promise<void>;
+	shutdown(): Promise<void>;
 }
 
 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();
+	}
 }

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<Mongoose>;
 	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,8 +26,10 @@ export const initShutdownListener = (server: Server | null, services: ShutdownTa
 
 		try {
 			server?.close();
-			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) {

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";
@@ -15,13 +14,15 @@ 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
 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;
+
 const CONCURRENCY: Record<JobType, number> = {
 	check: 500,
 	"geo-check": 20,
@@ -32,6 +33,10 @@ const CONCURRENCY: Record<JobType, number> = {
 
 export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	static SERVICE_NAME = SERVICE_NAME;
+
+	private initComplete: boolean = false;
+	private lastTickAt: number | null = null;
+
 	private inFlight: Record<JobType, number> = {
 		check: 0,
 		"geo-check": 0,
@@ -41,22 +46,25 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	};
 
 	constructor(
-		private logger: ILogger,
-		jobsRepository: IJobsRepository, // ← no modifier: forwarded to super
-		private monitorsRepository: IMonitorsRepository,
+		logger: ILogger, // forwarded to super
+		private isDbConnected: () => boolean,
+		jobsRepository: IJobsRepository, // forwarded
+		monitorsRepository: IMonitorsRepository, // forwarded
 		private checksRepository: IChecksRepository,
 		private checkService: ICheckService,
+		private bufferService: IBufferService,
 		private checkProducer: ICheckProducer,
 		private checkEvaluator: ICheckEvaluator,
 		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() {
@@ -65,10 +73,12 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 
 	static async create(
 		logger: ILogger,
+		isDbConnected: () => boolean,
 		jobsRepository: IJobsRepository,
 		monitorsRepository: IMonitorsRepository,
 		checksRepository: IChecksRepository,
 		checkService: ICheckService,
+		bufferService: IBufferService,
 		checkProducer: ICheckProducer,
 		checkEvaluator: ICheckEvaluator,
 		geoCheckPipeline: ICheckPipeline,
@@ -81,10 +91,12 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	): Promise<DBQueueWorker> {
 		const instance = new DBQueueWorker(
 			logger,
+			isDbConnected,
 			jobsRepository,
 			monitorsRepository,
 			checksRepository,
 			checkService,
+			bufferService,
 			checkProducer,
 			checkEvaluator,
 			geoCheckPipeline,
@@ -99,14 +111,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 		return instance;
 	}
 
-	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,
-	});
+	private getInFlightCount = () => Object.values(this.inFlight).reduce((sum, n) => sum + n, 0);
 
 	// ********************
 	// Stage 1:  This can eventually be offloaded to a separate service
@@ -204,6 +209,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 {
@@ -236,55 +242,71 @@ 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();
 		const ok = await this.init();
 		return { success: ok };
 	};
+
+	override drain = async () => {
+		this.draining = true;
+		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
+	};
+
+	// ************************
+	// Observability
+	// ************************
+
+	getHealth = () => {
+		return {
+			workerId: this.workerId,
+			mode: this.queueMode,
+			dbConnected: this.isDbConnected(),
+			initComplete: this.initComplete,
+			draining: this.draining,
+			lastTickAt: this.lastTickAt,
+			inFlight: this.getInFlightCount(),
+		};
+	};
+
+	countDueBacklog = () => {
+		return this.jobsRepository.countDueBacklog(Date.now());
+	};
+
+	countAliveWorkers = async () => {
+		const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS);
+		return workers.filter((worker) => worker.processesJobs).length;
+	};
 }

server/src/worker/worker.health-server.ts

@@ -0,0 +1,129 @@
+import http from "http";
+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`;
+
+export interface IHealthServer {
+	listen(): Promise<void>;
+	close(): Promise<void>;
+	address(): ReturnType<http.Server["address"]>;
+}
+
+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 metrics = async (): Promise<string> => {
+		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", "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)
+		);
+	};
+
+	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;
+		}
+
+		// 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<void>((resolve, reject) => {
+			let port = Number(this.port) || DEFAULT_HEALTH_PORT;
+			const lastPort = port + MAX_PORT_ATTEMPTS;
+
+			const onError = (error: NodeJS.ErrnoException) => {
+				if (error.code === "EADDRINUSE" && port < lastPort) {
+					this.logger.warn({
+						message: `Health port ${port} in use, retrying on ${port + 1}`,
+						service: SERVICE_NAME,
+						method: "listen",
+					});
+					port++;
+					this.server.listen(port);
+					return;
+				}
+				this.server.removeListener("error", onError);
+				reject(error);
+			};
+
+			this.server.on("error", onError);
+			this.server.listen(port, () => {
+				this.server.removeListener("error", onError);
+				this.logger.info({
+					message: `Health server listening on ${port}`,
+					service: SERVICE_NAME,
+					method: "listen",
+				});
+				resolve();
+			});
+		});
+	};
+
+	address = () => this.server.address();
+
+	close = () => {
+		return new Promise<void>((resolve) => {
+			this.server.close(() => resolve());
+		});
+	};
+}

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<void>;
@@ -67,10 +77,14 @@ export interface IJobScheduler {
 	getMetrics(): Promise<WorkerMetrics>;
 	getJobs(pagination: WorkerJobsPagination): Promise<WorkerJobsPage>;
 	flushQueues(): Promise<{ success: boolean }>;
+	init(): Promise<boolean>;
+	drain(): Promise<void>;
 	shutdown(): Promise<void>;
 }
 
 export interface IQueueWorker extends IJobScheduler {
 	readonly serviceName: string;
-	init(): Promise<boolean>;
+	getHealth(): WorkerHealth;
+	countDueBacklog(): Promise<number>;
+	countAliveWorkers(): Promise<number>;
 }

server/src/worker/worker.job-scheduler.ts

@@ -1,28 +1,39 @@
 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
 	) {}
 
 	get serviceName() {
 		return JobScheduler.SERVICE_NAME;
 	}
+	private heartbeatInFlight: Promise<void> = Promise.resolve();
 
+	protected processesJobs = false; // scheduler-only by default; DBQueueWorker flips this on when it runs job loops
 	protected stopped = false;
+	protected draining = false;
 	protected ticking: Partial<Record<JobType, boolean>> = {}; // Whether a loop is mid tick
 	protected tickFns = new Map<JobType, () => void>(); // Store tick fns so we can run them immediately on wake
 	protected timers = new Map<JobType | "heartbeat", NodeJS.Timeout>();
@@ -52,6 +63,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,
@@ -111,7 +131,8 @@ export class JobScheduler implements IJobScheduler {
 
 	getMetrics = async (): Promise<WorkerMetrics> => {
 		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<WorkerMetrics>(
 			(acc, job) => {
@@ -154,6 +175,60 @@ 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<boolean> {
+		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 {
+			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),
+				service: SERVICE_NAME,
+				method: "heartbeat",
+			});
+		}
+	};
+
+	drain = async () => {
+		this.stopped = true;
+		this.draining = true;
+	};
+
 	shutdown = async () => {
 		this.stopped = true;
 		// Stop all timers
@@ -161,6 +236,7 @@ export class JobScheduler implements IJobScheduler {
 			clearTimeout(timer);
 		}
 		this.timers.clear(); // Clear the map out
+		await this.heartbeatInFlight; // Wait for in flight jobs
 		//  Remove workers from registry
 		await this.queueWorkersRepository.deleteById(this.workerId);
 	};

server/test/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);
+		});
+	});
 });

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<string, unknown> = {}) =>
+	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<JobDocument[]>()).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();
+	});
+});

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();
+		});
+	});
 });

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 = <T>() => {
 	let resolve!: (value: T) => void;
@@ -77,6 +79,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 		updateScheduleById: jest.fn<any>().mockResolvedValue(true),
 		findAll: jest.fn<any>().mockResolvedValue([]),
 		findPage: jest.fn<any>().mockResolvedValue({ jobs: [], count: 0 }),
+		countDueBacklog: jest.fn<any>().mockResolvedValue(0),
 	};
 	const monitorsRepository = {
 		findByIds: jest.fn<any>().mockResolvedValue([makeMonitor()]),
@@ -91,6 +94,14 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 		toStatusResponse: jest.fn<any>().mockReturnValue({ status: "up" }),
 		toLastEvaluatedAt: jest.fn<any>().mockReturnValue(12345),
 	};
+	const bufferService = {
+		addToBuffer: jest.fn<any>(),
+		addGeoCheckToBuffer: jest.fn<any>(),
+		scheduleNextFlush: jest.fn<any>(),
+		flushBuffer: jest.fn<any>().mockResolvedValue(undefined),
+		flushGeoBuffer: jest.fn<any>().mockResolvedValue(undefined),
+		shutdown: jest.fn<any>().mockResolvedValue(undefined),
+	};
 	const checkProducer = {
 		produce: jest.fn<any>().mockResolvedValue({ status: { status: "up" }, check: { id: "c1" } }),
 	};
@@ -109,12 +120,15 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 		findRecent: jest.fn<any>().mockResolvedValue([]),
 	};
 	const logger = createMockLogger();
+	const isDbConnected = jest.fn<any>(() => true);
 
 	const mocks = {
+		isDbConnected,
 		jobsRepository,
 		monitorsRepository,
 		checksRepository,
 		checkService,
+		bufferService,
 		checkProducer,
 		checkEvaluator,
 		geoCheckPipeline,
@@ -127,10 +141,12 @@ 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,
 		mocks.checkService as any,
+		mocks.bufferService as any,
 		mocks.checkProducer as any,
 		mocks.checkEvaluator as any,
 		mocks.geoCheckPipeline as any,
@@ -171,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 () => {
@@ -229,10 +245,12 @@ 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,
 				mocks.checkService as any,
+				mocks.bufferService as any,
 				mocks.checkProducer as any,
 				mocks.checkEvaluator as any,
 				mocks.geoCheckPipeline as any,
@@ -246,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);
 		});
 	});
 
@@ -619,6 +637,161 @@ describe("DBQueueWorker", () => {
 		});
 	});
 
+	// ── drain ─────────────────────────────────────────────────────────────────────
+
+	describe("drain", () => {
+		// Hand the "check" loop exactly one job, then nothing.
+		const claimOnce = (job: Job) => {
+			let handed = false;
+			return jest.fn<any>(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<any>().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<any>().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
+		});
+	});
+
+	// ── 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<any>(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<any>().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);
+		});
+	});
+
+	// ── 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", () => {
@@ -650,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();
 
@@ -661,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 }]);
 		});
 	});
 

server/test/unit/services/healthServer.test.ts

@@ -0,0 +1,199 @@
+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<number> =>
+	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; 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, contentType: String(res.headers["content-type"] ?? "") }));
+			})
+			.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>): 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;
+	});
+
+	// 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<string, unknown> = {}) => {
+		const port = await freePort();
+		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;
+		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);
+		});
+	});
+
+	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);
+		});
+	});
+
+	describe("listen", () => {
+		it("walks to the next port when the configured one is in use", async () => {
+			const base = await freePort();
+			// Hold the base port so the health server's first bind hits EADDRINUSE.
+			const blocker = net.createServer();
+			await new Promise<void>((resolve) => blocker.listen(base, resolve));
+
+			try {
+				const worker = { getHealth: () => makeHealth(), countDueBacklog: async () => 0, countAliveWorkers: async () => 0 } as any;
+				const server = new HealthServer(createMockLogger(), String(base), worker);
+				await server.listen();
+				active = server;
+
+				// base is still held by the blocker, so the health server retried onto a higher port and is reachable there.
+				const addr = server.address();
+				const bound = typeof addr === "object" && addr ? addr.port : 0;
+				expect(bound).toBeGreaterThan(base);
+				expect((await get(bound, "/livez")).status).toBe(200);
+			} finally {
+				await new Promise<void>((resolve) => blocker.close(() => resolve()));
+			}
+		});
+	});
+});

server/test/unit/services/jobScheduler.test.ts

@@ -0,0 +1,145 @@
+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 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 makeMonitor = (overrides?: Partial<Monitor>): 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<string, any> }) => {
+	const jobsRepository = {
+		upsertJob: jest.fn<any>().mockResolvedValue(true),
+		upsertCleanupJob: jest.fn<any>().mockResolvedValue(true),
+	};
+	const queueWorkersRepository = {
+		upsert: jest.fn<any>().mockResolvedValue(undefined),
+		deleteById: jest.fn<any>().mockResolvedValue(undefined),
+	};
+	const monitorsRepository = {
+		findAll: jest.fn<any>().mockResolvedValue([]),
+	};
+	const logger = createMockLogger();
+	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<any>().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<any>().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<any>().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();
+
+			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();
+		});
+
+		it("stops wake() from arming a tick once drained", async () => {
+			const { scheduler } = createScheduler();
+			const tick = jest.fn<any>();
+			(scheduler as any).tickFns.set("check", tick);
+
+			await scheduler.drain();
+			scheduler.wake("check");
+
+			expect((scheduler as any).timers.has("check")).toBe(false);
+		});
+	});
+});