Merge branch 'fix/evluation-order' into fix/excess-polling

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

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
 		}
 	}
 

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