diff --git a/server/src/service/infrastructure/JobQueues/IJobQueue.ts b/server/src/service/infrastructure/JobQueues/IJobQueue.ts index a97c8eb6a..371a7c74a 100644 --- a/server/src/service/infrastructure/JobQueues/IJobQueue.ts +++ b/server/src/service/infrastructure/JobQueues/IJobQueue.ts @@ -1,4 +1,5 @@ import { Monitor } from "@/types/monitor.js"; +import { QueueWorker } from "@/types/settings.js"; export type QueueJobFailure = { monitorId: string | number; @@ -16,6 +17,7 @@ export type QueueMetrics = { jobsWithFailures: QueueJobFailure[]; totalRuns: number; totalFailures: number; + workers: QueueWorker[]; }; export type QueueJobSummary = { @@ -26,6 +28,8 @@ export type QueueJobSummary = { monitorGeoInterval: number | null; monitorActive: boolean | null; active: boolean; + lockedBy: string | null; + lockedUntil: number | null; lockedAt: number | null; runCount: number; failCount: number; diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 80655ab4c..cedf29901 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -1,4 +1,4 @@ -import { IMonitorsRepository } from "@/repositories/index.js"; +import { IMonitorsRepository, IQueueWorkersRepository } from "@/repositories/index.js"; import { ILogger } from "@/utils/logger.js"; import { MongoStore, Scheduler, IJob } from "less-simple-scheduler"; import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; @@ -8,6 +8,12 @@ import { type QueueMode } from "@/types/settings.js"; import { EnvConfig } from "@/service/system/settingsService.js"; const SERVICE_NAME = "JobQueue"; +// How long after its last heartbeat a worker is still considered alive. The +// model's TTL index uses the same window to garbage-collect stale records. +// Heartbeats are driven by the scheduler's `scheduler:heartbeat` event (emitted +// every lockMs/3, so ~5s with the default lockMs), comfortably inside this TTL. +const WORKER_TTL_MS = 30_000; + // For discriminating job types type MonitorJob = Omit & { data: Monitor }; const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); @@ -19,12 +25,21 @@ export class LessSimpleQueue implements IJobQueue { private logger: ILogger; private helper: IQueueHelper; private monitorsRepository: IMonitorsRepository; + private workersRepository: IQueueWorkersRepository; private readonly scheduler: Scheduler; + private heartbeatListener: ((workerId: string) => void) | null = null; - constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { + constructor( + logger: ILogger, + helper: IQueueHelper, + monitorsRepository: IMonitorsRepository, + workersRepository: IQueueWorkersRepository, + scheduler: Scheduler + ) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; + this.workersRepository = workersRepository; this.scheduler = scheduler; this.registerListeners(); } @@ -108,19 +123,45 @@ export class LessSimpleQueue implements IJobQueue { }); }; - static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig, queueMode: QueueMode) { + static async create( + logger: ILogger, + helper: IQueueHelper, + monitorsRepository: IMonitorsRepository, + workersRepository: IQueueWorkersRepository, + envSettings: EnvConfig, + queueMode: QueueMode + ) { const store = new MongoStore({ url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db", }); const scheduler = new Scheduler(store, { concurrency: 50, }); - const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler); await instance.init(queueMode); return instance; } + private startHeartbeat = (queueMode: QueueMode) => { + const beat = ( + workerId: string // This is the heartbeat function that registers workers + ) => + this.workersRepository.upsert(workerId, queueMode).catch((error: unknown) => { + this.logger.warn({ + message: `Worker heartbeat failed: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + method: "startHeartbeat", + }); + }); + beat(this.scheduler.workerId); // Beat immediately to register + if (this.heartbeatListener) { + this.scheduler.off("scheduler:heartbeat", this.heartbeatListener); // remove old listeners + } + this.heartbeatListener = beat; /// Keep a reference to the listener + this.scheduler.on("scheduler:heartbeat", beat); // beat on every heartbeat + }; + init = async (queueMode: QueueMode = "primary") => { try { await this.scheduler.start(); @@ -129,6 +170,9 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob()); this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob()); + // Register in the worker registry + this.startHeartbeat(queueMode); + // If this is a worker, return early - primary will populate queue if (queueMode === "worker") { this.logger.info({ @@ -284,11 +328,26 @@ export class LessSimpleQueue implements IJobQueue { }; shutdown = async () => { + // Stop beating before we remove this worker, so a heartbeat firing between + // the remove and the scheduler stopping can't re-insert it. + if (this.heartbeatListener) { + this.scheduler.off("scheduler:heartbeat", this.heartbeatListener); + this.heartbeatListener = null; + } + await this.workersRepository.deleteById(this.scheduler.workerId).catch((error: unknown) => { + this.logger.warn({ + message: `Failed to deregister worker on shutdown: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + method: "shutdown", + }); + }); this.scheduler.stop(); }; getMetrics = async () => { const jobs = await this.scheduler.getJobs(); + const now = Date.now(); + const metrics = jobs.reduce( (acc, job) => { const runCount = job.runCount ?? 0; @@ -302,7 +361,8 @@ export class LessSimpleQueue implements IJobQueue { acc.failingJobs++; } - if (job.lockedAt) { + // A job currently held by a live lock is in flight on some worker + if (job.lockedBy && job.lockedUntil && job.lockedUntil > now) { acc.activeJobs++; } @@ -326,8 +386,10 @@ export class LessSimpleQueue implements IJobQueue { jobsWithFailures: [], totalRuns: 0, totalFailures: 0, + workers: [], } ); + metrics.workers = await this.workersRepository.findRecent(WORKER_TTL_MS); return metrics; }; @@ -345,6 +407,8 @@ export class LessSimpleQueue implements IJobQueue { monitorActive: monitor?.isActive ?? null, active: job.active, repeat: job.repeat ?? null, + lockedBy: job.lockedBy ?? null, + lockedUntil: job.lockedUntil ?? null, lockedAt: job.lockedAt ?? null, runCount: job.runCount ?? 0, failCount: job.failCount ?? 0, diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index df265cf71..353a7ec7c 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -354,6 +354,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { }>, totalRuns: 0, totalFailures: 0, + workers: [], } ); return metrics; @@ -371,6 +372,8 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { monitorActive: job.data?.isActive ?? null, active: job.active, repeat: job.repeat ?? null, + lockedBy: null, + lockedUntil: null, lockedAt: job.lockedAt ?? null, runCount: job.runCount ?? 0, failCount: job.failCount ?? 0,