From f2990985486dc5071e8e5778b928a4c7779eb4ae Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 17 Sep 2026 12:30:56 -0700 Subject: [PATCH] bug fixes --- .../domain/checks/check.repository.mongo.ts | 10 +++++-- .../domain/jobs/job.repository.interface.ts | 2 +- .../src/domain/jobs/job.repository.mongo.ts | 7 ++--- server/src/service/bufferService.ts | 26 +++++++++++++++---- server/src/worker/worker.db-queue.ts | 18 ++++++++++--- .../evaluateCursorMultiWorker.test.ts | 24 ++++++++--------- .../test/integration/jobsRepository.test.ts | 23 ++++++++++++++++ .../test/unit/services/bufferService.test.ts | 18 +++++++++++++ .../test/unit/services/dbQueueWorker.test.ts | 16 ++++++++++++ 9 files changed, 118 insertions(+), 26 deletions(-) diff --git a/server/src/domain/checks/check.repository.mongo.ts b/server/src/domain/checks/check.repository.mongo.ts index 4e14558f9..82e89d0d9 100644 --- a/server/src/domain/checks/check.repository.mongo.ts +++ b/server/src/domain/checks/check.repository.mongo.ts @@ -366,11 +366,17 @@ class MongoChecksRepository implements IChecksRepository { const ids = pending.map((entry) => new mongoose.Types.ObjectId(entry.checkId)); - const times = pending.map((entry) => entry.createdAt); + // Bound the query to the pending time range so the monitorId + createdAt index is used + let from = Number.POSITIVE_INFINITY; + let to = Number.NEGATIVE_INFINITY; + for (const entry of pending) { + if (entry.createdAt < from) from = entry.createdAt; + if (entry.createdAt > to) to = entry.createdAt; + } const docs = await CheckModel.find({ "metadata.monitorId": new mongoose.Types.ObjectId(monitorId), - createdAt: { $gte: new Date(Math.min(...times)), $lte: new Date(Math.max(...times)) }, + createdAt: { $gte: new Date(from), $lte: new Date(to) }, _id: { $in: ids }, }) .sort({ createdAt: 1 }) diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index 39f797d98..08a16a4f5 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -36,7 +36,7 @@ export interface IJobsRepository { // ******************** upsertEvaluate(monitorId: string, pending: PendingCheck[], now: number): Promise; - // Removes processed checks from the evaluate row. + // Removes processed checks from the evaluate row. Returns false when this worker no longer holds the lease. pullEvaluated(id: string, checkIds: string[]): Promise; // ******************** diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index da9f95263..861c6bcfd 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -185,8 +185,8 @@ 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 + $addToSet: { + pendingChecks: { $each: pending }, // accumulate pending checks; a retried arm never duplicates an entry }, }; try { @@ -203,6 +203,7 @@ class MongoJobsRepository implements IJobsRepository { } }; + // Returns whether this worker still holds the lease. false means another worker has claimed the row and the caller must stop. pullEvaluated = async (id: string, checkIds: string[]) => { if (checkIds.length === 0) return true; @@ -210,7 +211,7 @@ class MongoJobsRepository implements IJobsRepository { { _id: id, lockedBy: this.workerId }, // Only the worker that claimed can pull checks { $pull: { pendingChecks: { checkId: { $in: checkIds } } } } ); - return res.modifiedCount === 1; + return res.matchedCount === 1; }; upsertJob = async (job: JobSeed) => { diff --git a/server/src/service/bufferService.ts b/server/src/service/bufferService.ts index eceaca4c7..576b07ab1 100755 --- a/server/src/service/bufferService.ts +++ b/server/src/service/bufferService.ts @@ -36,6 +36,9 @@ export class BufferService implements IBufferService { private dockerLogsService: IDockerLogsService; private jobsRepository: IJobsRepository; + // Checks that are stored but whose evaluate row could not be armed. Retried on the next flush so they are never unevaluated. + private unarmed: Map = new Map(); + constructor( logger: ILogger, checkService: ICheckService, @@ -102,20 +105,33 @@ export class BufferService implements IBufferService { } ingestChecks = async (checks: Check[]) => { - if (checks.length === 0) return; - await this.checksService.createChecks(checks); + if (checks.length > 0) await this.checksService.createChecks(checks); - const byMonitor = new Map(); + // Group by monitor, starting from any entries a previous flush failed to arm + const byMonitor = this.unarmed; + this.unarmed = 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); } + if (byMonitor.size === 0) return; const now = Date.now(); const ops: Promise[] = []; for (const [monitorId, pendingChecks] of byMonitor) { - ops.push(this.jobsRepository.upsertEvaluate(monitorId, pendingChecks, now)); + ops.push( + this.jobsRepository.upsertEvaluate(monitorId, pendingChecks, now).catch((error: unknown) => { + // The checks are already stored, so a failed arm is retried on the next flush rather than dropped + this.unarmed.set(monitorId, pendingChecks); + this.logger.error({ + message: `Could not arm evaluation for monitor ${monitorId}, retrying on next flush: ${error instanceof Error ? error.message : String(error)}`, + service: this.SERVICE_NAME, + method: "ingestChecks", + }); + return false; + }) + ); } await Promise.all(ops); }; @@ -144,7 +160,7 @@ export class BufferService implements IBufferService { } async flushBuffer() { - if (this.buffer.length === 0) { + if (this.buffer.length === 0 && this.unarmed.size === 0) { return; } // Take the batch first so a write that drains it can't be appended to mid-flush diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index d1cf4b4ea..05da1ee74 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -135,14 +135,26 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { 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.jobsRepository.pullEvaluated(job.id, [check.id]); + const leaseHeld = await this.jobsRepository.pullEvaluated(job.id, [check.id]); + if (!leaseHeld) { + // Another worker has claimed this row and is evaluating the remaining ids; stop here so nothing is applied twice + this.logger.warn({ message: `Lost lease on ${job.id} after check ${check.id}, stopping`, service: SERVICE_NAME, method: "runEvaluate" }); + return; + } current = evaluation.statusChange.monitor; // fresh statusWindow/status/counters for the next check } - // Clean up ids potentially orphaned retention cleanup + // Ids whose check was not returned (deleted by retention cleanup) would otherwise sit on the row forever 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); + if (missing.length > 0) { + this.logger.warn({ + message: `Dropping ${missing.length} pending checks with no stored check: ${missing.join(", ")}`, + service: SERVICE_NAME, + method: "runEvaluate", + }); + 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 index 221c33c78..57ce10e46 100644 --- a/server/test/integration/evaluateCursorMultiWorker.test.ts +++ b/server/test/integration/evaluateCursorMultiWorker.test.ts @@ -17,20 +17,20 @@ 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. +// Regression test for the dropped-check bug in the check-to-evaluate handoff. // -// 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. +// The evaluate step used to be driven by a timestamp cursor: the evaluator read checks with +// createdAt strictly after the monitor's lastEvaluatedAt and advanced that cursor to each +// check's createdAt. Each worker's BufferService flushes its own in-memory batch on its own +// timer, so with two processing workers a later check produced on worker B could be flushed and +// evaluated before an earlier check produced on worker A had been flushed. When A's batch landed, +// its check was older than the cursor and the evaluate query never returned it. The check existed +// in the collection but never reached 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. +// The evaluate row now carries the ids of every stored check, pushed at ingest and pulled as each +// is applied, so a late batch is always evaluated. 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 handoff semantics under test are MongoDB's, not a mock's. let mongod: MongoMemoryServer; diff --git a/server/test/integration/jobsRepository.test.ts b/server/test/integration/jobsRepository.test.ts index 467a4d670..3ff1ea277 100644 --- a/server/test/integration/jobsRepository.test.ts +++ b/server/test/integration/jobsRepository.test.ts @@ -376,6 +376,14 @@ describe("MongoJobsRepository", () => { expect(row?.nextScheduledAt).toBe(NOW); }); + it("never duplicates an entry when the same batch is armed twice (retry after a failed arm)", async () => { + await repo.upsertEvaluate("mon-1", pending("c1", "c2"), NOW); + await repo.upsertEvaluate("mon-1", pending("c1", "c2"), NOW + 1_000); + + const row = await readRow("evaluate:mon-1"); + expect(row?.pendingChecks).toEqual(pending("c1", "c2")); + }); + 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 @@ -445,6 +453,19 @@ describe("MongoJobsRepository", () => { expect(row?.pendingChecks.map((entry) => entry.checkId)).toEqual(["c1"]); }); + it("reports the lease as held even when the ids were already pulled", async () => { + await seedJob({ + _id: "evaluate:mon-1", + type: "evaluate", + intervalMs: null, + lockedBy: ownedBy(repo), + lockedUntil: NOW + LOCK_MS, + pendingChecks: [], + }); + + expect(await repo.pullEvaluated("evaluate:mon-1", ["c1"])).toBe(true); + }); + it("is a no-op for an empty id list", async () => { await seedJob({ _id: "evaluate:mon-1", @@ -474,6 +495,7 @@ describe("MongoJobsRepository", () => { intervalMs: 60_000, lockedBy: null, lockedUntil: null, + pendingChecks: [], runCount: 0, failCount: 0, lastFinishedAt: null, @@ -505,6 +527,7 @@ describe("MongoJobsRepository", () => { intervalMs: 86_400_000, lockedBy: null, lockedUntil: null, + pendingChecks: [], runCount: 0, failCount: 0, lastFinishedAt: null, diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index 06522abbc..2788af4be 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/server/test/unit/services/bufferService.test.ts @@ -407,6 +407,24 @@ describe("BufferService", () => { expect(jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-2", [{ checkId: "c3", createdAt }], expect.any(Number)); }); + it("retries a failed arm on the next flush without re-storing the checks", async () => { + const { service, checkService, jobsRepository, logger } = createService(); + (jobsRepository.upsertEvaluate as jest.Mock).mockRejectedValueOnce(new Error("arm failed")); + service.addToBuffer(makeCheck({ id: "c1" })); + + await service.flushBuffer(); + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ method: "ingestChecks" })); + + await service.flushBuffer(); // buffer is empty, only the retry runs + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + expect(jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2); + expect(jobsRepository.upsertEvaluate).toHaveBeenLastCalledWith("mon-1", [{ checkId: "c1", createdAt: expect.any(Number) }], expect.any(Number)); + + await service.flushBuffer(); // nothing left to retry + expect(jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2); + }); + it("ingestChecks with an empty batch neither writes nor arms", async () => { const { service, checkService, jobsRepository } = createService(); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 4ea611a4d..b8bf85688 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -421,6 +421,22 @@ describe("DBQueueWorker", () => { expect(mocks.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1", "c2"]); }); + it("evaluate job stops applying checks once the lease is lost so a second claimer does not double-apply", async () => { + const job = makeEvaluateJob("c1", "c2", "c3"); + const checksRepository = { findUnevaluatedByMonitorId: jest.fn().mockResolvedValue([{ id: "c1" }, { id: "c2" }, { id: "c3" }]) }; + const jobsRepository = { + ...createWorker().mocks.jobsRepository, + claimDueBatch: claimOnce(job), + pullEvaluated: jest.fn().mockResolvedValueOnce(true).mockResolvedValueOnce(false), // lease lost after c2 + }; + const { mocks } = await start({ mocks: { jobsRepository, checksRepository } }); + + expect(mocks.checkEvaluator.evaluate).toHaveBeenCalledTimes(2); + expect(mocks.dispatcher.dispatch).toHaveBeenCalledTimes(2); + expect(mocks.jobsRepository.pullEvaluated).toHaveBeenCalledTimes(2); // no trailing sweep either + expect(mocks.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ method: "runEvaluate" })); + }); + 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