From 142ecaaf8b64fb76352f853283b194e2d4ee439f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 15 Jun 2026 12:05:15 -0700 Subject: [PATCH] handle renewing locks for long running jobs --- .../domain/jobs/job.repository.interface.ts | 3 + .../src/domain/jobs/job.repository.mongo.ts | 12 + server/src/worker/worker.db-queue.ts | 28 +- .../unit/services/lessSimpleQueue.test.ts | 791 ------------------ .../unit/services/superSimpleQueue.test.ts | 712 ---------------- ...eueHelper.test.ts => workerHelper.test.ts} | 0 6 files changed, 41 insertions(+), 1505 deletions(-) delete mode 100644 server/test/unit/services/lessSimpleQueue.test.ts delete mode 100644 server/test/unit/services/superSimpleQueue.test.ts rename server/test/unit/services/{superSimpleQueueHelper.test.ts => workerHelper.test.ts} (100%) diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index 5a9bf901e..4def9cd86 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -20,6 +20,9 @@ export interface IJobsRepository { // locks up to `limit` due rows in one batch claimDueBatch(type: JobType, limit: number, now: number): Promise; + // Extends the locks we still hold. Returns how many were renewed + renewLocks(ids: string[], now: number): Promise; + // Completes a repeating job recordSuccess(id: string, nextScheduledAt: number, intervalMs: number, now: number): Promise; diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index a8c0a5d12..7b409c911 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -76,6 +76,15 @@ class MongoJobsRepository implements IJobsRepository { return docs.map(this.toEntity); }; + renewLocks = async (ids: string[], now: number) => { + if (ids.length === 0) return 0; + const res = await JobModel.updateMany( + { _id: { $in: ids }, lockedBy: this.workerId }, // only locks held by this worker can be renewed + { $set: { lockedUntil: now + LOCK_MS } } + ); + return res.modifiedCount; + }; + recordSuccess = async (id: string, scheduledRunAt: number, intervalMs: number, now: number) => { // Fixed rate job scheduling @@ -89,6 +98,7 @@ class MongoJobsRepository implements IJobsRepository { const res = await JobModel.updateOne( { _id: id, + lockedBy: this.workerId, // Only the worker that claimed can record success }, { $set: { @@ -109,6 +119,7 @@ class MongoJobsRepository implements IJobsRepository { const res = await JobModel.updateOne( { _id: id, + lockedBy: this.workerId, // Only the worker that claimed can record success }, { $set: { @@ -129,6 +140,7 @@ class MongoJobsRepository implements IJobsRepository { const res = await JobModel.updateOne( { _id: id, + lockedBy: this.workerId, // Only the worker that claimed can record success }, { $set: { diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index b26bb4e93..3d106445b 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -1,6 +1,6 @@ import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { IWorker, WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; -import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js"; +import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js"; import { randomUUID } from "node:crypto"; import { hostname } from "node:os"; import { ILogger } from "@/utils/logger.js"; @@ -20,6 +20,7 @@ const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; 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 CONCURRENCY: Record = { check: 500, @@ -179,7 +180,28 @@ export class DBQueueWorker implements IWorker { if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff }; + // Extend the lock while a job runs so a slow-but-alive job is never reclaimed. + // Returns false once we no longer hold the lock, so the renewal timer can stop. + private renewLock = async (job: Job): Promise => { + try { + const held = await this.jobsRepository.renewLocks([job.id], Date.now()); + if (held === 0) { + this.logger.warn({ message: `Lost lock on job ${job.id}, stopping renewal`, service: SERVICE_NAME, method: "renewLock" }); + return false; + } + return true; + } catch (error: unknown) { + // Transient failure, keep renewing; a missed renewal is absorbed by the 1/3 margin + this.logger.warn({ message: error instanceof Error ? error.message : String(error), service: SERVICE_NAME, method: "renewLock" }); + return true; + } + }; + private runJob = async (job: Job) => { + const renewTimer = setInterval(async () => { + const renewLock = await this.renewLock(job); + if (!renewLock) clearInterval(renewTimer); + }, LOCK_RENEW_MS); try { switch (job.type) { case "check": @@ -212,6 +234,8 @@ export class DBQueueWorker implements IWorker { service: SERVICE_NAME, method: `runJob:${job.type}`, }); + } finally { + clearInterval(renewTimer); } }; @@ -314,7 +338,7 @@ export class DBQueueWorker implements IWorker { clearTimeout(timer); } this.timers.clear(); // Clear the map out - // Deregister promptly so the worker drops off diagnostics before the TTL would + // Remove workers from registry await this.queueWorkersRepository.deleteById(this.workerId); }; getMetrics = async (): Promise => { diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts deleted file mode 100644 index 8b29df1bb..000000000 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ /dev/null @@ -1,791 +0,0 @@ -import { describe, expect, it, jest } from "@jest/globals"; -import { createMockLogger } from "../../helpers/createMockLogger.ts"; -import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; - -// ── Mock Scheduler / MongoStore ────────────────────────────────────────────── - -const createScheduler = () => { - const listeners: Record = {}; - return { - workerId: "host-1:1234:abcd", - on: jest.fn((event: string, cb: Function) => { - 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)); - }, - start: jest.fn().mockResolvedValue(true), - stop: jest.fn().mockResolvedValue(true), - addTemplate: jest.fn(), - addJob: jest.fn(), - addJobs: jest.fn().mockResolvedValue({ inserted: 0, upserted: 0, modified: 0, failures: [] }), - removeJob: jest.fn(), - getJob: jest.fn().mockResolvedValue(null), - getJobs: jest.fn().mockResolvedValue([]), - countJobs: jest.fn().mockResolvedValue(0), - getStats: jest.fn().mockResolvedValue({ jobs: 0, activeJobs: 0, failingJobs: 0, totalRuns: 0, totalFailures: 0, jobsWithFailures: [] }), - pauseJob: jest.fn().mockResolvedValue(true), - resumeJob: jest.fn().mockResolvedValue(true), - updateJob: jest.fn(), - flushJobs: jest.fn().mockResolvedValue(true), - }; -}; - -const mockSchedulerInstance = createScheduler(); -const MockScheduler = jest.fn().mockReturnValue(mockSchedulerInstance); -const MockMongoStore = jest.fn(); - -jest.unstable_mockModule("less-simple-scheduler", () => ({ - Scheduler: MockScheduler, - MongoStore: MockMongoStore, -})); - -const { LessSimpleWorker: LessSimpleQueue } = await import("../../../src/worker/worker.less-simple.ts"); - -const createQueueHelper = () => ({ - getHeartbeatJob: jest.fn().mockReturnValue(() => Promise.resolve()), - getHeartbeatGeoJob: jest.fn().mockReturnValue(() => Promise.resolve()), - getCleanupOrphanedJob: jest.fn().mockReturnValue(() => Promise.resolve()), - getCleanupRetentionJob: jest.fn().mockReturnValue(() => Promise.resolve()), -}); - -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, 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.queueMode as any - ); - return { queue, ...defaults }; -}; - -const makeMonitor = (overrides?: Partial): Monitor => - ({ - id: "mon-1", - teamId: "team-1", - type: "http", - interval: 60000, - isActive: true, - geoCheckEnabled: false, - geoCheckInterval: 300000, - geoCheckLocations: [], - ...overrides, - }) as Monitor; - -// ── Tests ──────────────────────────────────────────────────────────────────── - -describe("LessSimpleQueue", () => { - describe("serviceName", () => { - it("returns JobQueue", () => { - const { queue } = createQueue(); - expect(queue.serviceName).toBe("JobQueue"); - }); - }); - - // ── init (primary) ───────────────────────────────────────────────────────── - - describe("init - primary mode", () => { - it("starts scheduler, registers all templates, seeds monitors and system jobs", async () => { - const { queue, scheduler, monitorsRepository } = createQueue(); - - const result = await queue.init(); - - expect(result).toBe(true); - expect(scheduler.start).toHaveBeenCalled(); - expect(scheduler.addTemplate).toHaveBeenCalledWith("monitor-job", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("geo-check-job", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-orphaned", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-retention-job", expect.any(Function)); - expect(monitorsRepository.findAll).toHaveBeenCalled(); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" })); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-retention" })); - }); - - it("registers existing monitors in a single bulk write", async () => { - const monitors = [makeMonitor({ id: "m1" }), makeMonitor({ id: "m2" })]; - const { queue, scheduler, monitorsRepository } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockResolvedValue(monitors); - - await queue.init(); - - expect(scheduler.addJobs).toHaveBeenCalledTimes(1); - expect(scheduler.addJobs).toHaveBeenCalledWith( - expect.arrayContaining([ - expect.objectContaining({ id: "m1", template: "monitor-job", jitter: true }), - expect.objectContaining({ id: "m2", template: "monitor-job", jitter: true }), - ]) - ); - }); - - it("returns true and skips system jobs when findAll returns null", async () => { - const { queue, monitorsRepository, scheduler } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockResolvedValue(null); - - const result = await queue.init(); - - expect(result).toBe(true); - expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" })); - }); - - it("returns false and logs error on failure", async () => { - const { queue, logger, monitorsRepository } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockRejectedValue(new Error("db down")); - - const result = await queue.init(); - - expect(result).toBe(false); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("db down") })); - }); - - it("logs error with String(error) for non-Error exceptions", async () => { - const { queue, logger, monitorsRepository } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockRejectedValue("string error"); - - const result = await queue.init(); - - expect(result).toBe(false); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("string error") })); - }); - }); - - // ── init (worker) ────────────────────────────────────────────────────────── - - describe("init - worker mode", () => { - it("starts scheduler and registers all templates", async () => { - const { queue, scheduler } = createQueue(); - - const result = await queue.init("worker"); - - expect(result).toBe(true); - expect(scheduler.start).toHaveBeenCalled(); - expect(scheduler.addTemplate).toHaveBeenCalledWith("monitor-job", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("geo-check-job", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-orphaned", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-retention-job", expect.any(Function)); - }); - - it("does NOT seed monitor jobs (no findAll, primary populates the queue)", async () => { - const { queue, monitorsRepository } = createQueue(); - - await queue.init("worker"); - - expect(monitorsRepository.findAll).not.toHaveBeenCalled(); - }); - - it("does NOT seed cleanup system jobs", async () => { - const { queue, scheduler } = createQueue(); - - await queue.init("worker"); - - expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" })); - expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-retention" })); - expect(scheduler.addJob).not.toHaveBeenCalled(); - }); - - it("logs the initialized worker with its workerId", async () => { - const { queue, logger } = createQueue(); - - await queue.init("worker"); - - expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Worker queue: host-1:1234:abcd initialized" })); - }); - }); - - // ── registerListeners ────────────────────────────────────────────────────── - - describe("registerListeners (via constructor)", () => { - it("logs on scheduler:start", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("scheduler:start", "host-1:1234:abcd"); - expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler started" })); - }); - - it("logs on scheduler:stop", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("scheduler:stop", "host-1:1234:abcd"); - expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler stopped" })); - }); - - it("logs on scheduler:error with Error", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("scheduler:error", "w1", new Error("boom")); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 error: boom" })); - }); - - it("logs on scheduler:error with non-Error", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("scheduler:error", "w1", "string error"); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 error: string error", stack: undefined })); - }); - - it("logs on job:locked", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:locked", "w1", { id: "m1" }); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 locked" })); - }); - - it("logs on job:abort", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:abort", "w1", { id: "m1" }, "timeout"); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 aborted: timeout" })); - }); - - it("logs on job:attempt", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:attempt", "w1", { id: "m1" }, 2); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 attempt 2" })); - }); - - it("logs on job:complete", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:complete", "w1", { id: "m1" }); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 completed successfully" })); - }); - - it("logs on job:exhausted with Error", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:exhausted", "w1", { id: "m1" }, new Error("gave up")); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 exhausted all retries: gave up" })); - }); - - it("logs on job:exhausted with non-Error", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:exhausted", "w1", { id: "m1" }, "gave up"); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 exhausted all retries: gave up", stack: undefined })); - }); - - it("logs on job:fail with Error", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:fail", "w1", { id: "m1" }, new Error("oops"), 1); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 failed on attempt 1: oops" })); - }); - - it("logs on job:fail with non-Error", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:fail", "w1", { id: "m1" }, "oops", 1); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 failed on attempt 1: oops", stack: undefined })); - }); - - it("logs on job:start", () => { - const { scheduler, logger } = createQueue(); - scheduler.emit("job:start", "w1", { id: "m1" }); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 started" })); - }); - }); - - // ── addJob ───────────────────────────────────────────────────────────────── - - describe("addJob", () => { - it("adds monitor job to scheduler", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor()); - expect(scheduler.addJobs).toHaveBeenCalledWith( - expect.arrayContaining([expect.objectContaining({ id: "mon-1", template: "monitor-job", repeat: 60000, active: true })]) - ); - }); - - it("adds geo check job when geoCheckEnabled and type supports it", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 })); - expect(scheduler.addJobs).toHaveBeenCalledWith( - expect.arrayContaining([expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job", repeat: 300000 })]) - ); - }); - - it("skips geo job when type does not support geo checks", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: true, type: "hardware" })); - expect(scheduler.addJobs).toHaveBeenCalledWith(expect.not.arrayContaining([expect.objectContaining({ id: "mon-1-geo" })])); - }); - - it("skips geo job when geoCheckEnabled is false", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: false, type: "http" })); - expect(scheduler.addJobs).toHaveBeenCalledWith(expect.not.arrayContaining([expect.objectContaining({ id: "mon-1-geo" })])); - }); - }); - - // ── deleteJob ────────────────────────────────────────────────────────────── - - describe("deleteJob", () => { - it("removes monitor job", async () => { - const { queue, scheduler } = createQueue(); - await queue.deleteJob(makeMonitor()); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1"); - }); - - it("removes geo job if it exists", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.deleteJob(makeMonitor()); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - - it("does not remove geo job if it does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.deleteJob(makeMonitor()); - expect(scheduler.removeJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - }); - - // ── pauseJob ─────────────────────────────────────────────────────────────── - - describe("pauseJob", () => { - it("pauses monitor and geo job when geo exists", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.pauseJob(makeMonitor()); - - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1-geo"); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "Paused monitor mon-1" })); - }); - - it("does not pause geo job when it does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.pauseJob(makeMonitor()); - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.pauseJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - - it("throws when scheduler.pauseJob returns false for the main job", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.pauseJob as jest.Mock).mockResolvedValue(false); - await expect(queue.pauseJob(makeMonitor())).rejects.toThrow("Failed to pause monitor"); - }); - - it("logs error but does not throw when geo pauseJob returns false", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - (scheduler.pauseJob as jest.Mock).mockImplementation(async (id: string) => id !== "mon-1-geo"); - - await expect(queue.pauseJob(makeMonitor())).resolves.toBeUndefined(); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Failed to pause geo check job for monitor mon-1" })); - }); - }); - - // ── resumeJob ────────────────────────────────────────────────────────────── - - describe("resumeJob", () => { - it("resumes monitor and geo job when geo exists", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.resumeJob(makeMonitor()); - - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1-geo"); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "Resumed monitor mon-1" })); - }); - - it("does not resume geo job when it does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.resumeJob(makeMonitor()); - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.resumeJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - - it("throws when scheduler.resumeJob returns false for the main job", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.resumeJob as jest.Mock).mockResolvedValue(false); - await expect(queue.resumeJob(makeMonitor())).rejects.toThrow("Failed to resume monitor"); - }); - - it("logs error but does not throw when geo resumeJob returns false", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - (scheduler.resumeJob as jest.Mock).mockImplementation(async (id: string) => id !== "mon-1-geo"); - - await expect(queue.resumeJob(makeMonitor())).resolves.toBeUndefined(); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Failed to resume geo check job for monitor mon-1" })); - }); - }); - - // ── updateJob / syncGeoJob ───────────────────────────────────────────────── - - describe("updateJob", () => { - it("updates monitor job with new interval", async () => { - const { queue, scheduler } = createQueue(); - const monitor = makeMonitor({ interval: 120000 }); - await queue.updateJob(monitor); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", expect.objectContaining({ repeat: 120000, data: monitor })); - }); - - it("removes existing geo job when geoCheckEnabled is false", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.updateJob(makeMonitor({ geoCheckEnabled: false, type: "http" })); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - - it("removes existing geo job when type does not support geo checks", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.updateJob(makeMonitor({ geoCheckEnabled: true, type: "hardware" })); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - - it("updates existing geo job when geoCheckEnabled", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - const monitor = makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 }); - await queue.updateJob(monitor); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1-geo", expect.objectContaining({ repeat: 300000 })); - }); - - it("creates new geo job when enabled but no existing job", async () => { - const { queue, scheduler } = createQueue(); - const monitor = makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 }); - await queue.updateJob(monitor); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job" })); - }); - }); - - // ── shutdown ─────────────────────────────────────────────────────────────── - - describe("shutdown", () => { - it("stops the scheduler", async () => { - const { queue, scheduler } = createQueue(); - await queue.shutdown(); - expect(scheduler.stop).toHaveBeenCalled(); - }); - }); - - // ── getMetrics ───────────────────────────────────────────────────────────── - - describe("getMetrics", () => { - 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, workers: [] }); - }); - - it("maps the store's aggregate counters through", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getStats as jest.Mock).mockResolvedValue({ - jobs: 2, - activeJobs: 1, - failingJobs: 1, - totalRuns: 15, - totalFailures: 2, - jobsWithFailures: [], - }); - const metrics = await queue.getMetrics(); - expect(metrics).toMatchObject({ jobs: 2, activeJobs: 1, failingJobs: 1, totalRuns: 15, totalFailures: 2 }); - }); - - it("derives monitor fields for failing monitor jobs from job data", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getStats as jest.Mock).mockResolvedValue({ - jobs: 1, - activeJobs: 0, - failingJobs: 1, - totalRuns: 10, - totalFailures: 2, - jobsWithFailures: [{ id: "m1", data: { url: "http://a.com", type: "http" }, failedAt: 200, failCount: 2, failReason: "timeout" }], - }); - const metrics = await queue.getMetrics(); - expect(metrics.jobsWithFailures).toEqual([ - { monitorId: "m1", monitorUrl: "http://a.com", monitorType: "http", failedAt: 200, failCount: 2, failReason: "timeout" }, - ]); - }); - - it("leaves monitor fields null for non-monitor (cleanup) failures", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getStats as jest.Mock).mockResolvedValue({ - jobs: 1, - activeJobs: 0, - failingJobs: 1, - totalRuns: 1, - totalFailures: 1, - jobsWithFailures: [{ id: "cleanup-orphaned", data: undefined, failedAt: 100, failCount: 1, failReason: "boom" }], - }); - const metrics = await queue.getMetrics(); - expect(metrics.jobsWithFailures[0]).toEqual({ - monitorId: "cleanup-orphaned", - monitorUrl: null, - monitorType: null, - failedAt: 100, - failCount: 1, - failReason: "boom", - }); - }); - }); - - // ── 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", () => { - it("maps scheduler jobs to summaries", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { - id: "m1", - template: "monitor-job", - active: true, - repeat: 60000, - lockedAt: null, - runCount: 5, - failCount: 1, - lastError: "timeout", - lastStartedAt: 1000, - lastFinishedAt: 1100, - lastFailedAt: 900, - data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true }, - }, - ]); - const { jobs } = await queue.getJobs({}); - - expect(jobs).toHaveLength(1); - expect(jobs[0]).toEqual({ - monitorId: "m1", - monitorUrl: "http://a.com", - monitorType: "http", - monitorInterval: 60000, - monitorGeoInterval: 300000, - monitorActive: true, - active: true, - repeat: 60000, - lockedBy: null, - lockedUntil: null, - lockedAt: null, - runCount: 5, - failCount: 1, - failReason: "timeout", - lastRunAt: 1000, - lastFinishedAt: 1100, - lastRunTook: 100, - lastFailedAt: 900, - }); - }); - - it("returns null lastRunTook when job is locked", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { - id: "m1", - template: "monitor-job", - active: true, - lockedAt: 999, - runCount: 1, - lastStartedAt: 1000, - lastFinishedAt: 1100, - data: { url: "a", type: "http", interval: 60000 }, - }, - ]); - const { jobs } = await queue.getJobs({}); - expect(jobs[0].lastRunTook).toBeNull(); - }); - - it("leaves monitor fields null for non-monitor jobs", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, data: undefined }]); - const { jobs } = await queue.getJobs({}); - expect(jobs[0].monitorUrl).toBeNull(); - expect(jobs[0].monitorType).toBeNull(); - expect(jobs[0].monitorInterval).toBeNull(); - expect(jobs[0].monitorGeoInterval).toBeNull(); - expect(jobs[0].monitorActive).toBeNull(); - }); - - it("pushes pagination to the scheduler and returns its total count", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", template: "monitor-job", active: true, data: { url: "a", type: "http", interval: 60000 } }, - ]); - (scheduler.countJobs as jest.Mock).mockResolvedValue(42); - - const { jobs, count } = await queue.getJobs({ page: 2, rowsPerPage: 10 }); - - expect(scheduler.getJobs).toHaveBeenCalledWith({ skip: 20, limit: 10 }); - expect(count).toBe(42); - expect(jobs).toHaveLength(1); - }); - - it("requests all jobs when rowsPerPage is omitted", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.countJobs as jest.Mock).mockResolvedValue(7); - - const { count } = await queue.getJobs({}); - - expect(scheduler.getJobs).toHaveBeenCalledWith(undefined); - expect(count).toBe(7); - }); - }); - - // ── flushQueues ──────────────────────────────────────────────────────────── - - describe("flushQueues", () => { - it("stops, flushes, and reinitializes", async () => { - const { queue, scheduler } = createQueue(); - const result = await queue.flushQueues(); - expect(scheduler.stop).toHaveBeenCalled(); - expect(scheduler.flushJobs).toHaveBeenCalled(); - 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); - const result = await queue.flushQueues(); - expect(result.success).toBe(false); - }); - }); - - // ── static create ────────────────────────────────────────────────────────── - - describe("static create", () => { - const envSettings = { dbConnectionString: "mongodb://localhost:27017/test_db" } as any; - - it("creates a MongoStore + Scheduler and inits in primary mode", async () => { - 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, - workersRepository as any, - envSettings, - "primary" - ); - - expect(MockMongoStore).toHaveBeenCalledWith(expect.objectContaining({ url: "mongodb://localhost:27017/test_db" })); - expect(MockScheduler).toHaveBeenCalled(); - expect(instance).toBeInstanceOf(LessSimpleQueue); - expect(mockSchedulerInstance.start).toHaveBeenCalled(); - expect(monitorsRepository.findAll).toHaveBeenCalled(); - }); - - it("inits in worker mode without seeding monitors", async () => { - 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, - 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 deleted file mode 100644 index b9e9b8314..000000000 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ /dev/null @@ -1,712 +0,0 @@ -import { describe, expect, it, jest, beforeEach } from "@jest/globals"; -import { createMockLogger } from "../../helpers/createMockLogger.ts"; -import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; - -// ── Mock Scheduler ─────────────────────────────────────────────────────────── - -const createScheduler = () => { - const listeners: Record = {}; - return { - on: jest.fn((event: string, cb: Function) => { - listeners[event] = listeners[event] || []; - listeners[event].push(cb); - }), - emit: (event: string, ...args: unknown[]) => { - (listeners[event] || []).forEach((cb) => cb(...args)); - }, - start: jest.fn(), - stop: jest.fn().mockResolvedValue(true), - addTemplate: jest.fn(), - addJob: jest.fn(), - removeJob: jest.fn(), - getJob: jest.fn().mockResolvedValue(null), - getJobs: jest.fn().mockResolvedValue([]), - pauseJob: jest.fn().mockResolvedValue(true), - resumeJob: jest.fn().mockResolvedValue(true), - updateJob: jest.fn(), - flushJobs: jest.fn().mockResolvedValue(true), - }; -}; - -const mockSchedulerInstance = createScheduler(); -const MockScheduler = jest.fn().mockReturnValue(mockSchedulerInstance); - -jest.unstable_mockModule("super-simple-scheduler", () => ({ - default: MockScheduler, -})); - -const { SuperSimpleQueue } = await import("../../../src/worker/worker.super-simple.ts"); - -const createQueueHelper = () => ({ - getHeartbeatJob: jest.fn().mockReturnValue(() => Promise.resolve()), - getHeartbeatGeoJob: jest.fn().mockReturnValue(() => Promise.resolve()), - getCleanupOrphanedJob: jest.fn().mockReturnValue(() => Promise.resolve()), - getCleanupRetentionJob: jest.fn().mockReturnValue(() => Promise.resolve()), -}); - -const createMonitorsRepo = () => ({ - findAll: jest.fn().mockResolvedValue([]), -}); - -const createQueue = (overrides?: Record) => { - const logger = createMockLogger(); - const helper = createQueueHelper(); - const monitorsRepository = createMonitorsRepo(); - const scheduler = createScheduler(); - - const defaults = { logger, helper, monitorsRepository, scheduler, ...overrides }; - - const queue = new SuperSimpleQueue(defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.scheduler as any); - return { queue, ...defaults }; -}; - -const makeMonitor = (overrides?: Partial): Monitor => - ({ - id: "mon-1", - teamId: "team-1", - type: "http", - interval: 60000, - isActive: true, - geoCheckEnabled: false, - geoCheckInterval: 300000, - geoCheckLocations: [], - ...overrides, - }) as Monitor; - -// ── Tests ──────────────────────────────────────────────────────────────────── - -describe("SuperSimpleQueue", () => { - describe("serviceName", () => { - it("returns JobQueue", () => { - const { queue } = createQueue(); - expect(queue.serviceName).toBe("JobQueue"); - }); - }); - - // ── init ───────────────────────────────────────────────────────────────── - - describe("init", () => { - it("registers listeners, starts scheduler, adds templates and system jobs", async () => { - const { queue, scheduler, helper } = createQueue(); - - const result = await queue.init(); - - expect(result).toBe(true); - expect(scheduler.start).toHaveBeenCalled(); - expect(scheduler.addTemplate).toHaveBeenCalledWith("monitor-job", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("geo-check-job", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-orphaned", expect.any(Function)); - expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-retention-job", expect.any(Function)); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" })); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-retention" })); - }); - - it("schedules existing monitors with staggered offsets", async () => { - jest.useFakeTimers(); - const monitors = [makeMonitor({ id: "m1" }), makeMonitor({ id: "m2" })]; - const { queue, scheduler, monitorsRepository } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockResolvedValue(monitors); - - await queue.init(); - jest.runAllTimers(); - - // Each monitor gets addJob called (once for the monitor itself via addJob method) - // The init schedules via setTimeout → addJob, plus the two system jobs - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m1", template: "monitor-job" })); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m2", template: "monitor-job" })); - jest.useRealTimers(); - }); - - it("returns true when findAll returns null", async () => { - const { queue, monitorsRepository, scheduler } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockResolvedValue(null); - - const result = await queue.init(); - - expect(result).toBe(true); - // System jobs should not be added when monitors is null (early return) - expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" })); - }); - - it("returns false and logs error on failure", async () => { - const { queue, logger, monitorsRepository } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockRejectedValue(new Error("db down")); - - const result = await queue.init(); - - expect(result).toBe(false); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("db down") })); - }); - - it("logs error with String(error) for non-Error exceptions", async () => { - const { queue, logger, monitorsRepository } = createQueue(); - (monitorsRepository.findAll as jest.Mock).mockRejectedValue("string error"); - - const result = await queue.init(); - - expect(result).toBe(false); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("string error") })); - }); - }); - - // ── registerListeners ──────────────────────────────────────────────────── - - describe("registerListeners (via init)", () => { - it("logs on scheduler:start", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("scheduler:start"); - expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler started" })); - }); - - it("logs on scheduler:stop", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("scheduler:stop"); - expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler stopped" })); - }); - - it("logs on scheduler:error with Error", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("scheduler:error", new Error("boom")); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler error: boom" })); - }); - - it("logs on scheduler:error with non-Error", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("scheduler:error", "string error"); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler error: string error", stack: undefined })); - }); - - it("logs on job:abort", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:abort", { id: "m1" }, "timeout"); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 aborted: timeout" })); - }); - - it("logs on job:attempt", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:attempt", { id: "m1" }, 2); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 attempt 2" })); - }); - - it("logs on job:complete", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:complete", { id: "m1" }); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 completed successfully" })); - }); - - it("logs on job:exhausted with Error", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:exhausted", { id: "m1" }, new Error("gave up")); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 exhausted all retries: gave up" })); - }); - - it("logs on job:exhausted with non-Error", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:exhausted", { id: "m1" }, "gave up"); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 exhausted all retries: gave up", stack: undefined })); - }); - - it("logs on job:fail with Error", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:fail", { id: "m1" }, new Error("oops"), 1); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 failed on attempt 1: oops" })); - }); - - it("logs on job:fail with non-Error", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:fail", { id: "m1" }, "oops", 1); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 failed on attempt 1: oops", stack: undefined })); - }); - - it("logs on job:start", async () => { - const { queue, scheduler, logger } = createQueue(); - await queue.init(); - scheduler.emit("job:start", { id: "m1" }); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "m1 started" })); - }); - }); - - // ── addJob ─────────────────────────────────────────────────────────────── - - describe("addJob", () => { - it("adds monitor job to scheduler", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor()); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1", template: "monitor-job", repeat: 60000, active: true })); - }); - - it("adds geo check job when geoCheckEnabled and type supports it", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 })); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job", repeat: 300000 })); - }); - - it("skips geo job when type does not support geo checks", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: true, type: "hardware" })); - expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo" })); - }); - - it("skips geo job when geoCheckEnabled is false", async () => { - const { queue, scheduler } = createQueue(); - await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: false, type: "http" })); - expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo" })); - }); - }); - - // ── deleteJob ──────────────────────────────────────────────────────────── - - describe("deleteJob", () => { - it("removes monitor job", async () => { - const { queue, scheduler } = createQueue(); - await queue.deleteJob(makeMonitor()); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1"); - }); - - it("removes geo job if it exists", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.deleteJob(makeMonitor()); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - - it("does not remove geo job if it does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.deleteJob(makeMonitor()); - expect(scheduler.removeJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - }); - - // ── pauseJob ───────────────────────────────────────────────────────────── - - describe("pauseJob", () => { - it("pauses monitor and pauses geo job when geo exists", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.pauseJob(makeMonitor()); - - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1-geo"); - expect(scheduler.removeJob).not.toHaveBeenCalledWith("mon-1-geo"); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "Paused monitor mon-1" })); - }); - - it("refreshes scheduler data with current monitor on pause", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - const monitor = makeMonitor({ isActive: false, status: "paused" }); - - await queue.pauseJob(monitor); - - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", { data: monitor }); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1-geo", { data: monitor }); - }); - - it("does not pause geo job when it does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.pauseJob(makeMonitor()); - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.pauseJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - - it("does not refresh geo data when geo job does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.pauseJob(makeMonitor()); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", expect.objectContaining({ data: expect.anything() })); - expect(scheduler.updateJob).not.toHaveBeenCalledWith("mon-1-geo", expect.anything()); - }); - - it("throws when scheduler.pauseJob returns false for the main job", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.pauseJob as jest.Mock).mockResolvedValue(false); - await expect(queue.pauseJob(makeMonitor())).rejects.toThrow("Failed to pause monitor"); - }); - - it("does not refresh data when main pauseJob throws", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.pauseJob as jest.Mock).mockResolvedValue(false); - await expect(queue.pauseJob(makeMonitor())).rejects.toThrow(); - expect(scheduler.updateJob).not.toHaveBeenCalled(); - }); - - it("logs error but does not throw when geo pauseJob returns false", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - (scheduler.pauseJob as jest.Mock).mockImplementation(async (id: string) => id !== "mon-1-geo"); - - await expect(queue.pauseJob(makeMonitor())).resolves.toBeUndefined(); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Failed to pause geo check job for monitor mon-1" })); - }); - }); - - // ── resumeJob ──────────────────────────────────────────────────────────── - - describe("resumeJob", () => { - it("resumes monitor and geo job when geo exists", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.resumeJob(makeMonitor()); - - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1-geo"); - expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "Resumed monitor mon-1" })); - }); - - it("refreshes scheduler data with current monitor on resume", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - const monitor = makeMonitor({ isActive: true, status: "initializing" }); - - await queue.resumeJob(monitor); - - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", { data: monitor }); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1-geo", { data: monitor }); - }); - - it("does not resume geo job when it does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.resumeJob(makeMonitor()); - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1"); - expect(scheduler.resumeJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - - it("does not refresh geo data when geo job does not exist", async () => { - const { queue, scheduler } = createQueue(); - await queue.resumeJob(makeMonitor()); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", expect.objectContaining({ data: expect.anything() })); - expect(scheduler.updateJob).not.toHaveBeenCalledWith("mon-1-geo", expect.anything()); - }); - - it("throws when scheduler.resumeJob returns false for the main job", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.resumeJob as jest.Mock).mockResolvedValue(false); - await expect(queue.resumeJob(makeMonitor())).rejects.toThrow("Failed to resume monitor"); - }); - - it("does not refresh data when main resumeJob throws", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.resumeJob as jest.Mock).mockResolvedValue(false); - await expect(queue.resumeJob(makeMonitor())).rejects.toThrow(); - expect(scheduler.updateJob).not.toHaveBeenCalled(); - }); - - it("logs error but does not throw when geo resumeJob returns false", async () => { - const { queue, scheduler, logger } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - (scheduler.resumeJob as jest.Mock).mockImplementation(async (id: string) => id !== "mon-1-geo"); - - await expect(queue.resumeJob(makeMonitor())).resolves.toBeUndefined(); - expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Failed to resume geo check job for monitor mon-1" })); - }); - - it("preserves geo job through a pause/resume cycle", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - - await queue.pauseJob(makeMonitor()); - await queue.resumeJob(makeMonitor()); - - expect(scheduler.removeJob).not.toHaveBeenCalledWith("mon-1-geo"); - expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1-geo"); - expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - }); - - // ── updateJob ──────────────────────────────────────────────────────────── - - describe("updateJob", () => { - it("updates monitor job and syncs geo job", async () => { - const { queue, scheduler } = createQueue(); - const monitor = makeMonitor({ interval: 120000 }); - await queue.updateJob(monitor); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", expect.objectContaining({ repeat: 120000, data: monitor })); - }); - }); - - // ── syncGeoJob (via updateJob) ─────────────────────────────────────────── - - describe("syncGeoJob (via updateJob)", () => { - it("removes existing geo job when geoCheckEnabled is false", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.updateJob(makeMonitor({ geoCheckEnabled: false, type: "http" })); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - - it("removes existing geo job when type does not support geo checks", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - await queue.updateJob(makeMonitor({ geoCheckEnabled: true, type: "hardware" })); - expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo"); - }); - - it("does not remove when geo disabled and no existing geo job", async () => { - const { queue, scheduler } = createQueue(); - await queue.updateJob(makeMonitor({ geoCheckEnabled: false })); - expect(scheduler.removeJob).not.toHaveBeenCalledWith("mon-1-geo"); - }); - - it("updates existing geo job when geoCheckEnabled", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" }); - const monitor = makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 }); - await queue.updateJob(monitor); - expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1-geo", expect.objectContaining({ repeat: 300000 })); - }); - - it("creates new geo job when enabled but no existing job", async () => { - const { queue, scheduler } = createQueue(); - const monitor = makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 }); - await queue.updateJob(monitor); - expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job" })); - }); - }); - - // ── shutdown ────────────────────────────────────────────────────────────── - - describe("shutdown", () => { - it("stops the scheduler", async () => { - const { queue, scheduler } = createQueue(); - await queue.shutdown(); - expect(scheduler.stop).toHaveBeenCalled(); - }); - }); - - // ── getMetrics ─────────────────────────────────────────────────────────── - - describe("getMetrics", () => { - 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: [ - { - // 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 () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", runCount: 10, failCount: 2, lastFailedAt: 200, lastRunAt: 100, lockedAt: null, data: { url: "http://a.com", type: "http" } }, - { id: "m2", runCount: 5, failCount: 0, lastFailedAt: 0, lastRunAt: 50, lockedAt: 999, data: { url: "http://b.com", type: "ping" } }, - ]); - const metrics = await queue.getMetrics(); - - expect(metrics.jobs).toBe(2); - expect(metrics.totalRuns).toBe(15); - expect(metrics.totalFailures).toBe(2); - expect(metrics.activeJobs).toBe(1); - expect(metrics.failingJobs).toBe(1); - expect(metrics.jobsWithFailures).toHaveLength(1); - expect(metrics.jobsWithFailures[0]).toEqual(expect.objectContaining({ monitorId: "m1", failCount: 2, monitorUrl: "http://a.com" })); - }); - - it("handles jobs with undefined fields", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "m1", data: null }]); - const metrics = await queue.getMetrics(); - expect(metrics.jobs).toBe(1); - expect(metrics.totalRuns).toBe(0); - }); - - it("handles failing job with null data", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", runCount: 1, failCount: 1, lastFailedAt: 100, lastRunAt: 50, data: null, lastFailReason: "timeout" }, - ]); - const metrics = await queue.getMetrics(); - expect(metrics.jobsWithFailures).toHaveLength(1); - expect(metrics.jobsWithFailures[0].monitorUrl).toBeNull(); - expect(metrics.jobsWithFailures[0].monitorType).toBeNull(); - expect(metrics.jobsWithFailures[0].failReason).toBe("timeout"); - }); - - it("handles failing job with undefined lastFailedAt and lastFailReason", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "m1", runCount: 1, failCount: 1, data: { url: "http://a.com", type: "http" } }]); - const metrics = await queue.getMetrics(); - expect(metrics.jobsWithFailures[0].failedAt).toBeNull(); - expect(metrics.jobsWithFailures[0].failReason).toBeNull(); - }); - - it("does not count as failing when lastFailedAt < lastRunAt", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", runCount: 10, failCount: 1, lastFailedAt: 50, lastRunAt: 100, data: { url: "a" } }, - ]); - const metrics = await queue.getMetrics(); - expect(metrics.failingJobs).toBe(0); - expect(metrics.jobsWithFailures).toHaveLength(1); - }); - }); - - // ── getJobs ────────────────────────────────────────────────────────────── - - describe("getJobs", () => { - it("maps scheduler jobs to summaries", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { - id: "m1", - active: true, - repeat: 60000, - lockedAt: null, - runCount: 5, - failCount: 1, - lastFailReason: "timeout", - lastRunAt: 1000, - lastFinishedAt: 1100, - lastFailedAt: 900, - data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true }, - }, - ]); - const { jobs } = await queue.getJobs({}); - - expect(jobs).toHaveLength(1); - expect(jobs[0]).toEqual({ - monitorId: "m1", - monitorUrl: "http://a.com", - monitorType: "http", - monitorInterval: 60000, - monitorGeoInterval: 300000, - monitorActive: true, - active: true, - repeat: 60000, - lockedBy: null, - lockedUntil: null, - lockedAt: null, - runCount: 5, - failCount: 1, - failReason: "timeout", - lastRunAt: 1000, - lastFinishedAt: 1100, - lastRunTook: 100, - lastFailedAt: 900, - }); - }); - - it("returns null lastRunTook when job is locked", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", active: true, lockedAt: 999, runCount: 1, data: { url: "a", type: "http", interval: 60000 } }, - ]); - const { jobs } = await queue.getJobs({}); - expect(jobs[0].lastRunTook).toBeNull(); - }); - - it("handles jobs with missing data", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "m1", active: true, data: null }]); - const { jobs } = await queue.getJobs({}); - expect(jobs[0].monitorUrl).toBeNull(); - expect(jobs[0].monitorType).toBeNull(); - expect(jobs[0].monitorInterval).toBeNull(); - expect(jobs[0].monitorGeoInterval).toBeNull(); - expect(jobs[0].monitorActive).toBeNull(); - expect(jobs[0].repeat).toBeNull(); - }); - - it("exposes monitor.isActive as monitorActive in the summary", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", active: true, data: { url: "a", type: "http", interval: 60000, isActive: true } }, - { id: "m2", active: false, data: { url: "b", type: "http", interval: 60000, isActive: false } }, - ]); - const { jobs } = await queue.getJobs({}); - expect(jobs[0].monitorActive).toBe(true); - expect(jobs[1].monitorActive).toBe(false); - }); - - it("exposes geo check interval and repeat", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue([ - { id: "m1", active: true, repeat: 120000, data: { url: "a", type: "http", interval: 60000, geoCheckInterval: 600000 } }, - ]); - const { jobs } = await queue.getJobs({}); - expect(jobs[0].repeat).toBe(120000); - expect(jobs[0].monitorGeoInterval).toBe(600000); - }); - - it("paginates jobs and returns the total count", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue( - Array.from({ length: 25 }, (_, i) => ({ id: `m${i}`, active: true, data: { url: `a${i}`, type: "http", interval: 60000 } })) - ); - - const { jobs, count } = await queue.getJobs({ page: 1, rowsPerPage: 10 }); - expect(count).toBe(25); - expect(jobs).toHaveLength(10); - expect(jobs[0].monitorId).toBe("m10"); - }); - - it("returns all jobs when rowsPerPage is omitted", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.getJobs as jest.Mock).mockResolvedValue( - Array.from({ length: 25 }, (_, i) => ({ id: `m${i}`, active: true, data: { url: `a${i}`, type: "http", interval: 60000 } })) - ); - - const { jobs, count } = await queue.getJobs({}); - expect(count).toBe(25); - expect(jobs).toHaveLength(25); - }); - }); - - // ── flushQueues ────────────────────────────────────────────────────────── - - describe("flushQueues", () => { - it("stops, flushes, and reinitializes", async () => { - const { queue, scheduler } = createQueue(); - const result = await queue.flushQueues(); - expect(scheduler.stop).toHaveBeenCalled(); - expect(scheduler.flushJobs).toHaveBeenCalled(); - expect(result.success).toBe(true); - }); - - it("returns false when any step fails", async () => { - const { queue, scheduler } = createQueue(); - (scheduler.flushJobs as jest.Mock).mockResolvedValue(false); - const result = await queue.flushQueues(); - expect(result.success).toBe(false); - }); - }); - - // ── static create ──────────────────────────────────────────────────────── - - describe("static create", () => { - it("creates a SuperSimpleQueue instance with a Scheduler and calls init", async () => { - const logger = createMockLogger(); - const helper = createQueueHelper(); - const monitorsRepository = createMonitorsRepo(); - - const instance = await SuperSimpleQueue.create(logger as any, helper as any, monitorsRepository as any); - - expect(MockScheduler).toHaveBeenCalled(); - expect(instance).toBeInstanceOf(SuperSimpleQueue); - // init was called, which calls scheduler.start - expect(mockSchedulerInstance.start).toHaveBeenCalled(); - }); - }); -}); diff --git a/server/test/unit/services/superSimpleQueueHelper.test.ts b/server/test/unit/services/workerHelper.test.ts similarity index 100% rename from server/test/unit/services/superSimpleQueueHelper.test.ts rename to server/test/unit/services/workerHelper.test.ts