From fa0c0abcac8767f271605f76d31d3f6071b5b013 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 17 Sep 2026 11:57:19 -0700 Subject: [PATCH] Fix: Use Check IDs instead of timestamp to process checks --- .../checks/check.repository.interface.ts | 3 +- .../domain/checks/check.repository.mongo.ts | 12 +- server/src/domain/checks/check.service.ts | 5 - server/src/domain/jobs/job.model.ts | 12 ++ .../domain/jobs/job.repository.interface.ts | 9 +- .../src/domain/jobs/job.repository.mongo.ts | 46 ++-- server/src/domain/jobs/job.type.ts | 8 + server/src/service/bufferService.ts | 28 ++- server/src/worker/worker.db-queue.ts | 18 +- .../evaluateCursorMultiWorker.test.ts | 200 ++++++++++++++++++ .../test/integration/jobsRepository.test.ts | 126 +++++++++-- .../test/unit/services/bufferService.test.ts | 23 +- .../test/unit/services/dbQueueWorker.test.ts | 57 ++++- 13 files changed, 491 insertions(+), 56 deletions(-) create mode 100644 server/test/integration/evaluateCursorMultiWorker.test.ts diff --git a/server/src/domain/checks/check.repository.interface.ts b/server/src/domain/checks/check.repository.interface.ts index c48c126d1..aea539dca 100644 --- a/server/src/domain/checks/check.repository.interface.ts +++ b/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; findDockerContainerChecks(monitorId: string, containerName: string, dateRange: DateRange): Promise>; findSummaryByTeamId(teamId: string, dateRange: DateRange): Promise; - findUnevaluatedByMonitorId(monitorId: string, since: number): Promise; + findUnevaluatedByMonitorId(monitorId: string, pending: PendingCheck[]): Promise; getDailyStatusBuckets(monitorIds: string[], days: number, timezone: string): Promise; // update diff --git a/server/src/domain/checks/check.repository.mongo.ts b/server/src/domain/checks/check.repository.mongo.ts index 23106c0b7..4e14558f9 100644 --- a/server/src/domain/checks/check.repository.mongo.ts +++ b/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(); diff --git a/server/src/domain/checks/check.service.ts b/server/src/domain/checks/check.service.ts index 5eb071bab..2bc9f2a16 100644 --- a/server/src/domain/checks/check.service.ts +++ b/server/src/domain/checks/check.service.ts @@ -12,7 +12,6 @@ const SERVICE_NAME = "checkService"; export interface ICheckService { createChecks(checks: Check[]): Promise; - toLastEvaluatedAt(check: Check): number; toCheck(statusResponse: MonitorStatusResponse): Check | undefined; toStatusResponse(check: Check): MonitorStatusResponse; 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): Check | undefined => { const { monitorId, teamId, type, status, responseTime, code, message, payload, timings } = statusResponse; diff --git a/server/src/domain/jobs/job.model.ts b/server/src/domain/jobs/job.model.ts index f64ef53b8..0745cc0f4 100644 --- a/server/src/domain/jobs/job.model.ts +++ b/server/src/domain/jobs/job.model.ts @@ -47,6 +47,18 @@ const JobSchema = new Schema( 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, diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index 435e49118..39f797d98 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/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; // ******************** - // 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; + upsertEvaluate(monitorId: string, pending: PendingCheck[], now: number): Promise; + + // Removes processed checks from the evaluate row. + pullEvaluated(id: string, checkIds: string[]): Promise; // ******************** // Job CRUD diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index 809362192..da9f95263 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/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 }, diff --git a/server/src/domain/jobs/job.type.ts b/server/src/domain/jobs/job.type.ts index ae8fbddc3..a57b1a8c1 100644 --- a/server/src/domain/jobs/job.type.ts +++ b/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; diff --git a/server/src/service/bufferService.ts b/server/src/service/bufferService.ts index 672ec174c..eceaca4c7 100755 --- a/server/src/service/bufferService.ts +++ b/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; scheduleNextFlush(): void; flushBuffer(): Promise; flushGeoBuffer(): Promise; @@ -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(); + 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[] = []; + 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", diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index c076151ef..d1cf4b4ea 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/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) => { diff --git a/server/test/integration/evaluateCursorMultiWorker.test.ts b/server/test/integration/evaluateCursorMultiWorker.test.ts new file mode 100644 index 000000000..221c33c78 --- /dev/null +++ b/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().mockResolvedValue([]) }; +const stubDockerLogsService = { createDockerLogs: jest.fn().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 }).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() }, + 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() }, + dispatcher: { dispatch: jest.fn().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); + }); +}); diff --git a/server/test/integration/jobsRepository.test.ts b/server/test/integration/jobsRepository.test.ts index 287ba7d63..467a4d670 100644 --- a/server/test/integration/jobsRepository.test.ts +++ b/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(); 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(); + 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"); diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index d193c7b3d..06522abbc 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/server/test/unit/services/bufferService.test.ts @@ -45,6 +45,8 @@ const makeCheck = (overrides?: Partial): 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 () => { diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index a2d532c16..4ea611a4d 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -56,6 +56,7 @@ const makeJob = (overrides?: Partial): Job => ({ intervalMs: 60000, lockedBy: null, lockedUntil: null, + pendingChecks: [], runCount: 0, failCount: 0, lastFinishedAt: null, @@ -63,12 +64,22 @@ const makeJob = (overrides?: Partial): 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 }) => { const jobsRepository = { claimDueBatch: jest.fn().mockResolvedValue([]), upsertJob: jest.fn().mockResolvedValue(true), upsertCleanupJob: jest.fn().mockResolvedValue(true), upsertEvaluate: jest.fn().mockResolvedValue(true), + pullEvaluated: jest.fn().mockResolvedValue(true), recordSuccess: jest.fn().mockResolvedValue(true), recordFailure: jest.fn().mockResolvedValue(true), recordOneShot: jest.fn().mockResolvedValue(true), @@ -92,7 +103,6 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses }; const checkService = { toStatusResponse: jest.fn().mockReturnValue({ status: "up" }), - toLastEvaluatedAt: jest.fn().mockReturnValue(12345), }; const bufferService = { addToBuffer: jest.fn(), @@ -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().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().mockResolvedValue(null), updateById: jest.fn() }; + 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().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 = {