add a getHealth method to interface, stub out implementation

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<JobType, number> = {
 	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
+		};
+	};
 }

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