From 3b919044e6b7bf3616c4965cc1be0e9e2abb3e96 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 19 Jun 2026 10:25:59 -0700 Subject: [PATCH] wake worker on job change --- server/src/config/services.ts | 1 + .../maintenance-window.service.ts | 8 ++ server/src/worker/worker.db-queue.ts | 46 ++++++++++-- server/src/worker/worker.interface.ts | 2 + .../test/unit/services/dbQueueWorker.test.ts | 74 +++++++++++++++++++ .../services/maintenanceWindowService.test.ts | 27 ++++++- 6 files changed, 149 insertions(+), 9 deletions(-) diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 9de13f27b..2f1826241 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -334,6 +334,7 @@ export const initializeServices = async ({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, + worker, }); const monitorService = new MonitorService({ worker, diff --git a/server/src/domain/maintenance-windows/maintenance-window.service.ts b/server/src/domain/maintenance-windows/maintenance-window.service.ts index aef0f9fb4..b8601321f 100644 --- a/server/src/domain/maintenance-windows/maintenance-window.service.ts +++ b/server/src/domain/maintenance-windows/maintenance-window.service.ts @@ -1,6 +1,7 @@ import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { IWorker } from "@/worker/worker.interface.js"; import type { DurationUnit, MaintenanceWindow } from "@/domain/maintenance-windows/maintenance-window.type.js"; import { AppError } from "@/utils/AppError.js"; import { isWindowActive } from "@/utils/maintenanceWindow.js"; @@ -42,19 +43,23 @@ export class MaintenanceWindowService implements IMaintenanceWindowService { private monitorsRepository: IMonitorsRepository; private maintenanceWindowsRepository: IMaintenanceWindowsRepository; private jobsRepository: IJobsRepository; + private worker: IWorker; constructor({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, + worker, }: { monitorsRepository: IMonitorsRepository; maintenanceWindowsRepository: IMaintenanceWindowsRepository; jobsRepository: IJobsRepository; + worker: IWorker; }) { this.monitorsRepository = monitorsRepository; this.maintenanceWindowsRepository = maintenanceWindowsRepository; this.jobsRepository = jobsRepository; + this.worker = worker; } get serviceName() { @@ -78,6 +83,9 @@ export class MaintenanceWindowService implements IMaintenanceWindowService { await this.monitorsRepository.updateByIds(toInitializing, teamId, { status: "initializing" }, ["paused"]); // Rearm jobs to run soon (now + jitter to avoid herding) await this.jobsRepository.markMonitorsDue(toInitializing, now.getTime()); + // Wake the (possibly idle, backed-off) loops so the rearmed jobs run promptly instead of waiting up to POLL_MAX_MS + this.worker.wake("check"); + this.worker.wake("geo-check"); } }; diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index bef7a838e..24138695f 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -41,6 +41,17 @@ export class DBQueueWorker implements IWorker { "cleanup-retention": 0, }; + // Per job type poll intervals. These back off on idle and reset on work or wake() + private pollMs: Record = { + check: POLL_MS, + "geo-check": POLL_MS, + evaluate: POLL_MS, + "cleanup-orphaned": POLL_MS, + "cleanup-retention": POLL_MS, + }; + private ticking: Partial> = {}; // Whether a loop is mid tick + private tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake + constructor( private logger: ILogger, private jobsRepository: IJobsRepository, @@ -239,9 +250,10 @@ export class DBQueueWorker implements IWorker { }; private startLoop = (type: JobType) => { - let pollMs = POLL_MS; + this.pollMs[type] = POLL_MS; const tick = async () => { if (this.stopped) return; + this.ticking[type] = true; try { // Claim a batch sized to the free capacity const capacity = CONCURRENCY[type] - this.inFlight[type]; @@ -252,8 +264,8 @@ export class DBQueueWorker implements IWorker { this.inFlight[type]--; // job done, free the slot }); } - // Back off if no jobs founds, reset to base if jobs found - pollMs = capacity > 0 && jobs.length === 0 ? Math.min(pollMs * 2, POLL_MAX_MS) : POLL_MS; + // Back off if no jobs found, reset to base if jobs found + this.pollMs[type] = capacity > 0 && jobs.length === 0 ? Math.min(this.pollMs[type] * 2, POLL_MAX_MS) : POLL_MS; } catch (error: unknown) { this.logger.error({ message: error instanceof Error ? error.message : String(error), @@ -261,11 +273,26 @@ export class DBQueueWorker implements IWorker { method: `loop:${type}`, }); } - if (!this.stopped) this.timers.set(type, setTimeout(tick, pollMs)); // re-arm unless shutting down + // Clearing the flag and re-arming run synchronously (no await between), so wake() can never interleave here. + this.ticking[type] = false; + if (!this.stopped) this.timers.set(type, setTimeout(tick, this.pollMs[type])); // re-arm unless shutting down }; + this.tickFns.set(type, tick); tick(); }; + // Wake a worker loop after adding a job that's due immediately. + // Otherwise the loop could be delayed by POLL_MAX_MS if it is already fully backed off + wake = (type: JobType) => { + this.pollMs[type] = POLL_MS; + if (this.stopped || this.ticking[type]) return; // If it's not running or mid tick, do nothing + const tick = this.tickFns.get(type); + if (!tick) return; // Try to get a handle on the tick fn, if we can't we're done + const pending = this.timers.get(type); + if (pending) clearTimeout(pending); // Clear out scheduled tick if any + this.timers.set(type, setTimeout(tick, 0)); // Set immediate tick + }; + // Seed queue private reconcile = async () => { const now = Date.now(); @@ -315,7 +342,11 @@ export class DBQueueWorker implements IWorker { addJob = async (_monitorId: string, monitor: Monitor) => { const now = Date.now(); await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now, true)); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now, true)); + this.wake("check"); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now, true)); + this.wake("geo-check"); + } }; deleteJob = async (monitor: Monitor) => { await this.jobsRepository.deleteById(monitor.id); @@ -324,12 +355,15 @@ export class DBQueueWorker implements IWorker { await this.jobsRepository.setActiveById(monitor.id, false); }; resumeJob = async (monitor: Monitor) => { - await this.jobsRepository.setActiveById(monitor.id, true); + await this.jobsRepository.setActiveById(monitor.id, true); // a paused job's nextScheduledAt is now in the past, so it's due immediately + this.wake("check"); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) this.wake("geo-check"); }; updateJob = async (monitor: Monitor) => { await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval); if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now(), true)); // a newly enabled geo check should run right away + this.wake("geo-check"); } else { await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); } diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index ccaa79d12..fd4160124 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -1,5 +1,6 @@ import { Monitor } from "@/domain/monitors/monitor.types.js"; import { Check } from "@/domain/checks/check.type.js"; +import { JobType } from "@/domain/jobs/job.type.js"; import { MonitorPayloadMap, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js"; import { MonitorActionDecision } from "@/worker/worker.helper.js"; import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; @@ -70,6 +71,7 @@ export interface IWorker { pauseJob(monitor: Monitor): Promise; resumeJob(monitor: Monitor): Promise; updateJob(monitor: Monitor): Promise; + wake(type: JobType): void; shutdown(): Promise; getMetrics(): Promise; getJobs(pagination: WorkerJobsPagination): Promise; diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 1769ed573..cf15db8c0 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -479,6 +479,80 @@ describe("DBQueueWorker", () => { }); }); + // ── poll backoff + wake() ─────────────────────────────────────────────────── + // + // An idle loop grows its poll interval up to POLL_MAX_MS (5s). wake() fast-paths + // it back so a freshly-enqueued "run now" job doesn't wait out that backoff. + + describe("poll backoff and wake", () => { + const POLL_MAX_MS = 5000; + const claims = (mocks: any, type: string) => mocks.jobsRepository.claimDueBatch.mock.calls.filter((c: any[]) => c[0] === type).length; + + it("backs off the idle poll interval toward POLL_MAX_MS", async () => { + const { mocks } = await start({ queueMode: "worker" }); // claimDueBatch returns [] → always idle + await jest.advanceTimersByTimeAsync(15000); // let the check loop grow to the cap + const before = claims(mocks, "check"); + await jest.advanceTimersByTimeAsync(POLL_MAX_MS * 2); // ~2 polls at a 5s cadence + const delta = claims(mocks, "check") - before; + // At the cap that's ~2 polls; at the un-backed-off 250ms cadence it would be ~40. + expect(delta).toBeGreaterThanOrEqual(1); + expect(delta).toBeLessThanOrEqual(4); + }); + + it("wake() re-fires an idle, backed-off loop immediately", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + await jest.advanceTimersByTimeAsync(15000); // back off the check loop to the cap + const before = claims(mocks, "check"); + + worker.wake("check"); + await jest.advanceTimersByTimeAsync(1); + + expect(claims(mocks, "check")).toBe(before + 1); // fired now, not 5s later + }); + + it("wake() resets the cadence so the loop polls fast again, not at the 5s cap", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + await jest.advanceTimersByTimeAsync(15000); // backed off to the cap + worker.wake("check"); + await jest.advanceTimersByTimeAsync(1); // immediate tick (still idle → next poll at ~POLL_MS*2) + const before = claims(mocks, "check"); + + await jest.advanceTimersByTimeAsync(POLL_MS * 2); // 500ms + + // Polled again well within the old 5s cap, proving the interval was reset. + expect(claims(mocks, "check")).toBe(before + 1); + }); + + it("wake() is a no-op for a loop that isn't running in this process", async () => { + const { worker, mocks } = await start({ queueMode: "primary", queuePrimaryProcesses: false }); // no loops + expect(() => worker.wake("check")).not.toThrow(); + await jest.advanceTimersByTimeAsync(1); + expect(claims(mocks, "check")).toBe(0); + }); + + it("addJob wakes the check loop so a new monitor's first check runs promptly", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + await jest.advanceTimersByTimeAsync(15000); // backed off to the cap + const before = claims(mocks, "check"); + + await worker.addJob("m2", makeMonitor({ id: "m2" })); + await jest.advanceTimersByTimeAsync(1); + + expect(claims(mocks, "check")).toBe(before + 1); + }); + + it("resumeJob wakes the check loop so a resumed monitor runs promptly", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + await jest.advanceTimersByTimeAsync(15000); // backed off to the cap + const before = claims(mocks, "check"); + + await worker.resumeJob(makeMonitor()); + await jest.advanceTimersByTimeAsync(1); + + expect(claims(mocks, "check")).toBe(before + 1); + }); + }); + // ── shutdown ──────────────────────────────────────────────────────────────── describe("shutdown", () => { diff --git a/server/test/unit/services/maintenanceWindowService.test.ts b/server/test/unit/services/maintenanceWindowService.test.ts index 07d19e405..5c7387da8 100644 --- a/server/test/unit/services/maintenanceWindowService.test.ts +++ b/server/test/unit/services/maintenanceWindowService.test.ts @@ -3,6 +3,7 @@ import { MaintenanceWindowService } from "../../../src/domain/maintenance-window import type { IMaintenanceWindowsRepository } from "../../../src/domain/maintenance-windows/maintenance-window.repository.interface.ts"; import type { IMonitorsRepository } from "../../../src/domain/monitors/monitor.repository.interface.ts"; import type { IJobsRepository } from "../../../src/domain/jobs/job.repository.interface.ts"; +import type { IWorker } from "../../../src/worker/worker.interface.ts"; import type { MaintenanceWindow } from "../../../src/domain/maintenance-windows/maintenance-window.type.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── @@ -33,16 +34,23 @@ const createJobsRepo = () => markMonitorsDue: jest.fn().mockResolvedValue(0), }) as unknown as jest.Mocked; +const createWorker = () => + ({ + wake: jest.fn(), + }) as unknown as jest.Mocked; + const createService = (overrides?: { monitorsRepository?: ReturnType; maintenanceWindowsRepository?: ReturnType; jobsRepository?: ReturnType; + worker?: ReturnType; }) => { const monitorsRepository = overrides?.monitorsRepository ?? createMonitorsRepo(); const maintenanceWindowsRepository = overrides?.maintenanceWindowsRepository ?? createMaintenanceWindowsRepo(); const jobsRepository = overrides?.jobsRepository ?? createJobsRepo(); - const service = new MaintenanceWindowService({ monitorsRepository, maintenanceWindowsRepository, jobsRepository }); - return { service, monitorsRepository, maintenanceWindowsRepository, jobsRepository }; + const worker = overrides?.worker ?? createWorker(); + const service = new MaintenanceWindowService({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, worker }); + return { service, monitorsRepository, maintenanceWindowsRepository, jobsRepository, worker }; }; const makeWindow = (overrides?: Partial): MaintenanceWindow => ({ @@ -275,7 +283,7 @@ describe("MaintenanceWindowService", () => { it("flips covered monitors to initializing when deleting an active window", async () => { const maintenanceWindowsRepository = createMaintenanceWindowsRepo(); (maintenanceWindowsRepository.deleteById as jest.Mock).mockResolvedValue(makeActiveWindow({ monitorIds: ["mon-1", "mon-2"] })); - const { service, monitorsRepository, jobsRepository } = createService({ maintenanceWindowsRepository }); + const { service, monitorsRepository, jobsRepository, worker } = createService({ maintenanceWindowsRepository }); await service.deleteMaintenanceWindow({ id: "mw-1", teamId: "team-1" }); @@ -283,6 +291,19 @@ describe("MaintenanceWindowService", () => { expect(monitorsRepository.updateByIds).toHaveBeenCalledWith(["mon-1", "mon-2"], "team-1", { status: "initializing" }, ["paused"]); // Monitors leaving maintenance are re-armed to run soon instead of waiting out the interval expect(jobsRepository.markMonitorsDue).toHaveBeenCalledWith(["mon-1", "mon-2"], expect.any(Number)); + // ...and the idle loops are woken so the re-armed jobs run promptly + expect(worker.wake).toHaveBeenCalledWith("check"); + expect(worker.wake).toHaveBeenCalledWith("geo-check"); + }); + + it("does not wake the loops when no monitor leaves maintenance", async () => { + const maintenanceWindowsRepository = createMaintenanceWindowsRepo(); + (maintenanceWindowsRepository.deleteById as jest.Mock).mockResolvedValue(makeWindow({ active: false, monitorIds: ["mon-1"] })); + const { service, worker } = createService({ maintenanceWindowsRepository }); + + await service.deleteMaintenanceWindow({ id: "mw-1", teamId: "team-1" }); + + expect(worker.wake).not.toHaveBeenCalled(); }); it("excludes the deleted window from the overlap check", async () => {