From 9d969d67ca8d0d91643a5ae955b4e5ebe2c5431e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 25 Jun 2026 11:55:40 -0700 Subject: [PATCH] udpate tests --- .../test/integration/jobsRepository.test.ts | 36 ++++++ .../integration/schedulerReconcile.test.ts | 107 ++++++++++++++++ .../test/unit/services/dbQueueWorker.test.ts | 35 +++++- .../test/unit/services/healthServer.test.ts | 47 ++++++- .../test/unit/services/jobScheduler.test.ts | 116 ++++++++++++++++-- 5 files changed, 324 insertions(+), 17 deletions(-) create mode 100644 server/test/integration/schedulerReconcile.test.ts diff --git a/server/test/integration/jobsRepository.test.ts b/server/test/integration/jobsRepository.test.ts index 02b0ecc5d..78ac17b64 100644 --- a/server/test/integration/jobsRepository.test.ts +++ b/server/test/integration/jobsRepository.test.ts @@ -497,4 +497,40 @@ describe("MongoJobsRepository", () => { expect(await repo.findAll()).toHaveLength(2); }); }); + + // ── countDueBacklog (scaling signal) ───────────────────────────────────────── + describe("countDueBacklog", () => { + it("counts a due, active, unlocked check job", async () => { + await seedJob(); + expect(await repo.countDueBacklog(NOW)).toBe(1); + }); + + it("counts a check job whose lock has expired (reclaimable work)", async () => { + await seedJob({ lockedBy: "dead-worker", lockedUntil: NOW - 1 }); + expect(await repo.countDueBacklog(NOW)).toBe(1); + }); + + it("excludes not-yet-due, inactive, live-locked, and non-check rows", async () => { + await seedJob({ _id: "check:due", refId: "due" }); // counts + await seedJob({ _id: "check:future", refId: "future", nextScheduledAt: NOW + 1 }); // not due + await seedJob({ _id: "check:paused", refId: "paused", isActive: false }); // inactive + await seedJob({ _id: "check:locked", refId: "locked", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // live lock + await seedJob({ _id: "geo:mon", refId: "mon", type: "geo-check" }); // wrong type + await seedJob({ _id: "evaluate:mon", refId: "mon", type: "evaluate" }); // wrong type + + expect(await repo.countDueBacklog(NOW)).toBe(1); // only check:due + }); + + it("counts every qualifying check row across monitors", async () => { + await seedJob({ _id: "check:a", refId: "a" }); + await seedJob({ _id: "check:b", refId: "b" }); + await seedJob({ _id: "check:c", refId: "c", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // excluded + expect(await repo.countDueBacklog(NOW)).toBe(2); + }); + + it("returns 0 when nothing is due", async () => { + await seedJob({ nextScheduledAt: NOW + 10_000 }); + expect(await repo.countDueBacklog(NOW)).toBe(0); + }); + }); }); diff --git a/server/test/integration/schedulerReconcile.test.ts b/server/test/integration/schedulerReconcile.test.ts new file mode 100644 index 000000000..8b1a48f79 --- /dev/null +++ b/server/test/integration/schedulerReconcile.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it, beforeAll, afterAll, beforeEach } from "@jest/globals"; +import mongoose from "mongoose"; +import { MongoMemoryServer } from "mongodb-memory-server"; +import { JobScheduler } from "../../src/worker/worker.job-scheduler.ts"; +import MongoJobsRepository from "../../src/domain/jobs/job.repository.mongo.ts"; +import MongoQueueWorkersRepository from "../../src/domain/queue-workers/queue-worker.repository.mongo.ts"; +import MongoMonitorsRepository from "../../src/domain/monitors/monitor.repository.mongo.ts"; +import JobModel, { type JobDocument } from "../../src/domain/jobs/job.model.ts"; +import { MonitorModel } from "../../src/domain/monitors/monitor.model.ts"; +import { QueueWorkerModel } from "../../src/domain/queue-workers/queue-worker.model.ts"; +import { createMockLogger } from "../helpers/createMockLogger.ts"; + +// ── Real-Mongo harness ───────────────────────────────────────────────────────── +// The bug: a scheduler-only primary (queueMode=primary, queuePrimaryProcesses=false) +// never seeded the jobs collection, so a fresh start with existing monitors left the +// workers idle. The fix moved reconcile/heartbeat onto the base JobScheduler.init(). +// This proves init() against a live mongod actually writes the expected job rows. + +let mongod: MongoMemoryServer; + +beforeAll(async () => { + mongod = await MongoMemoryServer.create(); + await mongoose.connect(mongod.getUri()); + await JobModel.init(); +}, 120_000); + +afterAll(async () => { + await mongoose.disconnect(); + await mongod.stop(); +}); + +beforeEach(async () => { + await JobModel.deleteMany({}); + await MonitorModel.collection.deleteMany({}); + await QueueWorkerModel.deleteMany({}); +}); + +const WORKER_ID = "scheduler-primary-1"; + +const seedMonitor = (overrides: Record = {}) => + MonitorModel.create({ + userId: new mongoose.Types.ObjectId(), + teamId: new mongoose.Types.ObjectId(), + name: "mon", + type: "http", + url: "https://example.com", + interval: 60_000, + isActive: true, + ...overrides, + }); + +const makeScheduler = () => + new JobScheduler( + new MongoJobsRepository(WORKER_ID), + new MongoQueueWorkersRepository(), + new MongoMonitorsRepository(), + "primary", + createMockLogger(), + WORKER_ID + ); + +const idsOf = async () => (await JobModel.find().lean()).map((j) => j._id); + +describe("JobScheduler.init (scheduler-only primary reconcile)", () => { + it("seeds a check row per monitor plus the two cleanup rows", async () => { + const m1 = await seedMonitor(); + const m2 = await seedMonitor({ name: "mon-2" }); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); // stop the heartbeat interval + + const ids = await idsOf(); + expect(ids).toContain(`check:${m1.id}`); + expect(ids).toContain(`check:${m2.id}`); + expect(ids).toContain("cleanup-orphaned"); + expect(ids).toContain("cleanup-retention"); + }); + + it("seeds a geo-check row only for geo-enabled monitors", async () => { + const geo = await seedMonitor({ name: "geo", geoCheckEnabled: true }); + const plain = await seedMonitor({ name: "plain", geoCheckEnabled: false }); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); + + const ids = await idsOf(); + expect(ids).toContain(`geo-check:${geo.id}`); + expect(ids).not.toContain(`geo-check:${plain.id}`); + }); + + it("registers the primary in the queue_workers registry", async () => { + await seedMonitor(); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); + + // shutdown deregisters, so assert mid-flight: re-init then read before shutdown. + const scheduler2 = makeScheduler(); + await scheduler2.init(); + const row = await QueueWorkerModel.findById(WORKER_ID).lean(); + expect(row?.mode).toBe("primary"); + await scheduler2.shutdown(); + }); +}); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index dd210b3e2..cabf90e31 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -79,6 +79,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses updateScheduleById: jest.fn().mockResolvedValue(true), findAll: jest.fn().mockResolvedValue([]), findPage: jest.fn().mockResolvedValue({ jobs: [], count: 0 }), + countDueBacklog: jest.fn().mockResolvedValue(0), }; const monitorsRepository = { findByIds: jest.fn().mockResolvedValue([makeMonitor()]), @@ -186,7 +187,7 @@ describe("DBQueueWorker", () => { describe("init", () => { it("registers the worker in the heartbeat registry on startup", async () => { const { mocks } = await start({ queueMode: "worker" }); - expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker"); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true); }); it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => { @@ -263,7 +264,7 @@ describe("DBQueueWorker", () => { active = worker; await jest.advanceTimersByTimeAsync(1); - expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker"); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true); }); }); @@ -768,6 +769,29 @@ describe("DBQueueWorker", () => { }); }); + // ── metrics accessors ─────────────────────────────────────────────────────────── + + describe("metrics accessors", () => { + it("countDueBacklog delegates to the repo with the current time", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + mocks.jobsRepository.countDueBacklog.mockResolvedValue(7); + + expect(await worker.countDueBacklog()).toBe(7); + expect(mocks.jobsRepository.countDueBacklog).toHaveBeenCalledWith(expect.any(Number)); + }); + + it("countAliveWorkers counts only recently-seen nodes that process jobs", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + mocks.queueWorkersRepository.findRecent.mockResolvedValue([ + { processesJobs: true }, + { processesJobs: true }, + { processesJobs: false }, // scheduler-only primary: alive but not a processor + ]); + + expect(await worker.countAliveWorkers()).toBe(2); + }); + }); + // ── flushQueues ─────────────────────────────────────────────────────────────── describe("flushQueues", () => { @@ -799,7 +823,10 @@ describe("DBQueueWorker", () => { const { worker, mocks } = createWorker(); active = worker; mocks.jobsRepository.findAll.mockResolvedValue(rows); - mocks.queueWorkersRepository.findRecent.mockResolvedValue([{ workerId: "w" }]); + mocks.queueWorkersRepository.findRecent.mockResolvedValue([ + { workerId: "w", processesJobs: true }, + { workerId: "scheduler", processesJobs: false }, // scheduler-only primary excluded from metrics + ]); const metrics = await worker.getMetrics(); @@ -810,7 +837,7 @@ describe("DBQueueWorker", () => { expect(metrics.failingJobs).toBe(1); expect(metrics.jobsWithFailures).toHaveLength(1); expect(metrics.jobsWithFailures[0]).toEqual(expect.objectContaining({ failCount: 3, failReason: "boom" })); - expect(metrics.workers).toEqual([{ workerId: "w" }]); + expect(metrics.workers).toEqual([{ workerId: "w", processesJobs: true }]); }); }); diff --git a/server/test/unit/services/healthServer.test.ts b/server/test/unit/services/healthServer.test.ts index 208cd0902..1ca62d4bb 100644 --- a/server/test/unit/services/healthServer.test.ts +++ b/server/test/unit/services/healthServer.test.ts @@ -26,13 +26,13 @@ const freePort = (): Promise => }); }); -const get = (port: number, path: string): Promise<{ status: number; body: string }> => +const get = (port: number, path: string): Promise<{ status: number; body: string; contentType: string }> => new Promise((resolve, reject) => { http .get({ host: "127.0.0.1", port, path }, (res) => { let body = ""; res.on("data", (chunk) => (body += chunk)); - res.on("end", () => resolve({ status: res.statusCode ?? 0, body })); + res.on("end", () => resolve({ status: res.statusCode ?? 0, body, contentType: String(res.headers["content-type"] ?? "") })); }) .on("error", reject); }); @@ -58,9 +58,16 @@ describe("HealthServer", () => { active = null; }); - const setup = async (health: WorkerHealth) => { + // The metrics endpoint also calls countDueBacklog()/countAliveWorkers(); default them + // to 0 and let tests override (including to a rejection, to exercise the 500 path). + const setup = async (health: WorkerHealth, workerOverrides: Record = {}) => { const port = await freePort(); - const worker = { getHealth: () => health } as any; + const worker = { + getHealth: () => health, + countDueBacklog: async () => 0, + countAliveWorkers: async () => 0, + ...workerOverrides, + } as any; const server = new HealthServer(createMockLogger(), String(port), worker); await server.listen(); active = server; @@ -133,4 +140,36 @@ describe("HealthServer", () => { expect((await get(port, "/livez?probe=1")).status).toBe(200); }); }); + + describe("/metrics", () => { + it("200 with prometheus content-type and all four gauges", async () => { + const { port } = await setup(makeHealth({ inFlight: 3, draining: false }), { + countDueBacklog: async () => 5, + countAliveWorkers: async () => 2, + }); + const res = await get(port, "/metrics"); + + expect(res.status).toBe(200); + expect(res.contentType).toMatch(/text\/plain/); + expect(res.body).toContain("# TYPE checkmate_worker_due_backlog gauge"); + expect(res.body).toContain("checkmate_worker_jobs_in_flight 3"); + expect(res.body).toContain("checkmate_worker_due_backlog 5"); + expect(res.body).toContain("checkmate_worker_alive_total 2"); + expect(res.body).toContain("checkmate_worker_draining 0"); + }); + + it("reports draining as 1 when the worker is draining", async () => { + const { port } = await setup(makeHealth({ draining: true })); + expect((await get(port, "/metrics")).body).toContain("checkmate_worker_draining 1"); + }); + + it("500 when a metrics query fails", async () => { + const { port } = await setup(makeHealth(), { + countDueBacklog: async () => { + throw new Error("db down"); + }, + }); + expect((await get(port, "/metrics")).status).toBe(500); + }); + }); }); diff --git a/server/test/unit/services/jobScheduler.test.ts b/server/test/unit/services/jobScheduler.test.ts index dceb8760c..90d7a2b99 100644 --- a/server/test/unit/services/jobScheduler.test.ts +++ b/server/test/unit/services/jobScheduler.test.ts @@ -1,26 +1,124 @@ -import { describe, expect, it, jest } from "@jest/globals"; +import { describe, expect, it, jest, beforeEach, afterEach } from "@jest/globals"; import { JobScheduler } from "../../../src/worker/worker.job-scheduler.ts"; +import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; +import type { QueueMode } from "../../../src/domain/app-settings/app-settings.type.ts"; import { createMockLogger } from "../../helpers/createMockLogger.ts"; // ── Notes ────────────────────────────────────────────────────────────────────── // // JobScheduler is the scheduler-only primary base class: it owns the queue -// registry/heartbeat but processes no jobs and has no buffer. drain() is therefore -// a no-op beyond setting the stop flag — it must NOT tear down (clear timers or -// deregister the worker); that stays in shutdown(). The real draining (wait for -// in-flight, flush buffer) lives in DBQueueWorker and is covered separately. +// registry/heartbeat AND, on a primary, the reconcile that seeds the jobs +// collection — scheduling is its job whether or not it also processes. It runs no +// polling loops and has no buffer, so drain() is a no-op beyond setting the stop +// flag — it must NOT tear down (clear timers or deregister the worker); that stays +// in shutdown(). The processing tier (poll loops, buffer flush) lives in +// DBQueueWorker and is covered separately. -const createScheduler = () => { - const jobsRepository = {} as any; +const makeMonitor = (overrides?: Partial): Monitor => + ({ + id: "m1", + teamId: "team", + type: "http", + interval: 60000, + status: "up", + isActive: true, + geoCheckEnabled: false, + geoCheckInterval: 300000, + lastEvaluatedAt: 0, + ...overrides, + }) as Monitor; + +const createScheduler = (overrides?: { queueMode?: QueueMode; mocks?: Record }) => { + const jobsRepository = { + upsertJob: jest.fn().mockResolvedValue(true), + upsertCleanupJob: jest.fn().mockResolvedValue(true), + }; const queueWorkersRepository = { + upsert: jest.fn().mockResolvedValue(undefined), deleteById: jest.fn().mockResolvedValue(undefined), }; + const monitorsRepository = { + findAll: jest.fn().mockResolvedValue([]), + }; const logger = createMockLogger(); - const scheduler = new JobScheduler(jobsRepository, queueWorkersRepository as any, "worker-1"); - return { scheduler, mocks: { jobsRepository, queueWorkersRepository, logger } }; + const mocks = { jobsRepository, queueWorkersRepository, monitorsRepository, logger, ...overrides?.mocks }; + const scheduler = new JobScheduler( + mocks.jobsRepository as any, + mocks.queueWorkersRepository as any, + mocks.monitorsRepository as any, + overrides?.queueMode ?? "primary", + mocks.logger as any, + "worker-1" + ); + return { scheduler, mocks }; }; describe("JobScheduler", () => { + describe("init", () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(async () => { + jest.clearAllTimers(); + jest.useRealTimers(); + }); + + it("registers the worker in the heartbeat registry on startup", async () => { + const { scheduler, mocks } = createScheduler({ queueMode: "primary" }); + await scheduler.init(); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "primary", false); + }); + + it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "primary", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor()]) } }, + }); + await scheduler.init(); + + const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type); + expect(seededTypes).toContain("check"); + const cleanupTypes = mocks.jobsRepository.upsertCleanupJob.mock.calls.map((c: any[]) => c[0].type); + expect(cleanupTypes).toContain("cleanup-orphaned"); + expect(cleanupTypes).toContain("cleanup-retention"); + }); + + it("primary mode also seeds a geo-check row for geo-enabled monitors", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "primary", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor({ geoCheckEnabled: true })]) } }, + }); + await scheduler.init(); + + const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type); + expect(seededTypes).toContain("geo-check"); + }); + + it("worker mode heartbeats but does not reconcile (no monitor reads, no seeds)", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "worker", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor()]) } }, + }); + await scheduler.init(); + + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", false); + expect(mocks.monitorsRepository.findAll).not.toHaveBeenCalled(); + expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled(); + expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled(); + }); + + it("arms a heartbeat interval timer that shutdown() clears and deregisters", async () => { + const { scheduler, mocks } = createScheduler({ queueMode: "primary" }); + await scheduler.init(); + expect((scheduler as any).timers.has("heartbeat")).toBe(true); + + await scheduler.shutdown(); + expect((scheduler as any).timers.has("heartbeat")).toBe(false); + expect(mocks.queueWorkersRepository.deleteById).toHaveBeenCalledWith("worker-1"); + }); + }); + describe("drain", () => { it("sets the stop and draining flags without tearing down (no deregister)", async () => { const { scheduler, mocks } = createScheduler();