Fix: Use Check IDs instead of timestamp to process checks

server/src/domain/checks/check.repository.interface.ts

@@ -9,6 +9,7 @@ import type {
 	PageSpeedChecksResult,
 	UptimeChecksResult,
 } from "@/domain/checks/check.type.js";
+import { PendingCheck } from "@/domain/jobs/job.type.js";
 import type { MonitorType } from "@/domain/monitors/monitor.type.js";
 import { CheckFilter, DateRange } from "@/types/query.js";
 
@@ -43,7 +44,7 @@ export interface IChecksRepository {
 	): Promise<UptimeChecksResult | HardwareChecksResult | PageSpeedChecksResult | DockerChecksResult>;
 	findDockerContainerChecks(monitorId: string, containerName: string, dateRange: DateRange): Promise<Omit<DockerContainerStats, "restartsInRange">>;
 	findSummaryByTeamId(teamId: string, dateRange: DateRange): Promise<ChecksSummary>;
-	findUnevaluatedByMonitorId(monitorId: string, since: number): Promise<Check[]>;
+	findUnevaluatedByMonitorId(monitorId: string, pending: PendingCheck[]): Promise<Check[]>;
 	getDailyStatusBuckets(monitorIds: string[], days: number, timezone: string): Promise<DailyCheckBucket[]>;
 
 	// update

server/src/domain/checks/check.repository.mongo.ts

@@ -33,6 +33,7 @@ import {
 	getDockerTotalChecks,
 	getDockerUpChecks,
 } from "@/domain/checks/check.docker.aggregation.js";
+import { PendingCheck } from "@/domain/jobs/job.type.js";
 
 const SERVICE_NAME = "ChecksRepository";
 
@@ -360,10 +361,17 @@ class MongoChecksRepository implements IChecksRepository {
 		};
 	};
 
