Merge pull request #3695 from ajhollid/fix/less-simple-queue-tweaks

fix: less simple queue tweaks

server/package-lock.json

@@ -31,7 +31,7 @@
 				"jmespath": "^0.16.0",
 				"jsdom": "^26.1.0",
 				"jsonwebtoken": "9.0.2",
-				"less-simple-scheduler": "2.3.0",
+				"less-simple-scheduler": "2.4.1",
 				"mailersend": "^2.2.0",
 				"mjml": "^5.0.0-alpha.4",
 				"mongoose": "^8.3.3",
@@ -10417,9 +10417,9 @@
 			"license": "MIT"
 		},
 		"node_modules/less-simple-scheduler": {
-			"version": "2.3.0",
-			"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.3.0.tgz",
-			"integrity": "sha512-HL8f8YCcNav03oNTWFQxkWmoqtEzsIXNOgLcVmOEAgFxsALukOk4bDlem6SlZmjVAsip8Y50QUu6ZsblxqQYmA==",
+			"version": "2.4.1",
+			"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.4.1.tgz",
+			"integrity": "sha512-7Bgd5qLahFT3rv2QlY8QHrkkXLfUUgUpdDyabAiiy0WeEn1ay7Sacxcbge9TCZ34ZsySp+h1+NHeoTnWVU0vaQ==",
 			"license": "ISC",
 			"dependencies": {
 				"mongodb": "7.2.0"

server/package.json

@@ -53,7 +53,7 @@
 		"jmespath": "^0.16.0",
 		"jsdom": "^26.1.0",
 		"jsonwebtoken": "9.0.2",
-		"less-simple-scheduler": "2.3.0",
+		"less-simple-scheduler": "2.4.1",
 		"mailersend": "^2.2.0",
 		"mjml": "^5.0.0-alpha.4",
 		"mongoose": "^8.3.3",

server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts

@@ -14,6 +14,9 @@ const SERVICE_NAME = "JobQueue";
 // every lockMs/3, so ~5s with the default lockMs), comfortably inside this TTL.
 const WORKER_TTL_MS = 30_000;
 
+// Derived from the scheduler
+type SchedulerJobInput = Parameters<Scheduler["addJobs"]>[0][number];
+
 // For discriminating job types
 type MonitorJob = Omit<IJob, "data"> & { data: Monitor };
 const MONITOR_TEMPLATES = new Set<string>(["monitor-job", "geo-check-job"]);
@@ -138,7 +141,8 @@ export class LessSimpleQueue implements IJobQueue {
 			url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db",
 		});
 		const scheduler = new Scheduler(store, {
-			concurrency: 50,
+			concurrency: envSettings.queueMode === "primary" ? 0 : 50,
+			processEvery: 100,
 		});
 		const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode);
 		await instance.init(queueMode);
@@ -190,11 +194,9 @@ export class LessSimpleQueue implements IJobQueue {
 			if (!monitors) {
 				return true;
 			}
-			for (const monitor of monitors) {
-				const randomOffset = Math.floor(Math.random() * 100);
-				setTimeout(() => {
-					this.addJob(monitor.id, monitor);
-				}, randomOffset);
+			const jobInputs = monitors.flatMap((monitor) => this.buildJobInputs(monitor));
+			if (jobInputs.length > 0) {
+				await this.scheduler.addJobs(jobInputs);
 			}
 
 			this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, upsert: true });
@@ -211,32 +213,38 @@ export class LessSimpleQueue implements IJobQueue {
 		}
 	};
 
