diff --git a/server/package-lock.json b/server/package-lock.json index be543815d..72a485391 100644 --- a/server/package-lock.json +++ b/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" diff --git a/server/package.json b/server/package.json index c9b85e17c..c4ed47933 100755 --- a/server/package.json +++ b/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", diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index ae0b5df72..51dcd9460 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/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[0][number]; + // For discriminating job types type MonitorJob = Omit & { data: Monitor }; const MONITOR_TEMPLATES = new Set(["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) => { diff --git a/server/src/service/infrastructure/bufferService.ts b/server/src/service/infrastructure/bufferService.ts index 07be446c6..5a1fb3d12 100755 --- a/server/src/service/infrastructure/bufferService.ts +++ b/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; diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index 7a5ee5b88..f1f395e2d 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/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>().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>().mockRejectedValueOnce("string error"); - await jest.advanceTimersByTimeAsync(10); + await jest.advanceTimersByTimeAsync(1000); expect(logger.error).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts index 19b68dee1..0b1ec5bbc 100644 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ b/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" })])); }); });