diff --git a/server/src/config/services.ts b/server/src/config/services.ts index a60505c25..ebc17eaa0 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -227,7 +227,7 @@ export const initializeServices = async ({ monitorsRepository, }); - const bufferService = new BufferService(logger, checkService, geoChecksService, settingsService); + const bufferService = new BufferService(logger, checkService, geoChecksService, settingsService, jobsRepository); const statusService = new StatusService(logger, bufferService, monitorsRepository, monitorStatsRepository, checksRepository); diff --git a/server/src/service/bufferService.ts b/server/src/service/bufferService.ts index 38dfb3545..989ebf936 100755 --- a/server/src/service/bufferService.ts +++ b/server/src/service/bufferService.ts @@ -3,6 +3,7 @@ import type { GeoCheck } from "@/domain/geo-checks/geo-check.type.js"; import type { IGeoChecksService } from "../domain/geo-checks/geo-check.service.js"; import type { ILogger } from "@/utils/logger.js"; import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; +import type { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import { ICheckService } from "../domain/checks/check.service.js"; const SERVICE_NAME = "BufferService"; @@ -24,12 +25,20 @@ export class BufferService implements IBufferService { private bufferTimer: NodeJS.Timeout | null = null; private checksService: ICheckService; private geoChecksService: IGeoChecksService; + private jobsRepository: IJobsRepository; - constructor(logger: ILogger, checkService: ICheckService, geoChecksService: IGeoChecksService, settingsService: ISettingsService) { + constructor( + logger: ILogger, + checkService: ICheckService, + geoChecksService: IGeoChecksService, + settingsService: ISettingsService, + jobsRepository: IJobsRepository + ) { this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 1000 : 1000 * 60 * 1; // 1 minute this.logger = logger; this.checksService = checkService; this.geoChecksService = geoChecksService; + this.jobsRepository = jobsRepository; this.SERVICE_NAME = SERVICE_NAME; this.buffer = []; this.geoBuffer = []; @@ -93,16 +102,23 @@ export class BufferService implements IBufferService { }, this.BUFFER_TIMEOUT); } async flushBuffer() { + if (this.buffer.length === 0) { + return; + } + // Take the batch first so a write that drains it can't be appended to mid-flush + const batch = this.buffer; + this.buffer = []; try { - if (this.buffer.length > 0) { - this.logger.debug({ - message: `Flushing ${this.buffer.length} checks to database`, - service: this.SERVICE_NAME, - method: "flushBuffer", - }); - await this.checksService.createChecks(this.buffer); - this.buffer = []; - } + this.logger.debug({ + message: `Flushing ${batch.length} checks to database`, + service: this.SERVICE_NAME, + method: "flushBuffer", + }); + await this.checksService.createChecks(batch); + // Need to evaluate checks when they are flushed + const monitorIds = [...new Set(batch.map((check) => check.metadata.monitorId))]; + const now = Date.now(); + await Promise.all(monitorIds.map((monitorId) => this.jobsRepository.upsertEvaluate(monitorId, now))); } catch (error: unknown) { this.logger.error({ message: error instanceof Error ? error.message : "Unknown error", @@ -110,8 +126,7 @@ export class BufferService implements IBufferService { method: "flushBuffer", stack: error instanceof Error ? error.stack : undefined, }); - // Clear buffer even on error to prevent infinite retry loops - this.buffer = []; + // Batch is already cleared from the buffer; dropping it prevents infinite retry loops } } diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 2b42b8ca2..9088dc3b3 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -98,18 +98,18 @@ export class DBQueueWorker implements IWorker { return instance; } - // Helpers, builds job seeds - private toCheckJob = (monitor: Monitor, now: number): JobSeed => ({ + // Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd. + private toCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ id: jobId("check", monitor.id), type: "check", refId: monitor.id, isActive: monitor.isActive, - nextScheduledAt: now + Math.floor(Math.random() * monitor.interval), // Avoid herd + nextScheduledAt: immediate ? now : now + Math.floor(Math.random() * monitor.interval), // Avoid herd intervalMs: monitor.interval, }); - private toGeoCheckJob = (monitor: Monitor, now: number): JobSeed => ({ - ...this.toCheckJob(monitor, now), + private toGeoCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ + ...this.toCheckJob(monitor, now, immediate), id: jobId("geo-check", monitor.id), type: "geo-check", intervalMs: monitor.geoCheckInterval!, @@ -153,10 +153,7 @@ export class DBQueueWorker implements IWorker { if (!job.refId) return; const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId if (!monitor) return; - const produced = await this.checkProducer.produce(monitor); - if (produced) { - await this.jobsRepository.upsertEvaluate(monitor.id, Date.now()); - } + await this.checkProducer.produce(monitor); }; // ******************** @@ -182,7 +179,6 @@ export class DBQueueWorker implements IWorker { }; // Extend the lock while a job runs so a slow-but-alive job is never reclaimed. - // Returns false once we no longer hold the lock, so the renewal timer can stop. private renewLock = async (job: Job): Promise => { try { const held = await this.jobsRepository.renewLocks([job.id], Date.now()); @@ -192,7 +188,6 @@ export class DBQueueWorker implements IWorker { } return true; } catch (error: unknown) { - // Transient failure, keep renewing; a missed renewal is absorbed by the 1/3 margin this.logger.warn({ message: error instanceof Error ? error.message : String(error), service: SERVICE_NAME, method: "renewLock" }); return true; } @@ -316,8 +311,8 @@ export class DBQueueWorker implements IWorker { addJob = async (_monitorId: string, monitor: Monitor) => { const now = Date.now(); - await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now)); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, 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)); }; deleteJob = async (monitor: Monitor) => { await this.jobsRepository.deleteById(monitor.id); @@ -331,7 +326,7 @@ export class DBQueueWorker implements IWorker { 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())); + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now(), true)); // a newly enabled geo check should run right away } else { await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); } diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index 5a6418c70..7c344c086 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/server/test/unit/services/bufferService.test.ts @@ -4,6 +4,7 @@ import { createMockLogger } from "../../helpers/createMockLogger.ts"; import type { ICheckService } from "../../../src/domain/checks/check.service.ts"; import type { IGeoChecksService } from "../../../src/domain/geo-checks/geo-check.service.ts"; import type { ISettingsService } from "../../../src/domain/app-settings/app-settings.service.ts"; +import type { IJobsRepository } from "../../../src/domain/jobs/job.repository.interface.ts"; import type { Check } from "../../../src/domain/checks/check.type.ts"; import type { GeoCheck } from "../../../src/domain/geo-checks/geo-check.type.ts"; @@ -24,6 +25,11 @@ const createMockSettingsService = (nodeEnv: string = "development") => getSettings: jest.fn().mockReturnValue({ nodeEnv }), }) as unknown as jest.Mocked; +const createMockJobsRepository = () => + ({ + upsertEvaluate: jest.fn().mockResolvedValue(true), + }) as unknown as jest.Mocked; + const makeCheck = (overrides?: Partial): Check => ({ id: "check-1", @@ -47,8 +53,9 @@ const createService = (nodeEnv: string = "development") => { const checkService = createMockCheckService(); const geoChecksService = createMockGeoChecksService(); const settingsService = createMockSettingsService(nodeEnv); - const service = new BufferService(logger as any, checkService, geoChecksService, settingsService); - return { service, logger, checkService, geoChecksService }; + const jobsRepository = createMockJobsRepository(); + const service = new BufferService(logger as any, checkService, geoChecksService, settingsService, jobsRepository); + return { service, logger, checkService, geoChecksService, jobsRepository }; }; // ── Tests ──────────────────────────────────────────────────────────────────── @@ -344,6 +351,29 @@ describe("BufferService", () => { ); }); + it("arms the evaluate stage once per distinct monitor after the checks are stored", async () => { + const { service, jobsRepository } = createService(); + service.addToBuffer(makeCheck({ id: "c1", metadata: { monitorId: "mon-1", teamId: "team-1", type: "http" } })); + service.addToBuffer(makeCheck({ id: "c2", metadata: { monitorId: "mon-1", teamId: "team-1", type: "http" } })); + service.addToBuffer(makeCheck({ id: "c3", metadata: { monitorId: "mon-2", teamId: "team-1", type: "http" } })); + + await service.flushBuffer(); + + expect(jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2); + expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-1", expect.any(Number)); + expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-2", expect.any(Number)); + }); + + it("does not arm the evaluate stage when the check write fails", async () => { + const { service, checkService, jobsRepository } = createService(); + (checkService.createChecks as jest.Mock).mockRejectedValue(new Error("DB write failed")); + service.addToBuffer(makeCheck()); + + await service.flushBuffer(); + + expect(jobsRepository.upsertEvaluate).not.toHaveBeenCalled(); + }); + it("clears buffer even on error to prevent infinite retries", async () => { const { service, checkService, logger } = createService(); (checkService.createChecks as jest.Mock).mockRejectedValue(new Error("DB write failed")); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index e24c5790d..e60f82a95 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -270,7 +270,8 @@ describe("DBQueueWorker", () => { expect(mocks.monitorsRepository.findByIds).toHaveBeenCalledWith(["m1"]); expect(mocks.checkProducer.produce).toHaveBeenCalled(); - expect(mocks.jobsRepository.upsertEvaluate).toHaveBeenCalledWith("m1", expect.any(Number)); + // The check stage only produces; the buffer arms evaluate once the check is durably stored + expect(mocks.jobsRepository.upsertEvaluate).not.toHaveBeenCalled(); expect(mocks.jobsRepository.recordSuccess).toHaveBeenCalledWith(job.id, job.nextScheduledAt, job.intervalMs, expect.any(Number)); expect(mocks.jobsRepository.recordFailure).not.toHaveBeenCalled(); }); @@ -394,6 +395,17 @@ describe("DBQueueWorker", () => { expect(types).toContain("geo-check"); }); + it("addJob schedules new jobs to run immediately (no startup jitter)", async () => { + const { worker, mocks } = createWorker(); + active = worker; + const now = Date.now(); + await worker.addJob("m1", makeMonitor({ geoCheckEnabled: true })); + + for (const [seed] of mocks.jobsRepository.upsertJob.mock.calls as any[][]) { + expect(seed.nextScheduledAt).toBe(now); + } + }); + it("addJob seeds only a check row for non-geo monitors", async () => { const { worker, mocks } = createWorker(); active = worker;