diff --git a/server/test/integration/jobsRepository.test.ts b/server/test/integration/jobsRepository.test.ts index f8aedc06f..d09174b84 100644 --- a/server/test/integration/jobsRepository.test.ts +++ b/server/test/integration/jobsRepository.test.ts @@ -52,6 +52,10 @@ const seedJob = (overrides: Partial = {}): Promise => const readRow = (id: string) => JobModel.findById(id).lean(); +// 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; + describe("MongoJobsRepository", () => { let repo: MongoJobsRepository; @@ -140,7 +144,7 @@ describe("MongoJobsRepository", () => { // ── recordSuccess (fixed-rate reschedule) ──────────────────────────────────── describe("recordSuccess", () => { it("reschedules fixed-rate from the scheduled tick and releases the lease", async () => { - await seedJob({ lockedBy: "me", lockedUntil: NOW + LOCK_MS, runCount: 4 }); + 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); @@ -155,7 +159,7 @@ describe("MongoJobsRepository", () => { }); it("skips missed ticks when the worker has fallen behind", async () => { - await seedJob({ lockedBy: "me", lockedUntil: NOW + LOCK_MS }); + 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; @@ -169,12 +173,80 @@ describe("MongoJobsRepository", () => { 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 }); + + 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 + }); + }); + + // ── 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(); + 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: "me", lockedUntil: NOW + LOCK_MS, failCount: 2 }); + await seedJob({ lockedBy: ownedBy(repo), lockedUntil: NOW + LOCK_MS, failCount: 2 }); const ok = await repo.recordFailure("check:mon-1", new Error("connection refused"), NOW); @@ -188,7 +260,7 @@ describe("MongoJobsRepository", () => { }); it("stringifies a non-Error reason", async () => { - await seedJob(); + 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");