bugfix, move reconcile to job scheduler

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

@@ -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);

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

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

server/src/worker/worker.interface.ts

@@ -77,13 +77,13 @@ 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,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<Record<JobType, boolean>> = {}; // 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<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) => {
@@ -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<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 {
+			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;