From 60356a95fe0060f014adf1941315646e70584ae7 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:44:23 -0700 Subject: [PATCH 01/13] bump less-simple-scheduler --- server/package-lock.json | 8 ++++---- server/package.json | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/server/package-lock.json b/server/package-lock.json index 131589613..be543815d 100644 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -31,7 +31,7 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", - "less-simple-scheduler": "2.2.0", + "less-simple-scheduler": "2.3.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", @@ -10417,9 +10417,9 @@ "license": "MIT" }, "node_modules/less-simple-scheduler": { - "version": "2.2.0", - "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.2.0.tgz", - "integrity": "sha512-0bbFohlkcWZY2wfkGfDiYEbDxkshj0RuskkL0RLUDMn1Y3WjoPqy29PAnKNtnHG+DQqn0hMjZ0e3hVFJ4k76oA==", + "version": "2.3.0", + "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.3.0.tgz", + "integrity": "sha512-HL8f8YCcNav03oNTWFQxkWmoqtEzsIXNOgLcVmOEAgFxsALukOk4bDlem6SlZmjVAsip8Y50QUu6ZsblxqQYmA==", "license": "ISC", "dependencies": { "mongodb": "7.2.0" diff --git a/server/package.json b/server/package.json index 559299061..c9b85e17c 100755 --- a/server/package.json +++ b/server/package.json @@ -53,7 +53,7 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", - "less-simple-scheduler": "2.2.0", + "less-simple-scheduler": "2.3.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", From 8b2506cc3d8200d0481dddcf24cf4726060fdc7f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:44:49 -0700 Subject: [PATCH 02/13] add queue worker repository --- server/src/repositories/index.ts | 3 ++ .../queue-workers/IQueueWorkersRepository.ts | 7 ++++ .../MongoQueueWorkersRepository.ts | 33 +++++++++++++++++++ 3 files changed, 43 insertions(+) create mode 100644 server/src/repositories/queue-workers/IQueueWorkersRepository.ts create mode 100644 server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts diff --git a/server/src/repositories/index.ts b/server/src/repositories/index.ts index f45768b02..d36d9d189 100644 --- a/server/src/repositories/index.ts +++ b/server/src/repositories/index.ts @@ -42,3 +42,6 @@ export { default as MongoGeoChecksRepository } from "@/repositories/geo-checks/M export * from "@/repositories/dlq/IDLQRepository.js"; export { default as MongoDLQRepository } from "@/repositories/dlq/MongoDLQRepository.js"; + +export * from "@/repositories/queue-workers/IQueueWorkersRepository.js"; +export { default as MongoQueueWorkersRepository } from "@/repositories/queue-workers/MongoQueueWorkersRepository.js"; diff --git a/server/src/repositories/queue-workers/IQueueWorkersRepository.ts b/server/src/repositories/queue-workers/IQueueWorkersRepository.ts new file mode 100644 index 000000000..b1f7e76b1 --- /dev/null +++ b/server/src/repositories/queue-workers/IQueueWorkersRepository.ts @@ -0,0 +1,7 @@ +import type { QueueMode, QueueWorker } from "@/types/settings.js"; + +export interface IQueueWorkersRepository { + upsert(workerId: string, mode: QueueMode): Promise; + findRecent(maxAgeMs: number): Promise; + deleteById(workerId: string): Promise; +} diff --git a/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts b/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts new file mode 100644 index 000000000..b8155ea1d --- /dev/null +++ b/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts @@ -0,0 +1,33 @@ +import type { QueueMode, QueueWorker } from "@/types/settings.js"; +import type { IQueueWorkersRepository } from "./IQueueWorkersRepository.js"; +import type { QueueWorkerDocument } from "@/db/models/QueueWorker.js"; +import { QueueWorkerModel } from "@/db/models/QueueWorker.js"; +const SERVICE_NAME = "MongoQueueWorkersRepository"; + +class MongoQueueWorkersRepository implements IQueueWorkersRepository { + static SERVICE_NAME = SERVICE_NAME; + + protected toEntity = (doc: QueueWorkerDocument): QueueWorker => { + return { + workerId: doc._id, + mode: doc.mode, + lastSeenAt: doc.lastSeenAt.getTime(), + }; + }; + + upsert = async (workerId: string, mode: QueueMode): Promise => { + await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true }); + }; + + findRecent = async (maxAgeMs: number): Promise => { + const cutoff = new Date(Date.now() - maxAgeMs); + const docs = await QueueWorkerModel.find({ lastSeenAt: { $gte: cutoff } }); + return docs.map(this.toEntity); + }; + + deleteById = async (workerId: string): Promise => { + await QueueWorkerModel.deleteOne({ _id: workerId }); + }; +} + +export default MongoQueueWorkersRepository; From cffe3f57facb9fbcfeb6a210fe2ab145fe0014c4 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:45:37 -0700 Subject: [PATCH 03/13] add models --- server/src/db/models/QueueWorker.ts | 28 ++++++++++++++++++++++++++++ server/src/db/models/index.ts | 3 +++ 2 files changed, 31 insertions(+) create mode 100644 server/src/db/models/QueueWorker.ts diff --git a/server/src/db/models/QueueWorker.ts b/server/src/db/models/QueueWorker.ts new file mode 100644 index 000000000..76485f56c --- /dev/null +++ b/server/src/db/models/QueueWorker.ts @@ -0,0 +1,28 @@ +import { Schema, model } from "mongoose"; +import { QueueModes, type QueueMode } from "@/types/settings.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. +interface QueueWorkerDocument { + _id: string; + mode: QueueMode; + lastSeenAt: Date; + createdAt: Date; + updatedAt: Date; +} + +const QueueWorkerSchema = new Schema( + { + _id: { type: String, required: true }, + mode: { type: String, enum: QueueModes, required: true }, + lastSeenAt: { type: Date, default: Date.now, expires: 30 }, + }, + { timestamps: true } +); + +const QueueWorkerModel = model("QueueWorker", QueueWorkerSchema); + +export type { QueueWorkerDocument }; +export { QueueWorkerModel }; +export default QueueWorkerModel; diff --git a/server/src/db/models/index.ts b/server/src/db/models/index.ts index f0647f9eb..b825ca503 100644 --- a/server/src/db/models/index.ts +++ b/server/src/db/models/index.ts @@ -41,3 +41,6 @@ export * from "@/db/models/DLQItem.js"; export { default as DLQItemModel } from "@/db/models/DLQItem.js"; export * from "@/db/models/Tag.js"; export { default as TagModel } from "@/db/models/Tag.js"; + +export * from "@/db/models/QueueWorker.js"; +export { default as QueueWorkerModel } from "@/db/models/QueueWorker.js"; From b2526a5be92b5fa02d94507d9d27a3593ef54eec Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:45:56 -0700 Subject: [PATCH 04/13] init queue worker repo --- server/src/config/services.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/server/src/config/services.ts b/server/src/config/services.ts index ef6233212..514bd04d5 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -107,6 +107,7 @@ import { MongoTeamsRepository, MongoMaintenanceWindowsRepository, MongoSettingsRepository, + MongoQueueWorkersRepository, IMonitorsRepository, IChecksRepository, IGeoChecksRepository, @@ -248,6 +249,7 @@ export const initializeServices = async ({ const incidentsRepository = new MongoIncidentsRepository(); const teamsRepository = new MongoTeamsRepository(); const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); + const queueWorkersRepository = new MongoQueueWorkersRepository(); // Inject settings repository into settings service (now that DB is connected) (settingsService as SettingsService).setRepository(settingsRepository); @@ -360,7 +362,7 @@ export const initializeServices = async ({ let jobQueue: IJobQueue; switch (envSettings.queueType) { case "lessSimpleQueue": - jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, envSettings, queueMode); + jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); break; default: jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository); From c9eacff71f81e5a23b9c870ae567b9852073a870 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:46:23 -0700 Subject: [PATCH 05/13] add queue worker type --- server/src/types/settings.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/server/src/types/settings.ts b/server/src/types/settings.ts index d49bd53dd..4a4ff02a2 100644 --- a/server/src/types/settings.ts +++ b/server/src/types/settings.ts @@ -7,6 +7,12 @@ export type QueueType = (typeof QueueTypes)[number]; export const QueueModes = ["primary", "worker"] as const; export type QueueMode = (typeof QueueModes)[number]; +export type QueueWorker = { + workerId: string; // hostname:pid:uuid + mode: QueueMode; + lastSeenAt: number; // epoch ms of the last heartbeat +}; + export interface SettingsThresholds { cpu?: number; memory?: number; From cfed235d1f216e6364096a125d8acb392cc706f1 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:47:00 -0700 Subject: [PATCH 06/13] add worker stats --- .../infrastructure/JobQueues/IJobQueue.ts | 4 + .../JobQueues/LessSimpleQueue.ts | 74 +++++++++++++++++-- .../JobQueues/SuperSimpleQueue.ts | 3 + 3 files changed, 76 insertions(+), 5 deletions(-) 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, From 599c5995881dd028422a74d8c2d6e7374af61851 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:47:13 -0700 Subject: [PATCH 07/13] add tests --- .../unit/services/lessSimpleQueue.test.ts | 150 +++++++++++++++++- .../unit/services/superSimpleQueue.test.ts | 4 +- 2 files changed, 147 insertions(+), 7 deletions(-) diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts index b5a79445e..1b455ef1b 100644 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ b/server/test/unit/services/lessSimpleQueue.test.ts @@ -12,6 +12,9 @@ const createScheduler = () => { listeners[event] = listeners[event] || []; listeners[event].push(cb); }), + off: jest.fn((event: string, cb: Function) => { + listeners[event] = (listeners[event] || []).filter((fn) => fn !== cb); + }), emit: (event: string, ...args: unknown[]) => { (listeners[event] || []).forEach((cb) => cb(...args)); }, @@ -51,18 +54,34 @@ const createMonitorsRepo = () => ({ findAll: jest.fn().mockResolvedValue([]), }); +const createWorkersRepo = () => ({ + upsert: jest.fn().mockResolvedValue(undefined), + findRecent: jest.fn().mockResolvedValue([]), + deleteById: jest.fn().mockResolvedValue(undefined), +}); + const createQueue = (overrides?: Record) => { const logger = createMockLogger(); const helper = createQueueHelper(); const monitorsRepository = createMonitorsRepo(); + const workersRepository = createWorkersRepo(); const scheduler = createScheduler(); - const defaults = { logger, helper, monitorsRepository, scheduler, ...overrides }; + const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, ...overrides }; - const queue = new LessSimpleQueue(defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.scheduler as any); + const queue = new LessSimpleQueue( + defaults.logger as any, + defaults.helper as any, + defaults.monitorsRepository as any, + defaults.workersRepository as any, + defaults.scheduler as any + ); return { queue, ...defaults }; }; +// A lockedUntil far enough ahead that it is always > Date.now() at assertion time. +const FAR_FUTURE = 4_000_000_000_000; // year 2096 + const makeMonitor = (overrides?: Partial): Monitor => ({ id: "mon-1", @@ -112,7 +131,8 @@ describe("LessSimpleQueue", () => { (monitorsRepository.findAll as jest.Mock).mockResolvedValue(monitors); await queue.init(); - jest.runAllTimers(); + // advance past the <100ms seed offsets + jest.advanceTimersByTime(100); expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m1", template: "monitor-job" })); expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m2", template: "monitor-job" })); @@ -454,7 +474,7 @@ describe("LessSimpleQueue", () => { it("returns empty metrics when no jobs", async () => { const { queue } = createQueue(); const metrics = await queue.getMetrics(); - expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0 }); + expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] }); }); it("aggregates metrics from jobs", async () => { @@ -478,6 +498,9 @@ describe("LessSimpleQueue", () => { failCount: 0, lastFailedAt: 0, lastFinishedAt: 50, + // live lock held by a worker + lockedBy: "host-1:1234:abcd", + lockedUntil: FAR_FUTURE, lockedAt: 999, data: { url: "http://b.com", type: "ping" }, }, @@ -526,6 +549,56 @@ describe("LessSimpleQueue", () => { }); }); + // ── getMetrics: activeJobs (in-flight) ─────────────────────────────────────── + + describe("getMetrics - activeJobs", () => { + it("counts jobs held by a live lock as active", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: FAR_FUTURE, lockedAt: 100, data: { url: "a" } }, + { id: "m2", template: "monitor-job", lockedBy: "worker-b", lockedUntil: FAR_FUTURE, lockedAt: 250, data: { url: "b" } }, + ]); + const metrics = await queue.getMetrics(); + expect(metrics.activeJobs).toBe(2); + }); + + it("does not count unlocked jobs (lockedBy null)", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", lockedBy: null, lockedUntil: null, lockedAt: null, data: { url: "a" } }, + ]); + const metrics = await queue.getMetrics(); + expect(metrics.activeJobs).toBe(0); + }); + + it("does not count jobs whose lock has expired (lockedUntil <= now)", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: 1, lockedAt: 1, data: { url: "a" } }, + ]); + const metrics = await queue.getMetrics(); + expect(metrics.activeJobs).toBe(0); + }); + }); + + // ── getMetrics: worker registry ────────────────────────────────────────────── + + describe("getMetrics - worker registry", () => { + it("returns the alive workers from the registry", async () => { + const { queue, workersRepository } = createQueue(); + const alive = [ + { workerId: "host:1:aaa", mode: "primary", lastSeenAt: 1000 }, + { workerId: "host:2:bbb", mode: "worker", lastSeenAt: 2000 }, + ]; + (workersRepository.findRecent as jest.Mock).mockResolvedValue(alive); + + const metrics = await queue.getMetrics(); + + expect(workersRepository.findRecent).toHaveBeenCalledWith(30000); + expect(metrics.workers).toEqual(alive); + }); + }); + // ── getJobs ──────────────────────────────────────────────────────────────── describe("getJobs", () => { @@ -559,6 +632,8 @@ describe("LessSimpleQueue", () => { monitorActive: true, active: true, repeat: 60000, + lockedBy: null, + lockedUntil: null, lockedAt: null, runCount: 5, failCount: 1, @@ -628,8 +703,16 @@ describe("LessSimpleQueue", () => { const logger = createMockLogger(); const helper = createQueueHelper(); const monitorsRepository = createMonitorsRepo(); + const workersRepository = createWorkersRepo(); - const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "primary"); + const instance = await LessSimpleQueue.create( + logger as any, + helper as any, + monitorsRepository as any, + workersRepository as any, + envSettings, + "primary" + ); expect(MockMongoStore).toHaveBeenCalledWith(expect.objectContaining({ url: "mongodb://localhost:27017/test_db" })); expect(MockScheduler).toHaveBeenCalled(); @@ -642,11 +725,66 @@ describe("LessSimpleQueue", () => { const logger = createMockLogger(); const helper = createQueueHelper(); const monitorsRepository = createMonitorsRepo(); + const workersRepository = createWorkersRepo(); - const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "worker"); + const instance = await LessSimpleQueue.create( + logger as any, + helper as any, + monitorsRepository as any, + workersRepository as any, + envSettings, + "worker" + ); expect(instance).toBeInstanceOf(LessSimpleQueue); expect(monitorsRepository.findAll).not.toHaveBeenCalled(); }); }); + + // ── worker heartbeat / deregistration ──────────────────────────────────────── + + describe("worker heartbeat", () => { + it("writes a heartbeat on init with the workerId and mode (primary)", async () => { + const { queue, workersRepository } = createQueue(); + await queue.init("primary"); + expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "primary"); + }); + + it("writes a heartbeat on init with the workerId and mode (worker)", async () => { + const { queue, workersRepository } = createQueue(); + await queue.init("worker"); + expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "worker"); + }); + + it("beats again on each scheduler:heartbeat event", async () => { + const { queue, scheduler, workersRepository } = createQueue(); + await queue.init("worker"); + (workersRepository.upsert as jest.Mock).mockClear(); + + scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd"); + scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd"); + + expect(workersRepository.upsert).toHaveBeenCalledTimes(2); + expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "worker"); + }); + + it("stops beating after shutdown", async () => { + const { queue, scheduler, workersRepository } = createQueue(); + await queue.init("worker"); + await queue.shutdown(); + (workersRepository.upsert as jest.Mock).mockClear(); + + scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd"); + + expect(workersRepository.upsert).not.toHaveBeenCalled(); + }); + + it("deregisters the worker and stops the scheduler on shutdown", async () => { + const { queue, scheduler, workersRepository } = createQueue(); + await queue.init("worker"); + await queue.shutdown(); + expect(workersRepository.deleteById).toHaveBeenCalledWith("host-1:1234:abcd"); + expect(scheduler.stop).toHaveBeenCalled(); + }); + }); }); diff --git a/server/test/unit/services/superSimpleQueue.test.ts b/server/test/unit/services/superSimpleQueue.test.ts index 660d0b8f7..970706199 100644 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ b/server/test/unit/services/superSimpleQueue.test.ts @@ -489,7 +489,7 @@ describe("SuperSimpleQueue", () => { it("returns empty metrics when no jobs", async () => { const { queue } = createQueue(); const metrics = await queue.getMetrics(); - expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0 }); + expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] }); }); it("aggregates metrics from jobs", async () => { @@ -580,6 +580,8 @@ describe("SuperSimpleQueue", () => { monitorActive: true, active: true, repeat: 60000, + lockedBy: null, + lockedUntil: null, lockedAt: null, runCount: 5, failCount: 1, From caf8f22fec0523a06cf9b5fefed4cb902085a1c4 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 10:59:14 -0700 Subject: [PATCH 08/13] add worker table --- client/src/Pages/Logs/TabQueue.tsx | 34 ++++++- .../src/Pages/Logs/components/TableJobs.tsx | 21 +++++ .../Pages/Logs/components/TableWorkers.tsx | 94 +++++++++++++++++++ client/src/Pages/Logs/index.tsx | 2 +- client/src/Types/Queue.ts | 22 ++++- client/src/locales/en.json | 8 ++ 6 files changed, 174 insertions(+), 7 deletions(-) create mode 100644 client/src/Pages/Logs/components/TableWorkers.tsx diff --git a/client/src/Pages/Logs/TabQueue.tsx b/client/src/Pages/Logs/TabQueue.tsx index a39250be0..bb3aac368 100644 --- a/client/src/Pages/Logs/TabQueue.tsx +++ b/client/src/Pages/Logs/TabQueue.tsx @@ -2,7 +2,7 @@ import Stack from "@mui/material/Stack"; import Typography from "@mui/material/Typography"; import Box from "@mui/material/Box"; import { TableJobs, TableFailedJobs } from "./components/TableJobs"; - +import { TableWorkers } from "./components/TableWorkers"; import { useTheme } from "@mui/material"; import { useTranslation, Trans } from "react-i18next"; import { useGet, usePost } from "@/Hooks/UseApi"; @@ -25,6 +25,10 @@ export const TabQueue = () => { const { post, loading: isFlushing } = usePost(); const jobs = queueData?.jobs ?? []; + const workers = (queueData?.metrics?.workers ?? []).map((worker, idx) => ({ + ...worker, + id: idx, + })); const metrics = queueData?.metrics ?? null; const hasNeverRun = !isLoading && metrics !== null && (metrics.totalRuns ?? 0) === 0; const noQueueData = !isLoading && metrics === null && jobs.length === 0; @@ -46,6 +50,34 @@ export const TabQueue = () => { return ( + + + + {t("pages.logs.workers")} + + + + ), + }} + /> + + + + { ); }, }, + { + id: "lockedBy", + content: t("pages.logs.table.headers.lockedBy"), + render: (row) => { + const lockedBy = row.lockedBy; + if (!lockedBy) { + return -; + } + + const workerId = lockedBy.split(":")[2]; + const shortWorkerId = workerId.slice(workerId.length - 8); + return ( + + {`...${shortWorkerId}`} + + ); + }, + }, ]; return ( diff --git a/client/src/Pages/Logs/components/TableWorkers.tsx b/client/src/Pages/Logs/components/TableWorkers.tsx new file mode 100644 index 000000000..5d1f44860 --- /dev/null +++ b/client/src/Pages/Logs/components/TableWorkers.tsx @@ -0,0 +1,94 @@ +import { Table } from "@/Components/design-elements"; +import { Typography, useTheme } from "@mui/material"; +import { formatTimestamp } from "@/Utils/TimeUtils"; + +import { useTranslation } from "react-i18next"; +import type { QueueWorker } from "@/Types/Queue"; +import type { Header } from "@/Components/design-elements"; + +type QueueWorkerWithId = QueueWorker & { id: number }; + +interface TableWorkersProps { + queueWorkers: QueueWorkerWithId[]; +} + +export const TableWorkers = ({ queueWorkers }: TableWorkersProps) => { + const { t } = useTranslation(); + const theme = useTheme(); + + const cellSx = { + whiteSpace: "nowrap" as const, + overflow: "hidden", + textOverflow: "ellipsis", + maxWidth: 220, + }; + + const headers: Header[] = [ + { + id: "host", + content: t("pages.logs.table.headers.host"), + render: (row) => { + const host = row.workerId.split(":")[0]; + return ( + + {host} + + ); + }, + }, + { + id: "processId", + content: t("pages.logs.table.headers.processId"), + render: (row) => { + const processId = row.workerId.split(":")[1]; + return ( + + {processId} + + ); + }, + }, + { + id: "id", + content: t("pages.logs.table.headers.workerId"), + render: (row) => { + const workerId = row.workerId.split(":")[2]; + const shortWorkerId = workerId.slice(workerId.length - 8); + return ( + + {`...${shortWorkerId}`} + + ); + }, + }, + { + id: "mode", + content: t("pages.logs.table.headers.workerMode"), + render: (row) => {row.mode}, + }, + { + id: "lastSeen", + content: t("pages.logs.table.headers.lastSeenAt"), + render: (row) => ( + {formatTimestamp(row.lastSeenAt)} + ), + }, + ]; + + return ( + + ); +}; diff --git a/client/src/Pages/Logs/index.tsx b/client/src/Pages/Logs/index.tsx index b21e331c5..9d71e3253 100644 --- a/client/src/Pages/Logs/index.tsx +++ b/client/src/Pages/Logs/index.tsx @@ -8,7 +8,7 @@ import { useTranslation } from "react-i18next"; const LogsPage = () => { const { t } = useTranslation(); - const [activeTab, setActiveTab] = useState(0); + const [activeTab, setActiveTab] = useState(1); return ( Highlighted rows are jobs currently being processed by a worker.", "failedJobs": "Failed jobs", @@ -1000,6 +1002,11 @@ }, "table": { "headers": { + "host": "Host", + "processId": "Process ID", + "workerId": "Worker ID", + "workerMode": "Worker mode", + "lastSeenAt": "Last seen at", "state": "State", "timestamp": "Timestamp", "level": "Level", @@ -1010,6 +1017,7 @@ "failCount": "Count", "lastRunAt": "Last run at", "lockedAt": "Locked at", + "lockedBy": "Locked by", "lastFinishedAt": "Last finished at", "lastRunTook": "Last run took", "lastFailedAt": "Last failed", From b9fd29b906b70272ff2413c15ee62320af27030a Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:11:24 -0700 Subject: [PATCH 09/13] add compliant worker ID to super simple queue --- .../infrastructure/JobQueues/SuperSimpleQueue.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index 353a7ec7c..c96448728 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -3,6 +3,9 @@ import { ILogger } from "@/utils/logger.js"; import Scheduler from "super-simple-scheduler"; import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js"; +import { hostname } from "node:os"; +import { randomUUID } from "node:crypto"; + const SERVICE_NAME = "JobQueue"; type QueueJobFailure = { @@ -61,6 +64,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { private logger: ILogger; private helper: IQueueHelper; + private workerId: string; private monitorsRepository: IMonitorsRepository; private readonly scheduler: Scheduler; @@ -70,6 +74,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { this.monitorsRepository = monitorsRepository; this.scheduler = scheduler; this.registerListeners(); + this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`; } get serviceName() { @@ -354,7 +359,13 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { }>, totalRuns: 0, totalFailures: 0, - workers: [], + workers: [ + { + workerId: this.workerId, + mode: "primary", + lastSeenAt: Date.now(), + }, + ], } ); return metrics; @@ -372,7 +383,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { monitorActive: job.data?.isActive ?? null, active: job.active, repeat: job.repeat ?? null, - lockedBy: null, + lockedBy: job.lockedAt ? this.workerId : null, lockedUntil: null, lockedAt: job.lockedAt ?? null, runCount: job.runCount ?? 0, From 2b6734bb783446ccf552717a2cd0af4737e682f6 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:25:21 -0700 Subject: [PATCH 10/13] update tests --- .../unit/services/superSimpleQueue.test.ts | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/server/test/unit/services/superSimpleQueue.test.ts b/server/test/unit/services/superSimpleQueue.test.ts index 970706199..3c2506774 100644 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ b/server/test/unit/services/superSimpleQueue.test.ts @@ -486,10 +486,25 @@ describe("SuperSimpleQueue", () => { // ── getMetrics ─────────────────────────────────────────────────────────── describe("getMetrics", () => { - it("returns empty metrics when no jobs", async () => { + it("returns empty job metrics plus this process as the sole worker when no jobs", async () => { const { queue } = createQueue(); const metrics = await queue.getMetrics(); - expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] }); + expect(metrics).toEqual({ + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + workers: [ + { + // hostname:pid:uuid, matching less-simple-scheduler's workerId format + workerId: expect.stringMatching(/^.+:\d+:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/), + mode: "primary", + lastSeenAt: expect.any(Number), + }, + ], + }); }); it("aggregates metrics from jobs", async () => { From dfae6c8abdf1669a66f8bc3105c6673dd4d9b2ac Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:26:10 -0700 Subject: [PATCH 11/13] type metrics for super simple queue --- .../JobQueues/SuperSimpleQueue.ts | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index c96448728..ee542d2be 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -5,6 +5,7 @@ import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js" import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js"; import { hostname } from "node:os"; import { randomUUID } from "node:crypto"; +import { QueueMode } from "@/types/settings.js"; const SERVICE_NAME = "JobQueue"; @@ -316,7 +317,26 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { getMetrics = async () => { const jobs = await this.scheduler.getJobs(); - const metrics = jobs.reduce( + const metrics: { + jobs: number; + activeJobs: number; + failingJobs: number; + jobsWithFailures: Array<{ + monitorId: string | number; + monitorUrl: string; + monitorType: MonitorType; + failedAt: number | null; + failCount: number; + failReason: string | null; + }>; + totalRuns: number; + totalFailures: number; + workers: Array<{ + workerId: string; + mode: QueueMode; + lastSeenAt: number; + }>; + } = jobs.reduce( (acc, job) => { const runCount = job.runCount ?? 0; const failCount = job.failCount ?? 0; From f5a4a038983080fd2fc33ab77b3ecb2a4cfb7e38 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:31:11 -0700 Subject: [PATCH 12/13] reorder stop --- .../infrastructure/JobQueues/LessSimpleQueue.ts | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index cedf29901..ae0b5df72 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -27,6 +27,7 @@ export class LessSimpleQueue implements IJobQueue { private monitorsRepository: IMonitorsRepository; private workersRepository: IQueueWorkersRepository; private readonly scheduler: Scheduler; + private queueMode: QueueMode; private heartbeatListener: ((workerId: string) => void) | null = null; constructor( @@ -34,13 +35,15 @@ export class LessSimpleQueue implements IJobQueue { helper: IQueueHelper, monitorsRepository: IMonitorsRepository, workersRepository: IQueueWorkersRepository, - scheduler: Scheduler + scheduler: Scheduler, + queueMode: QueueMode ) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; this.workersRepository = workersRepository; this.scheduler = scheduler; + this.queueMode = queueMode; this.registerListeners(); } @@ -137,7 +140,7 @@ export class LessSimpleQueue implements IJobQueue { const scheduler = new Scheduler(store, { concurrency: 50, }); - const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode); await instance.init(queueMode); return instance; @@ -422,8 +425,14 @@ export class LessSimpleQueue implements IJobQueue { }; flushQueues = async () => { - const stopRes = await this.scheduler.stop(); - const flushRes = await this.scheduler.flushJobs(); + if (this.queueMode !== "primary") { + // Don't allow workers to flush, though this isn't currently reachable since workers have no API endpoints + this.logger.warn({ message: "Ignoring flush on non-primary worker", service: SERVICE_NAME, method: "flushQueues" }); + return { success: false }; + } + + const flushRes = await this.scheduler.flushJobs(); // Less Simple Queue must flush before stopping as it needs DB + const stopRes = await this.scheduler.stop(); // Destructive, other workers will abort their jobs const initRes = await this.init(); return { success: Boolean(stopRes && flushRes && initRes), From b4970d321d27b23565e1e92019258d9127e12939 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:31:22 -0700 Subject: [PATCH 13/13] update tests --- .../unit/services/lessSimpleQueue.test.ts | 28 +++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts index 1b455ef1b..19b68dee1 100644 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ b/server/test/unit/services/lessSimpleQueue.test.ts @@ -67,14 +67,15 @@ const createQueue = (overrides?: Record) => { const workersRepository = createWorkersRepo(); const scheduler = createScheduler(); - const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, ...overrides }; + const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, queueMode: "primary", ...overrides }; const queue = new LessSimpleQueue( defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.workersRepository as any, - defaults.scheduler as any + defaults.scheduler as any, + defaults.queueMode as any ); return { queue, ...defaults }; }; @@ -686,6 +687,29 @@ describe("LessSimpleQueue", () => { expect(result.success).toBe(true); }); + it("flushes before stopping, so the store is still connected for removeAll", async () => { + const { queue, scheduler } = createQueue(); + const order: string[] = []; + (scheduler.flushJobs as jest.Mock).mockImplementation(async () => { + order.push("flush"); + return true; + }); + (scheduler.stop as jest.Mock).mockImplementation(async () => { + order.push("stop"); + return true; + }); + await queue.flushQueues(); + expect(order).toEqual(["flush", "stop"]); + }); + + it("refuses to flush on a non-primary worker without touching the scheduler", async () => { + const { queue, scheduler } = createQueue({ queueMode: "worker" }); + const result = await queue.flushQueues(); + expect(result.success).toBe(false); + expect(scheduler.flushJobs).not.toHaveBeenCalled(); + expect(scheduler.stop).not.toHaveBeenCalled(); + }); + it("returns false when flush fails", async () => { const { queue, scheduler } = createQueue(); (scheduler.flushJobs as jest.Mock).mockResolvedValue(false);