-	addJob = async (monitorId: string, monitor: Monitor) => {
-		this.scheduler.addJob({
-			id: monitorId,
-			template: "monitor-job",
-			repeat: monitor.interval,
-			active: monitor.isActive,
-			data: monitor,
-			upsert: true,
-		});
+	// Builds the scheduler inputs for a monitor: always a monitor-job, plus a
+	// geo-check job when the monitor supports and enables geo checks.
+	private buildJobInputs = (monitor: Monitor): SchedulerJobInput[] => {
+		const inputs: SchedulerJobInput[] = [
+			{
+				id: monitor.id,
+				template: "monitor-job",
+				repeat: monitor.interval,
+				active: monitor.isActive,
+				data: monitor,
+				upsert: true,
+				jitter: true,
+			},
+		];
 
-		// Return early if we don't need geo checks
-		if (!supportsGeoCheck(monitor.type)) {
-			return;
-		}
-
-		// Add geo check job if enabled for HTTP monitors
-		if (monitor.geoCheckEnabled) {
-			this.scheduler.addJob({
-				id: `${monitorId}-geo`,
+		if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) {
+			inputs.push({
+				id: `${monitor.id}-geo`,
 				template: "geo-check-job",
 				repeat: monitor.geoCheckInterval,
 				active: monitor.isActive,
 				data: monitor,
 				upsert: true,
+				jitter: true,
 			});
 		}
+
+		return inputs;
+	};
+
+	addJob = async (_monitorId: string, monitor: Monitor) => {
+		await this.scheduler.addJobs(this.buildJobInputs(monitor));
 	};
 
 	deleteJob = async (monitor: Monitor) => {

server/src/service/infrastructure/bufferService.ts

@@ -26,7 +26,7 @@ export class BufferService implements IBufferService {
 	private geoChecksService: IGeoChecksService;
 
 	constructor(logger: ILogger, checkService: ICheckService, geoChecksService: IGeoChecksService, settingsService: ISettingsService) {
-		this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 10 : 1000 * 60 * 1; // 1 minute
+		this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 1000 : 1000 * 60 * 1; // 1 minute
 		this.logger = logger;
 		this.checksService = checkService;
 		this.geoChecksService = geoChecksService;

server/test/unit/services/bufferService.test.ts

@@ -81,7 +81,7 @@ describe("BufferService", () => {
 			const { logger } = createService("development");
 			expect(logger.info).toHaveBeenCalledWith(
 				expect.objectContaining({
-					message: expect.stringContaining("0.01s"),
+					message: expect.stringContaining("1s"),
 					service: "BufferService",
 					method: "constructor",
 				})
@@ -231,7 +231,7 @@ describe("BufferService", () => {
 			service.addToBuffer(makeCheck());
 			service.addGeoCheckToBuffer(makeGeoCheck());
 
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			expect(checkService.createChecks).toHaveBeenCalled();
 			expect(geoChecksService.createGeoChecks).toHaveBeenCalled();
@@ -242,11 +242,11 @@ describe("BufferService", () => {
 			(checkService.createChecks as jest.Mock).mockResolvedValue([]);
 
 			service.addToBuffer(makeCheck());
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			// Add another check and advance again to confirm rescheduling
 			service.addToBuffer(makeCheck({ id: "c2" }));
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			expect(checkService.createChecks).toHaveBeenCalledTimes(2);
 		});
@@ -256,7 +256,7 @@ describe("BufferService", () => {
 			(checkService.createChecks as jest.Mock).mockRejectedValueOnce(new Error("DB down"));
 			service.addToBuffer(makeCheck());
 
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			expect(logger.error).toHaveBeenCalledWith(
 				expect.objectContaining({
@@ -268,7 +268,7 @@ describe("BufferService", () => {
 			// Should still reschedule — add another check and flush
 			(checkService.createChecks as jest.Mock).mockResolvedValue([]);
 			service.addToBuffer(makeCheck({ id: "c2" }));
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			expect(checkService.createChecks).toHaveBeenCalledTimes(2);
 		});
@@ -278,7 +278,7 @@ describe("BufferService", () => {
 			// Override flushBuffer to throw past its own try/catch
 			service.flushBuffer = jest.fn<() => Promise<void>>().mockRejectedValueOnce(new Error("unexpected"));
 
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			expect(logger.error).toHaveBeenCalledWith(
 				expect.objectContaining({
@@ -292,7 +292,7 @@ describe("BufferService", () => {
 			const { service, logger } = createService();
 			service.flushBuffer = jest.fn<() => Promise<void>>().mockRejectedValueOnce("string error");
 
-			await jest.advanceTimersByTimeAsync(10);
+			await jest.advanceTimersByTimeAsync(1000);
 
 			expect(logger.error).toHaveBeenCalledWith(
 				expect.objectContaining({

server/test/unit/services/lessSimpleQueue.test.ts

@@ -22,6 +22,7 @@ const createScheduler = () => {
 		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([]),
@@ -125,19 +126,20 @@ describe("LessSimpleQueue", () => {
 			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-retention" }));
 		});
 
-		it("schedules existing monitors with staggered offsets", async () => {
-			jest.useFakeTimers();
+		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();
-			// advance past the <100ms seed offsets
-			jest.advanceTimersByTime(100);
 
-			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m1", template: "monitor-job" }));
-			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m2", template: "monitor-job" }));
-			jest.useRealTimers();
+			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 () => {
@@ -302,25 +304,29 @@ describe("LessSimpleQueue", () => {
 		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 }));
+			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.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job", repeat: 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.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo" }));
+			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.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo" }));
+			expect(scheduler.addJobs).toHaveBeenCalledWith(expect.not.arrayContaining([expect.objectContaining({ id: "mon-1-geo" })]));
 		});
 	});