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;