-	findUnevaluatedByMonitorId = async (monitorId: string, since: number) => {
+	findUnevaluatedByMonitorId = async (monitorId: string, pending: PendingCheck[]) => {
+		if (pending.length === 0) return [];
+
+		const ids = pending.map((entry) => new mongoose.Types.ObjectId(entry.checkId));
+
+		const times = pending.map((entry) => entry.createdAt);
+
 		const docs = await CheckModel.find({
 			"metadata.monitorId": new mongoose.Types.ObjectId(monitorId),
-			createdAt: { $gt: new Date(since) },
+			createdAt: { $gte: new Date(Math.min(...times)), $lte: new Date(Math.max(...times)) },
+			_id: { $in: ids },
 		})
 			.sort({ createdAt: 1 })
 			.lean<CheckDocument[]>();

server/src/domain/checks/check.service.ts

@@ -12,7 +12,6 @@ const SERVICE_NAME = "checkService";
 
 export interface ICheckService {
 	createChecks(checks: Check[]): Promise<Check[]>;
-	toLastEvaluatedAt(check: Check): number;
 	toCheck(statusResponse: MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>): Check | undefined;
 	toStatusResponse(check: Check): MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>;
 	getChecksByMonitor(params: {
@@ -55,10 +54,6 @@ export class CheckService implements ICheckService {
 		return this.checksRepository.createChecks(checks);
 	};
 
-	toLastEvaluatedAt = (check: Check) => {
-		return new Date(check.createdAt).getTime();
-	};
-
 	toCheck = (statusResponse: MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>): Check | undefined => {
 		const { monitorId, teamId, type, status, responseTime, code, message, payload, timings } = statusResponse;
 

server/src/domain/jobs/job.model.ts

@@ -47,6 +47,18 @@ const JobSchema = new Schema<JobDocument>(
 			default: null,
 		},
 
+		// Pending checks, inserted but not yet evaluated
+		pendingChecks: {
+			type: [
+				{
+					checkId: { type: String, required: true },
+					createdAt: { type: Number, required: true },
+				},
+			],
+			default: [],
+			_id: false,
+		},
+
 		// Observability
 		runCount: {
 			type: Number,

server/src/domain/jobs/job.repository.interface.ts

@@ -1,4 +1,4 @@
-import { Job, JobSeed, JobType } from "@/domain/jobs/job.type.js";
+import { Job, JobSeed, JobType, PendingCheck } from "@/domain/jobs/job.type.js";
 
 export type JobPageQuery = {
 	page?: number;
@@ -32,9 +32,12 @@ export interface IJobsRepository {
 	recordOneShot(id: string, now: number): Promise<boolean>;
 
 	// ********************
-	// Hand off to evaluator // This is accomplished by creating an "evaluation" type job
+	// Hand off to evaluator // This is accomplished pushing newly inserted Check IDs onto the evaluation row
 	// ********************
-	upsertEvaluate(monitorId: string, now: number): Promise<boolean>;
+	upsertEvaluate(monitorId: string, pending: PendingCheck[], now: number): Promise<boolean>;
+
+	// Removes processed checks from the evaluate row.
+	pullEvaluated(id: string, checkIds: string[]): Promise<boolean>;
 
 	// ********************
 	// Job CRUD

server/src/domain/jobs/job.repository.mongo.ts

@@ -1,6 +1,6 @@
 import { IJobsRepository, JobPageQuery } from "@/domain/jobs/job.repository.interface.js";
 import JobModel, { JobDocument } from "@/domain/jobs/job.model.js";
-import { JobType, Job, JobSeed, BACKOFF_MS, LOCK_MS, PARKED, jobId } from "@/domain/jobs/job.type.js";
+import { JobType, Job, JobSeed, BACKOFF_MS, LOCK_MS, PARKED, jobId, PendingCheck } from "@/domain/jobs/job.type.js";
 
 // Cap for rearm interval, never want to wait more than 15s.
 const REARM_JITTER_MAX_MS = 15_000;
@@ -18,6 +18,7 @@ class MongoJobsRepository implements IJobsRepository {
 			intervalMs: doc.intervalMs,
 			lockedBy: doc.lockedBy,
 			lockedUntil: doc.lockedUntil,
+			pendingChecks: doc.pendingChecks ?? [],
 			runCount: doc.runCount,
 			failCount: doc.failCount,
 			lastFinishedAt: doc.lastFinishedAt,
@@ -146,24 +147,27 @@ class MongoJobsRepository implements IJobsRepository {
 				_id: id,
 				lockedBy: this.workerId, // Only the worker that claimed can record success
 			},
-			{
-				$set: {
-					nextScheduledAt: PARKED,
-					lockedBy: null, // Remove lock
-					lockedUntil: null,
-					lastFinishedAt: now,
-					failCount: 0,
-					lastFailReason: null,
+			[
+				{
+					$set: {
+						// Re-arm if more checks were pushed while this run held the lease; otherwise park
+						nextScheduledAt: {
+							$cond: [{ $gt: [{ $size: { $ifNull: ["$pendingChecks", []] } }, 0] }, now, PARKED],
+						},
+						lockedBy: null, // Remove lock
+						lockedUntil: null,
+						lastFinishedAt: now,
+						failCount: 0,
+						lastFailReason: null,
+						runCount: { $add: ["$runCount", 1] },
+					},
 				},
-				$inc: {
-					runCount: 1, // Increment run count
-				},
-			}
+			]
 		);
 		return res.modifiedCount === 1;
 	};
 
-	upsertEvaluate = async (monitorId: string, now: number) => {
+	upsertEvaluate = async (monitorId: string, pending: PendingCheck[], now: number) => {
 		const filter = { type: "evaluate", refId: monitorId };
 		const update = {
 			$setOnInsert: {
@@ -181,6 +185,9 @@ class MongoJobsRepository implements IJobsRepository {
 			$min: {
 				nextScheduledAt: now, // no op if there is already a pending evaluation that is older
 			},
+			$push: {
+				pendingChecks: { $each: pending }, // accumulate pending checks
+			},
 		};
 		try {
 			const res = await JobModel.updateOne(filter, update, { upsert: true });
@@ -195,6 +202,17 @@ class MongoJobsRepository implements IJobsRepository {
 			throw error;
 		}
 	};
+
+	pullEvaluated = async (id: string, checkIds: string[]) => {
+		if (checkIds.length === 0) return true;
+
+		const res = await JobModel.updateOne(
+			{ _id: id, lockedBy: this.workerId }, // Only the worker that claimed can pull checks
+			{ $pull: { pendingChecks: { checkId: { $in: checkIds } } } }
+		);
+		return res.modifiedCount === 1;
+	};
+
 	upsertJob = async (job: JobSeed) => {
 		const res = await JobModel.updateOne(
 			{ _id: job.id },

server/src/domain/jobs/job.type.ts

@@ -4,6 +4,11 @@ export const LOCK_MS = 60_000;
 export const BACKOFF_MS = 5_000;
 export const PARKED = Number.MAX_SAFE_INTEGER;
 
+export type PendingCheck = {
+	checkId: string;
+	createdAt: number; // epoch ms, copied from Check.createdAt
+};
+
 // Canonical _id scheme: monitor-bound rows are `${type}:${refId}`; global rows (refId null: cleanup, egress) are just the type.
 export const jobId = (type: JobType, refId: string | null): string => (refId === null ? type : `${type}:${refId}`);
 
@@ -21,6 +26,9 @@ export type Job = {
 	lockedBy: string | null; // workerId holding the lease
 	lockedUntil: number | null; // epoch ms; lease expiry => reclaimable after a crash
 
+	// Evaluate these rows only.  These are checks that are inserted but not yet evaluated
+	pendingChecks: PendingCheck[];
+
 	// observability (replaces what getMetrics/getJobs)
 	runCount: number;
 	failCount: number;

server/src/service/bufferService.ts

@@ -7,12 +7,14 @@ import type { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"
 import { ICheckService } from "../domain/checks/check.service.js";
 import { DockerLog } from "@/domain/docker/docker-log.type.js";
 import { IDockerLogsService } from "@/domain/docker/docker-log.service.js";
+import { PendingCheck } from "@/domain/jobs/job.type.js";
 const SERVICE_NAME = "BufferService";
 
 export interface IBufferService {
 	addToBuffer(check: Check): void;
 	addGeoCheckToBuffer(geoCheck: GeoCheck): void;
 	addDockerLogToBuffer(dockerLog: DockerLog): void;
+	ingestChecks(checks: Check[]): Promise<void>;
 	scheduleNextFlush(): void;
 	flushBuffer(): Promise<void>;
 	flushGeoBuffer(): Promise<void>;
@@ -99,6 +101,25 @@ export class BufferService implements IBufferService {
 		}
 	}
 
+	ingestChecks = async (checks: Check[]) => {
+		if (checks.length === 0) return;
+		await this.checksService.createChecks(checks);
+
+		const byMonitor = new Map<string, PendingCheck[]>();
+		for (const check of checks) {
+			const pending = byMonitor.get(check.metadata.monitorId) ?? [];
+			pending.push({ checkId: check.id, createdAt: new Date(check.createdAt).getTime() });
+			byMonitor.set(check.metadata.monitorId, pending);
+		}
+
+		const now = Date.now();
+		const ops: Promise<boolean>[] = [];
+		for (const [monitorId, pendingChecks] of byMonitor) {
+			ops.push(this.jobsRepository.upsertEvaluate(monitorId, pendingChecks, now));
+		}
+		await Promise.all(ops);
+	};
+
 	scheduleNextFlush() {
 		if (this.bufferTimer) {
 			clearTimeout(this.bufferTimer);
@@ -121,6 +142,7 @@ export class BufferService implements IBufferService {
 			}
 		}, this.BUFFER_TIMEOUT);
 	}
+
 	async flushBuffer() {
 		if (this.buffer.length === 0) {
 			return;
@@ -134,11 +156,7 @@ export class BufferService implements IBufferService {
 				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)));
+			await this.ingestChecks(batch);
 		} catch (error: unknown) {
 			this.logger.error({
 				message: error instanceof Error ? error.message : "Unknown error",

server/src/worker/worker.db-queue.ts

@@ -120,19 +120,29 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	// ********************
 
 	private runEvaluate = async (job: Job) => {
-		if (!job.refId) return;
+		if (!job.refId || job.pendingChecks.length === 0) return;
+		const pendingCheckIds = job.pendingChecks.map((entry) => entry.checkId);
+
 		const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId
-		if (!monitor) return;
-		const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, monitor.lastEvaluatedAt);
+		if (!monitor) {
+			await this.jobsRepository.pullEvaluated(job.id, pendingCheckIds);
+			return;
+		}
+		const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, job.pendingChecks);
 
 		let current = monitor;
 		for (const check of checks) {
 			const status = this.checkService.toStatusResponse(check);
 			const evaluation = await this.checkEvaluator.evaluate(status, check, current);
 			await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications
-			await this.monitorsRepository.updateById(job.refId, current.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) });
+			await this.jobsRepository.pullEvaluated(job.id, [check.id]);
 			current = evaluation.statusChange.monitor; // fresh statusWindow/status/counters for the next check
 		}
+
+		// Clean up ids potentially orphaned retention cleanup
+		const found = new Set(checks.map((check) => check.id));
+		const missing = pendingCheckIds.filter((checkId) => !found.has(checkId));
+		if (missing.length > 0) await this.jobsRepository.pullEvaluated(job.id, missing);
 	};
 
 	private runGeoCheck = async (job: Job) => {

server/test/integration/evaluateCursorMultiWorker.test.ts

@@ -0,0 +1,200 @@
+import { describe, expect, it, jest, beforeAll, afterAll, beforeEach, afterEach } from "@jest/globals";
+import mongoose from "mongoose";
+import { MongoMemoryServer } from "mongodb-memory-server";
+import MongoChecksRepository from "../../src/domain/checks/check.repository.mongo.ts";
+import MongoJobsRepository from "../../src/domain/jobs/job.repository.mongo.ts";
+import { CheckModel } from "../../src/domain/checks/check.model.ts";
+import JobModel from "../../src/domain/jobs/job.model.ts";
+import { CheckService } from "../../src/domain/checks/check.service.ts";
+import { BufferService } from "../../src/service/bufferService.ts";
+import { DBQueueWorker } from "../../src/worker/worker.db-queue.ts";
+import { InMemoryMonitorsRepository } from "../helpers/InMemoryMonitorsRepository.ts";
+import { createMockLogger } from "../helpers/createMockLogger.ts";
+import type { Job } from "../../src/domain/jobs/job.type.ts";
+import type { Check } from "../../src/domain/checks/check.type.ts";
+import type { Monitor } from "../../src/domain/monitors/monitor.type.ts";
+import type { MonitorStatusResponse } from "../../src/types/network.ts";
+import type { ILogger } from "../../src/utils/logger.ts";
+
+// ── Real-Mongo harness ─────────────────────────────────────────────────────────
+// The evaluate step hands work from the check job to the evaluator through the database:
+// each worker's BufferService flushes its own in-memory batch on its own timer, then queues
+// one evaluate job per monitor. The evaluator reads checks with createdAt strictly after the
+// monitor's lastEvaluatedAt and advances that cursor to each check's createdAt.
+//
+// With two processing workers the flush timers are not aligned, so a later check produced on
+// worker B can be flushed and evaluated before an earlier check produced on worker A has
+// been flushed. When A's batch finally lands, its check is older than the cursor and the
+// evaluate query never returns it. The check exists in the collection but never reaches the
+// status window, stats, incidents or notifications.
+//
+// This drives the real time-series query, the real jobs collection, the real buffer flush and
+// the real DBQueueWorker evaluate path against an in-process mongod, so the cursor semantics
+// under test are MongoDB's, not a mock's.
+
+let mongod: MongoMemoryServer;
+
+beforeAll(async () => {
+	mongod = await MongoMemoryServer.create();
+	await mongoose.connect(mongod.getUri());
+	await CheckModel.createCollection(); // timeseries collections must exist before insert
+	await JobModel.init();
+}, 120_000);
+
+afterAll(async () => {
+	await mongoose.disconnect();
+	await mongod.stop();
+});
+
+beforeEach(async () => {
+	await CheckModel.deleteMany({});
+	await JobModel.deleteMany({});
+});
+
+// ── Helpers ──────────────────────────────────────────────────────────────────
+
+const MONITOR_ID = new mongoose.Types.ObjectId().toString();
+const TEAM_ID = new mongoose.Types.ObjectId().toString();
+const T0 = Date.parse("2026-01-01T00:00:00.000Z");
+const INTERVAL_MS = 30_000;
+
+const makeMonitor = (): Monitor =>
+	({
+		id: MONITOR_ID,
+		teamId: TEAM_ID,
+		type: "http",
+		interval: INTERVAL_MS,
+		status: "up",
+		isActive: true,
+		statusWindow: [],
+		statusWindowSize: 5,
+		statusWindowThreshold: 60,
+		recentChecks: [],
+		lastEvaluatedAt: 0,
+	}) as unknown as Monitor;
+
+const makeStatusResponse = (): MonitorStatusResponse => ({
+	monitorId: MONITOR_ID,
+	teamId: TEAM_ID,
+	type: "http",
+	status: true,
+	code: 200,
+	message: "OK",
+	responseTime: 100,
+});
+
+// Production stamps createdAt at the moment the probe ran, on the producing worker; pin it so
+// the two workers' checks have a known order regardless of wall clock.
+const producedAt = (check: Check, at: number): Check => {
+	const createdAt = new Date(at).toISOString();
+	return { ...check, createdAt, updatedAt: createdAt };
+};
+
+const stubSettingsService = { getSettings: () => ({ nodeEnv: "production" }) };
+const stubGeoChecksService = { createGeoChecks: jest.fn<any>().mockResolvedValue([]) };
+const stubDockerLogsService = { createDockerLogs: jest.fn<any>().mockResolvedValue(0) };
+
+describe("Evaluate cursor with more than one processing worker", () => {
+	let logger: ILogger;
+	let monitorsRepository: InMemoryMonitorsRepository;
+	let checksRepository: MongoChecksRepository;
+	let checkService: CheckService;
+	let bufferA: BufferService;
+	let bufferB: BufferService;
+	let evaluatedCheckIds: string[];
+	let evaluator: DBQueueWorker;
+	let evaluatorJobsRepository: MongoJobsRepository;
+
+	// One BufferService per worker process, each flushing into the same collection and jobs table.
+	const createWorkerBuffer = (workerId: string) =>
+		new BufferService(
+			logger,
+			checkService,
+			stubGeoChecksService as any,
+			stubDockerLogsService as any,
+			stubSettingsService as any,
+			new MongoJobsRepository(workerId)
+		);
+
+	// Claim every due evaluate job and run it through the worker's real per-job path.
+	const runDueEvaluateJobs = async () => {
+		const jobs = await evaluatorJobsRepository.claimDueBatch("evaluate", 20, Date.now());
+		const runJob = (evaluator as unknown as { runJob: (job: Job) => Promise<void> }).runJob;
+		for (const job of jobs) await runJob(job);
+		return jobs.length;
+	};
+
+	beforeEach(() => {
+		logger = createMockLogger() as unknown as ILogger;
+		monitorsRepository = new InMemoryMonitorsRepository();
+		monitorsRepository.seed(makeMonitor());
+		checksRepository = new MongoChecksRepository(logger);
+		checkService = new CheckService(monitorsRepository, logger, checksRepository);
+		bufferA = createWorkerBuffer("worker-a");
+		bufferB = createWorkerBuffer("worker-b");
+
+		evaluatedCheckIds = [];
+		evaluatorJobsRepository = new MongoJobsRepository("worker-b");
+		evaluator = new DBQueueWorker({
+			logger,
+			isDbConnected: () => true,
+			jobsRepository: evaluatorJobsRepository,
+			monitorsRepository,
+			checksRepository,
+			checkService,
+			bufferService: bufferB,
+			checkProducer: { produce: jest.fn<any>() },
+			checkEvaluator: {
+				evaluate: async (status: MonitorStatusResponse, check: Check, monitor: Monitor) => {
+					evaluatedCheckIds.push(check.id);
+					return {
+						monitor,
+						status,
+						check,
+						statusChange: { monitor, statusChanged: false, prevStatus: monitor.status, code: 200, timestamp: 0 },
+						decision: {},
+					};
+				},
+			} as any,
+			geoCheckPipeline: { run: jest.fn<any>() },
+			dispatcher: { dispatch: jest.fn<any>().mockResolvedValue(undefined) },
+			helper: {} as any,
+			queueWorkersRepository: {} as any,
+			queueMode: "worker",
+			queuePrimaryProcesses: true,
+			workerId: "worker-b",
+		});
+	});
+
+	afterEach(async () => {
+		await bufferA.shutdown();
+		await bufferB.shutdown();
+	});
+
+	it("evaluates every persisted check even when a later worker's batch flushes first", async () => {
+		// t=0: worker A holds the check job, probes, and buffers the result. Its flush timer has not fired.
+		const c1 = producedAt(checkService.toCheck(makeStatusResponse())!, T0);
+		bufferA.addToBuffer(c1);
+
+		// t=30s: worker B claims the next run of the same job, probes, and buffers the result.
+		const c2 = producedAt(checkService.toCheck(makeStatusResponse())!, T0 + INTERVAL_MS);
+		bufferB.addToBuffer(c2);
+
+		// t=45s: worker B's flush timer fires first. c2 is inserted and an evaluate job is queued.
+		await bufferB.flushBuffer();
+		expect(await runDueEvaluateJobs()).toBe(1);
+		expect(evaluatedCheckIds).toEqual([c2.id]);
+
+		// t=60s: worker A's flush timer fires. c1 is inserted and the evaluate job is pulled due again.
+		await bufferA.flushBuffer();
+		expect(await runDueEvaluateJobs()).toBe(1);
+
+		// Both checks are durably stored and visible in history.
+		const stored = await CheckModel.find({ "metadata.monitorId": new mongoose.Types.ObjectId(MONITOR_ID) }).lean();
+		expect(stored.map((doc) => String(doc._id)).sort()).toEqual([c1.id, c2.id].sort());
+
+		// Every stored check must reach the evaluator exactly once.
+		expect(evaluatedCheckIds).toHaveLength(2);
+		expect(evaluatedCheckIds).toContain(c1.id);
+	});
+});

server/test/integration/jobsRepository.test.ts

@@ -3,7 +3,7 @@ import mongoose from "mongoose";
 import { MongoMemoryServer } from "mongodb-memory-server";
 import MongoJobsRepository from "../../src/domain/jobs/job.repository.mongo.ts";
 import JobModel, { type JobDocument } from "../../src/domain/jobs/job.model.ts";
-import { BACKOFF_MS, LOCK_MS, PARKED, type Job, type JobType } from "../../src/domain/jobs/job.type.ts";
+import { BACKOFF_MS, LOCK_MS, PARKED, type Job, type JobType, type PendingCheck } from "../../src/domain/jobs/job.type.ts";
 
 // ── Real-Mongo harness ─────────────────────────────────────────────────────────
 // The headline guarantee of the queue — "exactly one claimer wins" — IS MongoDB's
@@ -58,6 +58,9 @@ const ownedBy = (r: MongoJobsRepository) => (r as unknown as { workerId: string
 
 const WORKER_ID = "worker-under-test";
 
+// Pending entries as ingest pushes them: one per inserted check, stamped with the check's createdAt.
+const pending = (...checkIds: string[]): PendingCheck[] => checkIds.map((checkId, i) => ({ checkId, createdAt: NOW + i }));
+
 describe("MongoJobsRepository", () => {
 	let repo: MongoJobsRepository;
 
@@ -239,6 +242,23 @@ describe("MongoJobsRepository", () => {
 			expect(row?.failCount).toBe(2);
 			expect(row?.lastFailReason).toBe("timeout");
 		});
+
+		it("re-arms instead of parking when checks were pushed while the run held the lease", async () => {
+			await seedJob({
+				_id: "evaluate:mon-1",
+				type: "evaluate",
+				intervalMs: null,
+				lockedBy: ownedBy(repo),
+				lockedUntil: NOW + LOCK_MS,
+				pendingChecks: pending("c-late"), // arrived mid-run, not yet pulled
+			});
+
+			expect(await repo.recordOneShot("evaluate:mon-1", NOW + 1_000)).toBe(true);
+			const row = await readRow("evaluate:mon-1");
+			expect(row?.nextScheduledAt).toBe(NOW + 1_000); // due now, not PARKED
+			expect(row?.lockedBy).toBeNull();
+			expect(row?.runCount).toBe(1);
+		});
 	});
 
 	// ── renewLocks (lease renewal) ───────────────────────────────────────────────
@@ -325,10 +345,10 @@ describe("MongoJobsRepository", () => {
 		});
 	});
 
-	// ── upsertEvaluate (handoff + $min coalescing) ───────────────────────────────
+	// ── upsertEvaluate (handoff: $push ids + $min due time) ──────────────────────
 	describe("upsertEvaluate", () => {
-		it("creates a single evaluate row keyed by monitor", async () => {
-			await repo.upsertEvaluate("mon-1", NOW);
+		it("creates a single evaluate row keyed by monitor carrying the pushed ids", async () => {
+			await repo.upsertEvaluate("mon-1", pending("c1", "c2"), NOW);
 
 			const row = await readRow("evaluate:mon-1");
 			expect(row?.type).toBe("evaluate");
@@ -336,11 +356,12 @@ describe("MongoJobsRepository", () => {
 			expect(row?.nextScheduledAt).toBe(NOW);
 			expect(row?.intervalMs).toBeNull();
 			expect(row?.isActive).toBe(true);
+			expect(row?.pendingChecks).toEqual(pending("c1", "c2"));
 		});
 
 		it("coalesces rapid checks: a later call never pushes runAt out ($min)", async () => {
-			await repo.upsertEvaluate("mon-1", NOW);
-			await repo.upsertEvaluate("mon-1", NOW + 9_000);
+			await repo.upsertEvaluate("mon-1", pending("c1"), NOW);
+			await repo.upsertEvaluate("mon-1", pending("c2"), NOW + 9_000);
 
 			const rows = await JobModel.find({ type: "evaluate", refId: "mon-1" }).lean<JobDocument[]>();
 			expect(rows).toHaveLength(1);
@@ -348,19 +369,96 @@ describe("MongoJobsRepository", () => {
 		});
 
 		it("lowers runAt when an earlier evaluation arrives ($min)", async () => {
-			await repo.upsertEvaluate("mon-1", NOW + 9_000);
-			await repo.upsertEvaluate("mon-1", NOW);
+			await repo.upsertEvaluate("mon-1", pending("c1"), NOW + 9_000);
+			await repo.upsertEvaluate("mon-1", pending("c2"), NOW);
 
 			const row = await readRow("evaluate:mon-1");
 			expect(row?.nextScheduledAt).toBe(NOW);
 		});
 
-		it("concurrent upserts for one monitor produce exactly one row", async () => {
-			const workers = Array.from({ length: 20 }, (_, i) => new MongoJobsRepository(`worker-${i}`));
-			await Promise.all(workers.map((w) => w.upsertEvaluate("mon-1", NOW)));
+		it("accumulates ids across flushes so a late batch is never lost", async () => {
+			await repo.upsertEvaluate("mon-1", pending("c2"), NOW); // later check, flushed first
+			await repo.upsertEvaluate("mon-1", pending("c1"), NOW + 15_000); // earlier check, flushed later
 
-			const count = await JobModel.countDocuments({ type: "evaluate", refId: "mon-1" });
-			expect(count).toBe(1);
+			const row = await readRow("evaluate:mon-1");
+			expect(row?.pendingChecks.map((entry) => entry.checkId)).toEqual(["c2", "c1"]);
+		});
+
+		it("keeps pushing onto a row that is currently leased by a worker", async () => {
+			await seedJob({
+				_id: "evaluate:mon-1",
+				type: "evaluate",
+				intervalMs: null,
+				lockedBy: "other-worker",
+				lockedUntil: NOW + LOCK_MS,
+				pendingChecks: pending("c1"),
+			});
+
+			await repo.upsertEvaluate("mon-1", pending("c2"), NOW);
+
+			const row = await readRow("evaluate:mon-1");
+			expect(row?.lockedBy).toBe("other-worker"); // lease untouched
+			expect(row?.pendingChecks.map((entry) => entry.checkId)).toEqual(["c1", "c2"]);
+		});
+
+		it("concurrent upserts for one monitor produce exactly one row holding every id", async () => {
+			const workers = Array.from({ length: 20 }, (_, i) => new MongoJobsRepository(`worker-${i}`));
+			await Promise.all(workers.map((w, i) => w.upsertEvaluate("mon-1", pending(`c${i}`), NOW)));
+
+			const rows = await JobModel.find({ type: "evaluate", refId: "mon-1" }).lean<JobDocument[]>();
+			expect(rows).toHaveLength(1);
+			expect(rows[0].pendingChecks).toHaveLength(20);
+		});
+	});
+
+	// ── pullEvaluated (evaluator removes applied ids) ───────────────────────────
+	describe("pullEvaluated", () => {
+		it("removes only the listed ids and leaves the rest", async () => {
+			await seedJob({
+				_id: "evaluate:mon-1",
+				type: "evaluate",
+				intervalMs: null,
+				lockedBy: ownedBy(repo),
+				lockedUntil: NOW + LOCK_MS,
+				pendingChecks: pending("c1", "c2", "c3"),
+			});
+
+			expect(await repo.pullEvaluated("evaluate:mon-1", ["c1", "c3"])).toBe(true);
+
+			const row = await readRow("evaluate:mon-1");
+			expect(row?.pendingChecks.map((entry) => entry.checkId)).toEqual(["c2"]);
+		});
+
+		it("is fenced on the lease: another worker's row is untouched", async () => {
+			await seedJob({
+				_id: "evaluate:mon-1",
+				type: "evaluate",
+				intervalMs: null,
+				lockedBy: "other-worker",
+				lockedUntil: NOW + LOCK_MS,
+				pendingChecks: pending("c1"),
+			});
+
+			expect(await repo.pullEvaluated("evaluate:mon-1", ["c1"])).toBe(false);
+
+			const row = await readRow("evaluate:mon-1");
+			expect(row?.pendingChecks.map((entry) => entry.checkId)).toEqual(["c1"]);
+		});
+
+		it("is a no-op for an empty id list", async () => {
+			await seedJob({
+				_id: "evaluate:mon-1",
+				type: "evaluate",
+				intervalMs: null,
+				lockedBy: ownedBy(repo),
+				lockedUntil: NOW + LOCK_MS,
+				pendingChecks: pending("c1"),
+			});
+
+			expect(await repo.pullEvaluated("evaluate:mon-1", [])).toBe(true);
+
+			const row = await readRow("evaluate:mon-1");
+			expect(row?.pendingChecks).toHaveLength(1);
 		});
 	});
 
@@ -487,7 +585,7 @@ describe("MongoJobsRepository", () => {
 		it("deleteById drops every row for the monitor", async () => {
 			await seedJob({ _id: "check:mon-1", type: "check" });
 			await seedJob({ _id: "geo:mon-1", type: "geo-check" });
-			await repo.upsertEvaluate("mon-1", NOW);
+			await repo.upsertEvaluate("mon-1", pending("c1"), NOW);
 
 			await repo.deleteById("mon-1");
 

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

@@ -45,6 +45,8 @@ const makeCheck = (overrides?: Partial<Check>): Check =>
 		statusCode: 200,
 		responseTime: 100,
 		message: "OK",
+		createdAt: "2026-01-01T00:00:00.000Z",
+		updatedAt: "2026-01-01T00:00:00.000Z",
 		...overrides,
 	}) as Check;
 
@@ -392,9 +394,26 @@ describe("BufferService", () => {
 
 			await service.flushBuffer();
 
+			const createdAt = Date.parse("2026-01-01T00:00:00.000Z");
 			expect(jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2);
-			expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-1", expect.any(Number));
-			expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-2", expect.any(Number));
+			expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith(
+				"mon-1",
+				[
+					{ checkId: "c1", createdAt },
+					{ checkId: "c2", createdAt },
+				],
+				expect.any(Number)
+			);
+			expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-2", [{ checkId: "c3", createdAt }], expect.any(Number));
+		});
+
+		it("ingestChecks with an empty batch neither writes nor arms", async () => {
+			const { service, checkService, jobsRepository } = createService();
+
+			await service.ingestChecks([]);
+
+			expect(checkService.createChecks).not.toHaveBeenCalled();
+			expect(jobsRepository.upsertEvaluate).not.toHaveBeenCalled();
 		});
 
 		it("does not arm the evaluate stage when the check write fails", async () => {

server/test/unit/services/dbQueueWorker.test.ts

@@ -56,6 +56,7 @@ const makeJob = (overrides?: Partial<Job>): Job => ({
 	intervalMs: 60000,
 	lockedBy: null,
 	lockedUntil: null,
+	pendingChecks: [],
 	runCount: 0,
 	failCount: 0,
 	lastFinishedAt: null,
@@ -63,12 +64,22 @@ const makeJob = (overrides?: Partial<Job>): Job => ({
 	...overrides,
 });
 
+// An evaluate row as ingest leaves it: one pending entry per stored check
+const makeEvaluateJob = (...checkIds: string[]): Job =>
+	makeJob({
+		id: "evaluate:m1",
+		type: "evaluate",
+		intervalMs: null,
+		pendingChecks: checkIds.map((checkId, i) => ({ checkId, createdAt: 1000 + i })),
+	});
+
 const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses?: boolean; mocks?: Record<string, any> }) => {
 	const jobsRepository = {
 		claimDueBatch: jest.fn<any>().mockResolvedValue([]),
 		upsertJob: jest.fn<any>().mockResolvedValue(true),
 		upsertCleanupJob: jest.fn<any>().mockResolvedValue(true),
 		upsertEvaluate: jest.fn<any>().mockResolvedValue(true),
+		pullEvaluated: jest.fn<any>().mockResolvedValue(true),
 		recordSuccess: jest.fn<any>().mockResolvedValue(true),
 		recordFailure: jest.fn<any>().mockResolvedValue(true),
 		recordOneShot: jest.fn<any>().mockResolvedValue(true),
@@ -92,7 +103,6 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 	};
 	const checkService = {
 		toStatusResponse: jest.fn<any>().mockReturnValue({ status: "up" }),
-		toLastEvaluatedAt: jest.fn<any>().mockReturnValue(12345),
 	};
 	const bufferService = {
 		addToBuffer: jest.fn<any>(),
@@ -369,27 +379,62 @@ describe("DBQueueWorker", () => {
 		});
 
 		it("parks a one-shot job (intervalMs null) via recordOneShot instead of rescheduling", async () => {
-			const job = makeJob({ id: "evaluate:m1", type: "evaluate", intervalMs: null });
+			const job = makeEvaluateJob("c1");
 			const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) } } });
 
 			expect(mocks.jobsRepository.recordOneShot).toHaveBeenCalledWith(job.id, expect.any(Number));
 			expect(mocks.jobsRepository.recordSuccess).not.toHaveBeenCalled();
 		});
 
-		it("evaluate job dispatches each unevaluated check and advances lastEvaluatedAt", async () => {
-			const job = makeJob({ id: "evaluate:m1", type: "evaluate", intervalMs: null });
+		it("evaluate job loads the row's pending checks, dispatches each, and pulls its id after dispatch", async () => {
+			const job = makeEvaluateJob("c1");
 			const checksRepository = { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([{ id: "c1" }]) };
 			const { mocks } = await start({
 				mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checksRepository },
 			});
 
+			expect(checksRepository.findUnevaluatedByMonitorId).toHaveBeenCalledWith("m1", job.pendingChecks);
 			expect(mocks.checkEvaluator.evaluate).toHaveBeenCalled();
 			expect(mocks.dispatcher.dispatch).toHaveBeenCalled();
-			expect(mocks.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { lastEvaluatedAt: 12345 });
+			expect(mocks.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1"]);
+			// The monitor cursor is gone: nothing writes lastEvaluatedAt
+			expect(mocks.monitorsRepository.updateById).not.toHaveBeenCalled();
+		});
+
+		it("evaluate job with no pending checks does nothing", async () => {
+			const job = makeEvaluateJob();
+			const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) } } });
+
+			expect(mocks.monitorsRepository.findByIdLean).not.toHaveBeenCalled();
+			expect(mocks.checksRepository.findUnevaluatedByMonitorId).not.toHaveBeenCalled();
+			expect(mocks.jobsRepository.recordOneShot).toHaveBeenCalledWith(job.id, expect.any(Number));
+		});
+
+		it("evaluate job for a deleted monitor drops every pending id", async () => {
+			const job = makeEvaluateJob("c1", "c2");
+			const monitorsRepository = { findByIdLean: jest.fn<any>().mockResolvedValue(null), updateById: jest.fn<any>() };
+			const { mocks } = await start({
+				mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, monitorsRepository },
+			});
+
+			expect(mocks.checkEvaluator.evaluate).not.toHaveBeenCalled();
+			expect(mocks.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1", "c2"]);
+		});
+
+		it("evaluate job pulls ids whose check no longer exists so they do not sit on the row forever", async () => {
+			const job = makeEvaluateJob("c1", "c-gone");
+			const checksRepository = { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([{ id: "c1" }]) }; // c-gone was cleaned up
+			const { mocks } = await start({
+				mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checksRepository },
+			});
+
+			expect(mocks.checkEvaluator.evaluate).toHaveBeenCalledTimes(1);
+			expect(mocks.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1"]);
+			expect(mocks.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c-gone"]);
 		});
 
 		it("evaluate job reads the monitor once and threads the post-write monitor across the backlog", async () => {
-			const job = makeJob({ id: "evaluate:m1", type: "evaluate", intervalMs: null });
+			const job = makeEvaluateJob("c1", "c2");
 			const initialMonitor = makeMonitor();
 			const postWriteMonitor = makeMonitor({ status: "down" }); // what updateStatusWindowAndChecks would return after check c1
 			const monitorsRepository = {