wake worker on job change

server/src/config/services.ts

@@ -334,6 +334,7 @@ export const initializeServices = async ({
 		monitorsRepository,
 		maintenanceWindowsRepository,
 		jobsRepository,
+		worker,
 	});
 	const monitorService = new MonitorService({
 		worker,

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");
 		}
 	};
 

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<JobType, number> = {
+		check: POLL_MS,
+		"geo-check": POLL_MS,
+		evaluate: POLL_MS,
+		"cleanup-orphaned": POLL_MS,
+		"cleanup-retention": POLL_MS,
+	};
+	private ticking: Partial<Record<JobType, boolean>> = {}; // Whether a loop is mid tick
+	private tickFns = new Map<JobType, () => 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");
 		}

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<void>;
 	resumeJob(monitor: Monitor): Promise<void>;
 	updateJob(monitor: Monitor): Promise<void>;
+	wake(type: JobType): void;
 	shutdown(): Promise<void>;
 	getMetrics(): Promise<WorkerMetrics>;
 	getJobs(pagination: WorkerJobsPagination): Promise<WorkerJobsPage>;

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", () => {

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<IJobsRepository>;
 
+const createWorker = () =>
+	({
+		wake: jest.fn(),
+	}) as unknown as jest.Mocked<IWorker>;
+
 const createService = (overrides?: {
 	monitorsRepository?: ReturnType<typeof createMonitorsRepo>;
 	maintenanceWindowsRepository?: ReturnType<typeof createMaintenanceWindowsRepo>;
 	jobsRepository?: ReturnType<typeof createJobsRepo>;
+	worker?: ReturnType<typeof createWorker>;
 }) => {
 	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>): 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 () => {