629 lines
27 KiB
TypeScript
629 lines
27 KiB
TypeScript
import { describe, expect, it, beforeAll, afterAll, beforeEach } from "@jest/globals";
|
|
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";
|
|
|
|
// ── Real-Mongo harness ─────────────────────────────────────────────────────────
|
|
// The headline guarantee of the queue — "exactly one claimer wins" — IS MongoDB's
|
|
// single-document atomicity on findOneAndUpdate, so it can only be exercised against a
|
|
// live engine. We boot a throwaway in-process mongod for this suite.
|
|
|
|
let mongod: MongoMemoryServer;
|
|
|
|
beforeAll(async () => {
|
|
mongod = await MongoMemoryServer.create();
|
|
await mongoose.connect(mongod.getUri());
|
|
await JobModel.init(); // build the indexes the claim scan relies on
|
|
}, 120_000);
|
|
|
|
afterAll(async () => {
|
|
await mongoose.disconnect();
|
|
await mongod.stop();
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
await JobModel.deleteMany({});
|
|
});
|
|
|
|
// ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
const NOW = 1_000_000;
|
|
|
|
// Insert a job row directly, full control over every field. Defaults describe a due,
|
|
// unlocked, active check job.
|
|
const seedJob = (overrides: Partial<JobDocument> = {}): Promise<JobDocument> =>
|
|
JobModel.create({
|
|
_id: overrides._id ?? `check:${overrides.refId ?? "mon-1"}`,
|
|
type: "check" as JobType,
|
|
refId: "mon-1",
|
|
isActive: true,
|
|
nextScheduledAt: NOW,
|
|
intervalMs: 60_000,
|
|
lockedBy: null,
|
|
lockedUntil: null,
|
|
runCount: 0,
|
|
failCount: 0,
|
|
lastFinishedAt: null,
|
|
lastFailReason: null,
|
|
...overrides,
|
|
});
|
|
|
|
const readRow = (id: string) => JobModel.findById(id).lean<JobDocument>();
|
|
|
|
// record*/renewLocks are fenced on the repo's private workerId, so a pre-seeded "held"
|
|
// lock must be stamped with the same id the repo will check against.
|
|
const ownedBy = (r: MongoJobsRepository) => (r as unknown as { workerId: string }).workerId;
|
|
|
|
const WORKER_ID = "worker-under-test";
|
|
|
|
describe("MongoJobsRepository", () => {
|
|
let repo: MongoJobsRepository;
|
|
|
|
beforeEach(() => {
|
|
repo = new MongoJobsRepository(WORKER_ID);
|
|
});
|
|
|
|
// ── claimDue ───────────────────────────────────────────────────────────────
|
|
describe("claimDue", () => {
|
|
it("claims a due, unlocked job and leases it", async () => {
|
|
await seedJob();
|
|
|
|
const claimed = await repo.claimDue("check", NOW);
|
|
|
|
expect(claimed?.id).toBe("check:mon-1");
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.lockedBy).not.toBeNull();
|
|
expect(row?.lockedUntil).toBe(NOW + LOCK_MS);
|
|
});
|
|
|
|
it("returns null when nothing is due yet", async () => {
|
|
await seedJob({ nextScheduledAt: NOW + 1 });
|
|
expect(await repo.claimDue("check", NOW)).toBeNull();
|
|
});
|
|
|
|
it("skips inactive (paused) jobs", async () => {
|
|
await seedJob({ isActive: false });
|
|
expect(await repo.claimDue("check", NOW)).toBeNull();
|
|
});
|
|
|
|
it("only claims the requested type", async () => {
|
|
await seedJob({ _id: "geo:mon-1", type: "geo-check" });
|
|
expect(await repo.claimDue("check", NOW)).toBeNull();
|
|
});
|
|
|
|
it("does not claim a job held by a live lease", async () => {
|
|
await seedJob({ lockedBy: "other-worker", lockedUntil: NOW + LOCK_MS });
|
|
expect(await repo.claimDue("check", NOW)).toBeNull();
|
|
});
|
|
|
|
it("reclaims a job whose lease has expired (crash recovery)", async () => {
|
|
await seedJob({ lockedBy: "dead-worker", lockedUntil: NOW - 1 });
|
|
|
|
const claimed = await repo.claimDue("check", NOW);
|
|
|
|
expect(claimed?.id).toBe("check:mon-1");
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.lockedBy).not.toBe("dead-worker");
|
|
expect(row?.lockedUntil).toBe(NOW + LOCK_MS);
|
|
});
|
|
|
|
it("returns the oldest-due job first", async () => {
|
|
await seedJob({ _id: "check:newer", refId: "newer", nextScheduledAt: NOW });
|
|
await seedJob({ _id: "check:older", refId: "older", nextScheduledAt: NOW - 5_000 });
|
|
|
|
const claimed = await repo.claimDue("check", NOW);
|
|
expect(claimed?.id).toBe("check:older");
|
|
});
|
|
|
|
it("lets exactly one of many concurrent claimers win a single due job", async () => {
|
|
await seedJob();
|
|
|
|
// Distinct repo instances = distinct workerIds = realistic competing workers.
|
|
const workers = Array.from({ length: 20 }, (_, i) => new MongoJobsRepository(`worker-${i}`));
|
|
const results = await Promise.all(workers.map((w) => w.claimDue("check", NOW)));
|
|
|
|
const winners = results.filter((r): r is Job => r !== null);
|
|
expect(winners).toHaveLength(1);
|
|
expect(winners[0].id).toBe("check:mon-1");
|
|
});
|
|
|
|
it("hands two concurrent claimers two distinct due jobs (no double-claim)", async () => {
|
|
await seedJob({ _id: "check:a", refId: "a" });
|
|
await seedJob({ _id: "check:b", refId: "b" });
|
|
|
|
const [w1, w2] = [new MongoJobsRepository("worker-1"), new MongoJobsRepository("worker-2")];
|
|
const [r1, r2] = await Promise.all([w1.claimDue("check", NOW), w2.claimDue("check", NOW)]);
|
|
|
|
expect(r1).not.toBeNull();
|
|
expect(r2).not.toBeNull();
|
|
expect(r1?.id).not.toBe(r2?.id);
|
|
expect([r1?.id, r2?.id].sort()).toEqual(["check:a", "check:b"]);
|
|
});
|
|
});
|
|
|
|
// ── recordSuccess (fixed-rate reschedule) ────────────────────────────────────
|
|
describe("recordSuccess", () => {
|
|
it("reschedules fixed-rate from the scheduled tick and releases the lease", async () => {
|
|
await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS, runCount: 4 });
|
|
|
|
const ok = await repo.recordSuccess("check:mon-1", NOW, 60_000, NOW + 1_500);
|
|
|
|
expect(ok).toBe(true);
|
|
const row = await readRow("check:mon-1");
|
|
// next tick measured from the scheduled time (NOW), not completion time
|
|
expect(row?.nextScheduledAt).toBe(NOW + 60_000);
|
|
expect(row?.lockedBy).toBeNull();
|
|
expect(row?.lockedUntil).toBeNull();
|
|
expect(row?.lastFinishedAt).toBe(NOW + 1_500);
|
|
expect(row?.runCount).toBe(5);
|
|
});
|
|
|
|
it("skips missed ticks when the worker has fallen behind, with jitter to avoid a herd", async () => {
|
|
await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS });
|
|
|
|
// scheduled NOW, interval 60s, but completion is 200s late → NOW+60s is already past
|
|
const now = NOW + 200_000;
|
|
await repo.recordSuccess("check:mon-1", NOW, 60_000, now);
|
|
|
|
const row = await readRow("check:mon-1");
|
|
// no catch-up burst: rebased to one interval from now, plus up to one interval of jitter
|
|
// so behind-jobs don't re-synchronize. Never earlier than now + interval (no over-checking).
|
|
expect(row?.nextScheduledAt).toBeGreaterThanOrEqual(now + 60_000);
|
|
expect(row?.nextScheduledAt).toBeLessThan(now + 120_000);
|
|
});
|
|
|
|
it("returns false for an unknown id", async () => {
|
|
expect(await repo.recordSuccess("missing", NOW, 60_000, NOW)).toBe(false);
|
|
});
|
|
|
|
it("does not release a lock owned by another worker (fenced no-op)", async () => {
|
|
// A worker whose lease expired and was reclaimed must not clobber the new owner's lock.
|
|
await seedJob({ lockedBy: "other-worker", lockedUntil: NOW + LOCK_MS, runCount: 7, failCount: 2, lastFailReason: "timeout" });
|
|
|
|
const ok = await repo.recordSuccess("check:mon-1", NOW, 60_000, NOW + 1_000);
|
|
|
|
expect(ok).toBe(false);
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.lockedBy).toBe("other-worker"); // lock untouched
|
|
expect(row?.lockedUntil).toBe(NOW + LOCK_MS);
|
|
expect(row?.runCount).toBe(7); // not double-counted
|
|
expect(row?.failCount).toBe(2); // failure state untouched
|
|
expect(row?.lastFailReason).toBe("timeout");
|
|
});
|
|
|
|
it("clears the failure state so failCount counts consecutive failures", async () => {
|
|
await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS, failCount: 3, lastFailReason: "connection refused" });
|
|
|
|
await repo.recordSuccess("check:mon-1", NOW, 60_000, NOW + 1_000);
|
|
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.failCount).toBe(0);
|
|
expect(row?.lastFailReason).toBeNull();
|
|
});
|
|
});
|
|
|
|
// ── recordOneShot (park after a single run) ─────────────────────────────────
|
|
describe("recordOneShot", () => {
|
|
it("parks the job, releases the lease, counts the run and clears the failure state", async () => {
|
|
await seedJob({
|
|
intervalMs: null,
|
|
lockedBy: ownedBy(repo),
|
|
lockedUntil: NOW + LOCK_MS,
|
|
runCount: 1,
|
|
failCount: 2,
|
|
lastFailReason: "connection refused",
|
|
});
|
|
|
|
const ok = await repo.recordOneShot("check:mon-1", NOW + 1_000);
|
|
|
|
expect(ok).toBe(true);
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.nextScheduledAt).toBe(PARKED);
|
|
expect(row?.lockedBy).toBeNull();
|
|
expect(row?.lockedUntil).toBeNull();
|
|
expect(row?.lastFinishedAt).toBe(NOW + 1_000);
|
|
expect(row?.runCount).toBe(2);
|
|
expect(row?.failCount).toBe(0);
|
|
expect(row?.lastFailReason).toBeNull();
|
|
});
|
|
|
|
it("does not touch a lock owned by another worker (fenced no-op)", async () => {
|
|
await seedJob({ intervalMs: null, lockedBy: "other-worker", lockedUntil: NOW + LOCK_MS, failCount: 2, lastFailReason: "timeout" });
|
|
|
|
expect(await repo.recordOneShot("check:mon-1", NOW + 1_000)).toBe(false);
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.nextScheduledAt).toBe(NOW); // not parked
|
|
expect(row?.lockedBy).toBe("other-worker");
|
|
expect(row?.failCount).toBe(2);
|
|
expect(row?.lastFailReason).toBe("timeout");
|
|
});
|
|
});
|
|
|
|
// ── renewLocks (lease renewal) ───────────────────────────────────────────────
|
|
describe("renewLocks", () => {
|
|
it("extends the lease on a lock this worker holds", async () => {
|
|
await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS });
|
|
|
|
const renewed = await repo.renewLocks(["check:mon-1"], NOW + 1_000);
|
|
|
|
expect(renewed).toBe(1);
|
|
expect((await readRow("check:mon-1"))?.lockedUntil).toBe(NOW + 1_000 + LOCK_MS);
|
|
});
|
|
|
|
it("does not renew a lock held by another worker (fenced)", async () => {
|
|
await seedJob({ lockedBy: "other-worker", lockedUntil: NOW + LOCK_MS });
|
|
|
|
const renewed = await repo.renewLocks(["check:mon-1"], NOW + 1_000);
|
|
|
|
expect(renewed).toBe(0);
|
|
expect((await readRow("check:mon-1"))?.lockedUntil).toBe(NOW + LOCK_MS); // unchanged
|
|
});
|
|
|
|
it("renews only the locks this worker holds out of a mixed batch", async () => {
|
|
await seedJob({ _id: "check:mine", refId: "mine", lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS });
|
|
await seedJob({ _id: "check:theirs", refId: "theirs", lockedBy: "other-worker", lockedUntil: NOW + LOCK_MS });
|
|
|
|
const renewed = await repo.renewLocks(["check:mine", "check:theirs"], NOW + 1_000);
|
|
|
|
expect(renewed).toBe(1);
|
|
expect((await readRow("check:mine"))?.lockedUntil).toBe(NOW + 1_000 + LOCK_MS);
|
|
expect((await readRow("check:theirs"))?.lockedUntil).toBe(NOW + LOCK_MS);
|
|
});
|
|
|
|
it("returns 0 for an empty id list without touching the db", async () => {
|
|
expect(await repo.renewLocks([], NOW)).toBe(0);
|
|
});
|
|
|
|
// Regression: a renewed lock keeps a slow-but-alive job from being reclaimed once its
|
|
// original lease would have expired — preventing the duplicate-execution path.
|
|
it("keeps a job from being reclaimed after the original lease window passes", async () => {
|
|
await seedJob();
|
|
await repo.claimDue("check", NOW); // this worker now holds the lease until NOW + LOCK_MS
|
|
|
|
// just before expiry, the still-running job renews its own lock
|
|
const renewAt = NOW + LOCK_MS - 1_000;
|
|
expect(await repo.renewLocks(["check:mon-1"], renewAt)).toBe(1);
|
|
|
|
// a second worker scans PAST the original lease end — without renewal it would reclaim
|
|
const other = new MongoJobsRepository("competing-worker");
|
|
expect(await other.claimDue("check", NOW + LOCK_MS + 1)).toBeNull();
|
|
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.lockedBy).toBe(ownedBy(repo)); // still held by the original worker
|
|
expect(row?.lockedUntil).toBe(renewAt + LOCK_MS);
|
|
});
|
|
});
|
|
|
|
// ── recordFailure (backoff) ──────────────────────────────────────────────────
|
|
describe("recordFailure", () => {
|
|
it("backs off, records the reason, bumps failCount, releases the lease", async () => {
|
|
await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS, failCount: 2 });
|
|
|
|
const ok = await repo.recordFailure("check:mon-1", new Error("connection refused"), NOW);
|
|
|
|
expect(ok).toBe(true);
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.nextScheduledAt).toBe(NOW + BACKOFF_MS);
|
|
expect(row?.lastFailReason).toBe("connection refused");
|
|
expect(row?.lastFinishedAt).toBe(NOW);
|
|
expect(row?.failCount).toBe(3);
|
|
expect(row?.lockedBy).toBeNull();
|
|
expect(row?.lockedUntil).toBeNull();
|
|
});
|
|
|
|
it("stringifies a non-Error reason", async () => {
|
|
await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS });
|
|
await repo.recordFailure("check:mon-1", "weird non-error", NOW);
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.lastFailReason).toBe("weird non-error");
|
|
});
|
|
|
|
it("returns false for an unknown id", async () => {
|
|
expect(await repo.recordFailure("missing", new Error("x"), NOW)).toBe(false);
|
|
});
|
|
});
|
|
|
|
// ── upsertEvaluate (handoff + $min coalescing) ───────────────────────────────
|
|
describe("upsertEvaluate", () => {
|
|
it("creates a single evaluate row keyed by monitor", async () => {
|
|
await repo.upsertEvaluate("mon-1", NOW);
|
|
|
|
const row = await readRow("evaluate:mon-1");
|
|
expect(row?.type).toBe("evaluate");
|
|
expect(row?.refId).toBe("mon-1");
|
|
expect(row?.nextScheduledAt).toBe(NOW);
|
|
expect(row?.intervalMs).toBeNull();
|
|
expect(row?.isActive).toBe(true);
|
|
});
|
|
|
|
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);
|
|
|
|
const rows = await JobModel.find({ type: "evaluate", refId: "mon-1" }).lean<JobDocument[]>();
|
|
expect(rows).toHaveLength(1);
|
|
expect(rows[0].nextScheduledAt).toBe(NOW); // earliest pending wins
|
|
});
|
|
|
|
it("lowers runAt when an earlier evaluation arrives ($min)", async () => {
|
|
await repo.upsertEvaluate("mon-1", NOW + 9_000);
|
|
await repo.upsertEvaluate("mon-1", 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)));
|
|
|
|
const count = await JobModel.countDocuments({ type: "evaluate", refId: "mon-1" });
|
|
expect(count).toBe(1);
|
|
});
|
|
});
|
|
|
|
// ── monitor lifecycle CRUD ───────────────────────────────────────────────────
|
|
describe("CRUD", () => {
|
|
it("upsertJob inserts, then refreshes config without clobbering scheduling state", async () => {
|
|
const job: Job = {
|
|
id: "check:mon-1",
|
|
type: "check",
|
|
refId: "mon-1",
|
|
isActive: true,
|
|
nextScheduledAt: NOW,
|
|
intervalMs: 60_000,
|
|
lockedBy: null,
|
|
lockedUntil: null,
|
|
runCount: 0,
|
|
failCount: 0,
|
|
lastFinishedAt: null,
|
|
lastFailReason: null,
|
|
};
|
|
await repo.upsertJob(job);
|
|
|
|
// simulate a running job that has advanced + holds a lease
|
|
await JobModel.updateOne({ _id: "check:mon-1" }, { $set: { nextScheduledAt: NOW + 5_000, lockedBy: "live", runCount: 7 } });
|
|
|
|
// a reconcile sweep re-upserts the same id with new config
|
|
await repo.upsertJob({ ...job, intervalMs: 30_000, isActive: false });
|
|
|
|
const row = await readRow("check:mon-1");
|
|
expect(row?.intervalMs).toBe(30_000); // config refreshed ($set)
|
|
expect(row?.isActive).toBe(false);
|
|
expect(row?.nextScheduledAt).toBe(NOW + 5_000); // scheduling untouched ($setOnInsert)
|
|
expect(row?.lockedBy).toBe("live");
|
|
expect(row?.runCount).toBe(7);
|
|
});
|
|
|
|
it("upsertCleanupJob resets nextScheduledAt on an existing row so it is due again on restart", async () => {
|
|
const job: Job = {
|
|
id: "cleanup-orphaned",
|
|
type: "cleanup-orphaned",
|
|
refId: null,
|
|
isActive: true,
|
|
nextScheduledAt: NOW,
|
|
intervalMs: 86_400_000,
|
|
lockedBy: null,
|
|
lockedUntil: null,
|
|
runCount: 0,
|
|
failCount: 0,
|
|
lastFinishedAt: null,
|
|
lastFailReason: null,
|
|
};
|
|
await repo.upsertCleanupJob(job);
|
|
|
|
// row has already advanced a day into the future after a previous run
|
|
await JobModel.updateOne({ _id: "cleanup-orphaned" }, { $set: { nextScheduledAt: NOW + 86_400_000, runCount: 3 } });
|
|
|
|
// a restart reconcile re-upserts and pulls it due now
|
|
await repo.upsertCleanupJob({ ...job, nextScheduledAt: NOW + 1_000 });
|
|
|
|
const row = await readRow("cleanup-orphaned");
|
|
expect(row?.nextScheduledAt).toBe(NOW + 1_000); // schedule reset ($set)
|
|
expect(row?.runCount).toBe(3); // run history untouched ($setOnInsert)
|
|
});
|
|
|
|
it("setActiveById flips check and geo rows together", async () => {
|
|
await seedJob({ _id: "check:mon-1", type: "check" });
|
|
await seedJob({ _id: "geo:mon-1", type: "geo-check" });
|
|
|
|
await repo.setActiveById("mon-1", false);
|
|
|
|
const rows = await JobModel.find({ refId: "mon-1" }).lean<JobDocument[]>();
|
|
expect(rows).toHaveLength(2);
|
|
expect(rows.every((r) => r.isActive === false)).toBe(true);
|
|
});
|
|
|
|
it("updateScheduleById changes the interval on a single typed row", async () => {
|
|
await seedJob({ _id: "check:mon-1", type: "check", intervalMs: 60_000 });
|
|
await seedJob({ _id: "geo:mon-1", type: "geo-check", intervalMs: 60_000 });
|
|
|
|
await repo.updateScheduleById("mon-1", "check", 15_000);
|
|
|
|
expect((await readRow("check:mon-1"))?.intervalMs).toBe(15_000);
|
|
expect((await readRow("geo:mon-1"))?.intervalMs).toBe(60_000); // untouched
|
|
});
|
|
|
|
it("markMonitorsDue re-arms check and geo rows, capping the jitter for long intervals", async () => {
|
|
const armAt = NOW + 5_000_000; // well past the seeded nextScheduledAt so the jittered value is unambiguous
|
|
await seedJob({ _id: "check:mon-1", refId: "mon-1", type: "check", intervalMs: 10_000, nextScheduledAt: NOW }); // shorter than the cap
|
|
await seedJob({ _id: "geo:mon-1", refId: "mon-1", type: "geo-check", intervalMs: 3_600_000, nextScheduledAt: NOW }); // far longer than the cap
|
|
|
|
const modified = await repo.markMonitorsDue(["mon-1"], armAt);
|
|
|
|
expect(modified).toBe(2);
|
|
const check = await readRow("check:mon-1");
|
|
const geo = await readRow("geo:mon-1");
|
|
// Short interval spreads over its own interval; long interval is capped so it still runs soon
|
|
expect(check!.nextScheduledAt).toBeGreaterThanOrEqual(armAt);
|
|
expect(check!.nextScheduledAt).toBeLessThan(armAt + 10_000);
|
|
expect(geo!.nextScheduledAt).toBeGreaterThanOrEqual(armAt);
|
|
expect(geo!.nextScheduledAt).toBeLessThan(armAt + 15_000); // REARM_JITTER_MS, not the hour-long interval
|
|
});
|
|
|
|
it("markMonitorsDue only touches the listed monitors and ignores evaluate/cleanup rows", async () => {
|
|
await seedJob({ _id: "check:mon-1", refId: "mon-1", type: "check", nextScheduledAt: NOW });
|
|
await seedJob({ _id: "check:mon-2", refId: "mon-2", type: "check", nextScheduledAt: NOW });
|
|
await seedJob({ _id: "evaluate:mon-1", refId: "mon-1", type: "evaluate", nextScheduledAt: NOW });
|
|
|
|
const modified = await repo.markMonitorsDue(["mon-1"], NOW + 1_000);
|
|
|
|
expect(modified).toBe(1); // only check:mon-1
|
|
expect((await readRow("check:mon-2"))?.nextScheduledAt).toBe(NOW); // other monitor untouched
|
|
expect((await readRow("evaluate:mon-1"))?.nextScheduledAt).toBe(NOW); // evaluate row untouched
|
|
});
|
|
|
|
it("markMonitorsDue is a no-op for an empty monitor list", async () => {
|
|
await seedJob({ _id: "check:mon-1", refId: "mon-1", type: "check", nextScheduledAt: NOW });
|
|
|
|
const modified = await repo.markMonitorsDue([], NOW + 1_000);
|
|
|
|
expect(modified).toBe(0);
|
|
expect((await readRow("check:mon-1"))?.nextScheduledAt).toBe(NOW);
|
|
});
|
|
|
|
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.deleteById("mon-1");
|
|
|
|
expect(await JobModel.countDocuments({ refId: "mon-1" })).toBe(0);
|
|
});
|
|
|
|
it("deleteByMonitorIdsNotIn drops jobs for missing monitors but keeps valid and global rows", async () => {
|
|
await seedJob({ _id: "check:keep", refId: "keep", type: "check" });
|
|
await seedJob({ _id: "geo:keep", refId: "keep", type: "geo-check" });
|
|
await seedJob({ _id: "check:gone", refId: "gone", type: "check" });
|
|
await seedJob({ _id: "cleanup-orphaned", refId: null, type: "cleanup-orphaned" });
|
|
await seedJob({ _id: "cleanup-retention", refId: null, type: "cleanup-retention" });
|
|
|
|
const deleted = await repo.deleteByMonitorIdsNotIn(["keep"]);
|
|
|
|
expect(deleted).toBe(1); // only check:gone
|
|
const remaining = (await JobModel.find().lean<JobDocument[]>()).map((d) => d._id).sort();
|
|
expect(remaining).toEqual(["check:keep", "cleanup-orphaned", "cleanup-retention", "geo:keep"]);
|
|
});
|
|
|
|
it("findById returns all of a monitor's rows", async () => {
|
|
await seedJob({ _id: "check:mon-1", type: "check" });
|
|
await seedJob({ _id: "geo:mon-1", type: "geo-check" });
|
|
|
|
const jobs = await repo.findById("mon-1");
|
|
expect(jobs.map((j) => j.id).sort()).toEqual(["check:mon-1", "geo:mon-1"]);
|
|
});
|
|
});
|
|
|
|
// ── observability ────────────────────────────────────────────────────────────
|
|
describe("observability", () => {
|
|
it("findPage returns a page plus the total count", async () => {
|
|
await seedJob({ _id: "check:a", refId: "a", nextScheduledAt: NOW + 1 });
|
|
await seedJob({ _id: "check:b", refId: "b", nextScheduledAt: NOW + 2 });
|
|
await seedJob({ _id: "check:c", refId: "c", nextScheduledAt: NOW + 3 });
|
|
|
|
const { jobs, count } = await repo.findPage({ page: 0, rowsPerPage: 2 });
|
|
expect(count).toBe(3);
|
|
expect(jobs).toHaveLength(2);
|
|
expect(jobs.map((j) => j.id)).toEqual(["check:a", "check:b"]); // sorted by nextScheduledAt
|
|
});
|
|
|
|
it("findPage returns everything when rowsPerPage is 0", async () => {
|
|
await seedJob({ _id: "check:a", refId: "a" });
|
|
await seedJob({ _id: "check:b", refId: "b" });
|
|
|
|
const { jobs, count } = await repo.findPage({});
|
|
expect(count).toBe(2);
|
|
expect(jobs).toHaveLength(2);
|
|
});
|
|
|
|
it("findAll returns every row", async () => {
|
|
await seedJob({ _id: "check:a", refId: "a" });
|
|
await seedJob({ _id: "geo:b", refId: "b", type: "geo-check" });
|
|
|
|
expect(await repo.findAll()).toHaveLength(2);
|
|
});
|
|
});
|
|
|
|
// ── countDueBacklog (scaling signal) ─────────────────────────────────────────
|
|
describe("countDueBacklog", () => {
|
|
it("counts a due, active, unlocked check job", async () => {
|
|
await seedJob();
|
|
expect(await repo.countDueBacklog(NOW)).toBe(1);
|
|
});
|
|
|
|
it("counts a check job whose lock has expired (reclaimable work)", async () => {
|
|
await seedJob({ lockedBy: "dead-worker", lockedUntil: NOW - 1 });
|
|
expect(await repo.countDueBacklog(NOW)).toBe(1);
|
|
});
|
|
|
|
it("excludes not-yet-due, inactive, live-locked, and non-check rows", async () => {
|
|
await seedJob({ _id: "check:due", refId: "due" }); // counts
|
|
await seedJob({ _id: "check:future", refId: "future", nextScheduledAt: NOW + 1 }); // not due
|
|
await seedJob({ _id: "check:paused", refId: "paused", isActive: false }); // inactive
|
|
await seedJob({ _id: "check:locked", refId: "locked", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // live lock
|
|
await seedJob({ _id: "geo:mon", refId: "mon", type: "geo-check" }); // wrong type
|
|
await seedJob({ _id: "evaluate:mon", refId: "mon", type: "evaluate" }); // wrong type
|
|
|
|
expect(await repo.countDueBacklog(NOW)).toBe(1); // only check:due
|
|
});
|
|
|
|
it("counts every qualifying check row across monitors", async () => {
|
|
await seedJob({ _id: "check:a", refId: "a" });
|
|
await seedJob({ _id: "check:b", refId: "b" });
|
|
await seedJob({ _id: "check:c", refId: "c", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // excluded
|
|
expect(await repo.countDueBacklog(NOW)).toBe(2);
|
|
});
|
|
|
|
it("returns 0 when nothing is due", async () => {
|
|
await seedJob({ nextScheduledAt: NOW + 10_000 });
|
|
expect(await repo.countDueBacklog(NOW)).toBe(0);
|
|
});
|
|
});
|
|
|
|
// ── global rows: egress recovery job ──────────────────────────────────────
|
|
|
|
describe("global rows", () => {
|
|
const repo = new MongoJobsRepository(WORKER_ID);
|
|
const egressRow = (nextScheduledAt: number) =>
|
|
seedJob({ _id: "egress", type: "egress" as JobType, refId: null, nextScheduledAt, intervalMs: 30_000 });
|
|
|
|
it("deleteGlobalJob removes the row by its canonical _id and leaves other global rows alone", async () => {
|
|
await egressRow(NOW);
|
|
await seedJob({ _id: "cleanup-orphaned", type: "cleanup-orphaned" as JobType, refId: null });
|
|
|
|
expect(await repo.deleteGlobalJob("egress")).toBe(true);
|
|
|
|
expect(await readRow("egress")).toBeNull();
|
|
expect(await readRow("cleanup-orphaned")).not.toBeNull();
|
|
expect(await repo.deleteGlobalJob("egress")).toBe(false);
|
|
});
|
|
|
|
it("deleteGlobalJobIfUnchanged removes the row only while nextScheduledAt matches the claimed value", async () => {
|
|
await egressRow(NOW);
|
|
|
|
// Another worker started a new episode and rescheduled the row after this run claimed it
|
|
await JobModel.updateOne({ _id: "egress" }, { $set: { nextScheduledAt: NOW + 30_000 } });
|
|
expect(await repo.deleteGlobalJobIfUnchanged("egress", NOW)).toBe(false);
|
|
expect(await readRow("egress")).not.toBeNull();
|
|
|
|
// Nobody touched it: the row goes
|
|
expect(await repo.deleteGlobalJobIfUnchanged("egress", NOW + 30_000)).toBe(true);
|
|
expect(await readRow("egress")).toBeNull();
|
|
});
|
|
|
|
it("recordSuccess after a refused conditional delete is a no-op once the lock has moved on", async () => {
|
|
await egressRow(NOW);
|
|
const [claimed] = await repo.claimDueBatch("egress", 1, NOW);
|
|
expect(claimed?.id).toBe("egress");
|
|
|
|
// Row deleted by the handler (recovered), then recordSuccess runs from the queue: nothing to update
|
|
expect(await repo.deleteGlobalJobIfUnchanged("egress", claimed!.nextScheduledAt)).toBe(true);
|
|
expect(await repo.recordSuccess("egress", claimed!.nextScheduledAt, 30_000, NOW + 1_000)).toBe(false);
|
|
expect(await readRow("egress")).toBeNull();
|
|
});
|
|
});
|
|
});
|