537 lines
23 KiB
TypeScript
537 lines
23 KiB
TypeScript
import { describe, expect, it, jest, beforeEach, afterEach } from "@jest/globals";
|
|
import { DBQueueWorker } from "../../../src/worker/worker.db-queue.ts";
|
|
import { LOCK_MS } from "../../../src/domain/jobs/job.type.ts";
|
|
import type { Job } from "../../../src/domain/jobs/job.type.ts";
|
|
import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts";
|
|
import type { QueueMode } from "../../../src/domain/app-settings/app-settings.type.ts";
|
|
import { createMockLogger } from "../../helpers/createMockLogger.ts";
|
|
|
|
// ── Notes ──────────────────────────────────────────────────────────────────────
|
|
//
|
|
// DBQueueWorker is the orchestrator: it drives the polling loops, claims batches
|
|
// sized to free capacity, runs jobs (check / evaluate / geo-check / cleanup) and
|
|
// renews their locks, and registers itself in the worker heartbeat. The repository
|
|
// claim/lock atomicity is covered in integration/jobsRepository.test.ts; here we
|
|
// test only the orchestration around it, with the repository fully mocked.
|
|
//
|
|
// Fake timers are mandatory: init() starts self-re-arming setTimeout loops and a
|
|
// heartbeat setInterval, so without them the loops would leak across tests. The
|
|
// loop re-arms at POLL_MS (250ms) and the lock renews at LOCK_MS / 3 (20s).
|
|
|
|
const POLL_MS = 250;
|
|
const LOCK_RENEW_MS = LOCK_MS / 3;
|
|
|
|
const deferred = <T>() => {
|
|
let resolve!: (value: T) => void;
|
|
let reject!: (reason?: unknown) => void;
|
|
const promise = new Promise<T>((res, rej) => {
|
|
resolve = res;
|
|
reject = rej;
|
|
});
|
|
return { promise, resolve, reject };
|
|
};
|
|
|
|
const makeMonitor = (overrides?: Partial<Monitor>): Monitor =>
|
|
({
|
|
id: "m1",
|
|
teamId: "team",
|
|
type: "http",
|
|
interval: 60000,
|
|
status: "up",
|
|
isActive: true,
|
|
geoCheckEnabled: false,
|
|
geoCheckInterval: 300000,
|
|
lastEvaluatedAt: 0,
|
|
...overrides,
|
|
}) as Monitor;
|
|
|
|
const makeJob = (overrides?: Partial<Job>): Job => ({
|
|
id: "check:m1",
|
|
type: "check",
|
|
refId: "m1",
|
|
isActive: true,
|
|
nextScheduledAt: 1000,
|
|
intervalMs: 60000,
|
|
lockedBy: null,
|
|
lockedUntil: null,
|
|
runCount: 0,
|
|
failCount: 0,
|
|
lastFinishedAt: null,
|
|
lastFailReason: null,
|
|
...overrides,
|
|
});
|
|
|
|
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),
|
|
recordSuccess: jest.fn<any>().mockResolvedValue(true),
|
|
recordFailure: jest.fn<any>().mockResolvedValue(true),
|
|
recordOneShot: jest.fn<any>().mockResolvedValue(true),
|
|
renewLocks: jest.fn<any>().mockResolvedValue(1),
|
|
deleteById: jest.fn<any>().mockResolvedValue(true),
|
|
deleteByIdAndType: jest.fn<any>().mockResolvedValue(true),
|
|
setActiveById: jest.fn<any>().mockResolvedValue(true),
|
|
updateScheduleById: jest.fn<any>().mockResolvedValue(true),
|
|
findAll: jest.fn<any>().mockResolvedValue([]),
|
|
findPage: jest.fn<any>().mockResolvedValue({ jobs: [], count: 0 }),
|
|
};
|
|
const monitorsRepository = {
|
|
findByIds: jest.fn<any>().mockResolvedValue([makeMonitor()]),
|
|
findAll: jest.fn<any>().mockResolvedValue([]),
|
|
updateById: jest.fn<any>().mockResolvedValue({}),
|
|
};
|
|
const checksRepository = {
|
|
findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([]),
|
|
};
|
|
const checkService = {
|
|
toStatusResponse: jest.fn<any>().mockReturnValue({ status: "up" }),
|
|
toLastEvaluatedAt: jest.fn<any>().mockReturnValue(12345),
|
|
};
|
|
const checkProducer = {
|
|
produce: jest.fn<any>().mockResolvedValue({ status: { status: "up" }, check: { id: "c1" } }),
|
|
};
|
|
const checkEvaluator = {
|
|
evaluate: jest.fn<any>().mockResolvedValue({ monitor: makeMonitor(), decision: {} }),
|
|
};
|
|
const geoCheckPipeline = { run: jest.fn<any>().mockResolvedValue(null) };
|
|
const dispatcher = { dispatch: jest.fn<any>().mockResolvedValue(undefined) };
|
|
const helper = {
|
|
getCleanupOrphanedJob: jest.fn<any>().mockReturnValue(jest.fn<any>().mockResolvedValue(undefined)),
|
|
getCleanupRetentionJob: jest.fn<any>().mockReturnValue(jest.fn<any>().mockResolvedValue(undefined)),
|
|
};
|
|
const queueWorkersRepository = {
|
|
upsert: jest.fn<any>().mockResolvedValue(undefined),
|
|
deleteById: jest.fn<any>().mockResolvedValue(undefined),
|
|
findRecent: jest.fn<any>().mockResolvedValue([]),
|
|
};
|
|
const logger = createMockLogger();
|
|
|
|
const mocks = {
|
|
jobsRepository,
|
|
monitorsRepository,
|
|
checksRepository,
|
|
checkService,
|
|
checkProducer,
|
|
checkEvaluator,
|
|
geoCheckPipeline,
|
|
dispatcher,
|
|
helper,
|
|
queueWorkersRepository,
|
|
logger,
|
|
...overrides?.mocks,
|
|
};
|
|
|
|
const worker = new DBQueueWorker(
|
|
mocks.logger as any,
|
|
mocks.jobsRepository as any,
|
|
mocks.monitorsRepository as any,
|
|
mocks.checksRepository as any,
|
|
mocks.checkService as any,
|
|
mocks.checkProducer as any,
|
|
mocks.checkEvaluator as any,
|
|
mocks.geoCheckPipeline as any,
|
|
mocks.dispatcher as any,
|
|
mocks.helper as any,
|
|
mocks.queueWorkersRepository as any,
|
|
overrides?.queueMode ?? "worker",
|
|
overrides?.queuePrimaryProcesses ?? true,
|
|
"worker-1"
|
|
);
|
|
return { worker, mocks };
|
|
};
|
|
|
|
describe("DBQueueWorker", () => {
|
|
let active: DBQueueWorker | null = null;
|
|
|
|
beforeEach(() => {
|
|
jest.useFakeTimers();
|
|
});
|
|
|
|
afterEach(async () => {
|
|
if (active) await active.shutdown();
|
|
active = null;
|
|
jest.clearAllTimers();
|
|
jest.useRealTimers();
|
|
});
|
|
|
|
const start = async (opts?: Parameters<typeof createWorker>[0]) => {
|
|
const ctx = createWorker(opts);
|
|
active = ctx.worker;
|
|
await ctx.worker.init();
|
|
await jest.advanceTimersByTimeAsync(1); // flush the immediate tick + any job it launched
|
|
return ctx;
|
|
};
|
|
|
|
// ── init() wiring ─────────────────────────────────────────────────────────
|
|
|
|
describe("init", () => {
|
|
it("registers the worker in the heartbeat registry on startup", async () => {
|
|
const { mocks } = await start({ queueMode: "worker" });
|
|
expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker");
|
|
});
|
|
|
|
it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => {
|
|
const { mocks } = await start({
|
|
queueMode: "primary",
|
|
queuePrimaryProcesses: false, // isolate reconcile from the polling loops
|
|
mocks: {
|
|
monitorsRepository: { findByIds: jest.fn<any>(), findAll: jest.fn<any>().mockResolvedValue([makeMonitor()]), updateById: jest.fn<any>() },
|
|
},
|
|
});
|
|
|
|
const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
|
|
expect(seededTypes).toContain("check");
|
|
const cleanupTypes = mocks.jobsRepository.upsertCleanupJob.mock.calls.map((c: any[]) => c[0].type);
|
|
expect(cleanupTypes).toContain("cleanup-orphaned");
|
|
expect(cleanupTypes).toContain("cleanup-retention");
|
|
});
|
|
|
|
it("primary mode also seeds a geo-check row for geo-enabled monitors", async () => {
|
|
const geoMonitor = makeMonitor({ type: "http", geoCheckEnabled: true });
|
|
const { mocks } = await start({
|
|
queueMode: "primary",
|
|
queuePrimaryProcesses: false,
|
|
mocks: {
|
|
monitorsRepository: { findByIds: jest.fn<any>(), findAll: jest.fn<any>().mockResolvedValue([geoMonitor]), updateById: jest.fn<any>() },
|
|
},
|
|
});
|
|
|
|
const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
|
|
expect(seededTypes).toContain("geo-check");
|
|
});
|
|
|
|
it("worker mode does not reconcile (never seeds jobs)", async () => {
|
|
const { mocks } = await start({ queueMode: "worker" });
|
|
expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled();
|
|
expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("starts a polling loop per job type, each claiming up to its concurrency cap", async () => {
|
|
const { mocks } = await start({ queueMode: "worker" });
|
|
|
|
const claimedTypes = mocks.jobsRepository.claimDueBatch.mock.calls.map((c: any[]) => c[0]);
|
|
expect(claimedTypes).toEqual(expect.arrayContaining(["check", "geo-check", "evaluate", "cleanup-orphaned", "cleanup-retention"]));
|
|
expect(mocks.jobsRepository.claimDueBatch).toHaveBeenCalledWith("check", 500, expect.any(Number));
|
|
expect(mocks.jobsRepository.claimDueBatch).toHaveBeenCalledWith("evaluate", 20, expect.any(Number));
|
|
expect(mocks.jobsRepository.claimDueBatch).toHaveBeenCalledWith("cleanup-orphaned", 1, expect.any(Number));
|
|
});
|
|
|
|
it("does not start polling loops when primary and queuePrimaryProcesses is false", async () => {
|
|
const { mocks } = await start({ queueMode: "primary", queuePrimaryProcesses: false });
|
|
expect(mocks.jobsRepository.claimDueBatch).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("create() constructs the worker and runs init", async () => {
|
|
const { mocks } = createWorker(); // reuse the mock set; the un-init'd instance is discarded
|
|
const worker = await DBQueueWorker.create(
|
|
mocks.logger as any,
|
|
mocks.jobsRepository as any,
|
|
mocks.monitorsRepository as any,
|
|
mocks.checksRepository as any,
|
|
mocks.checkService as any,
|
|
mocks.checkProducer as any,
|
|
mocks.checkEvaluator as any,
|
|
mocks.geoCheckPipeline as any,
|
|
mocks.dispatcher as any,
|
|
mocks.helper as any,
|
|
mocks.queueWorkersRepository as any,
|
|
"worker",
|
|
true,
|
|
"worker-1"
|
|
);
|
|
active = worker;
|
|
await jest.advanceTimersByTimeAsync(1);
|
|
|
|
expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker");
|
|
});
|
|
});
|
|
|
|
// ── runJob: the per-type execution paths ────────────────────────────────────
|
|
|
|
describe("runJob", () => {
|
|
// Hand the "check" loop exactly one job, then nothing, so the assertion target runs once.
|
|
const claimOnce = (job: Job) => {
|
|
let handed = false;
|
|
return jest.fn<any>(async (type: string) => {
|
|
if (type === job.type && !handed) {
|
|
handed = true;
|
|
return [job];
|
|
}
|
|
return [];
|
|
});
|
|
};
|
|
|
|
it("runs a check job and reschedules it on success", async () => {
|
|
const job = makeJob({ type: "check", refId: "m1", intervalMs: 60000 });
|
|
const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) } } });
|
|
|
|
expect(mocks.monitorsRepository.findByIds).toHaveBeenCalledWith(["m1"]);
|
|
expect(mocks.checkProducer.produce).toHaveBeenCalled();
|
|
expect(mocks.jobsRepository.upsertEvaluate).toHaveBeenCalledWith("m1", expect.any(Number));
|
|
expect(mocks.jobsRepository.recordSuccess).toHaveBeenCalledWith(job.id, job.nextScheduledAt, job.intervalMs, expect.any(Number));
|
|
expect(mocks.jobsRepository.recordFailure).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("records a failure (and not a success) when the job handler throws", async () => {
|
|
const job = makeJob({ type: "check" });
|
|
const failingProducer = { produce: jest.fn<any>().mockRejectedValue(new Error("network down")) };
|
|
const { mocks } = await start({
|
|
mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: failingProducer },
|
|
});
|
|
|
|
expect(mocks.jobsRepository.recordFailure).toHaveBeenCalledWith(job.id, expect.any(Error), expect.any(Number));
|
|
expect(mocks.jobsRepository.recordSuccess).not.toHaveBeenCalled();
|
|
});
|
|
|
|
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 { 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 });
|
|
const checksRepository = { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([{ id: "c1" }]) };
|
|
const { mocks } = await start({
|
|
mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checksRepository },
|
|
});
|
|
|
|
expect(mocks.checkEvaluator.evaluate).toHaveBeenCalled();
|
|
expect(mocks.dispatcher.dispatch).toHaveBeenCalled();
|
|
expect(mocks.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { lastEvaluatedAt: 12345 });
|
|
});
|
|
|
|
it("cleanup-orphaned job runs the helper's cleanup function", async () => {
|
|
const job = makeJob({ id: "cleanup-orphaned", type: "cleanup-orphaned", refId: null, intervalMs: 86400000 });
|
|
const cleanupFn = jest.fn<any>().mockResolvedValue(undefined);
|
|
const helper = {
|
|
getCleanupOrphanedJob: jest.fn<any>().mockReturnValue(cleanupFn),
|
|
getCleanupRetentionJob: jest.fn<any>().mockReturnValue(jest.fn<any>().mockResolvedValue(undefined)),
|
|
};
|
|
await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, helper } });
|
|
|
|
expect(cleanupFn).toHaveBeenCalled();
|
|
});
|
|
|
|
it("cleanup-retention job runs the helper's retention function", async () => {
|
|
const job = makeJob({ id: "cleanup-retention", type: "cleanup-retention", refId: null, intervalMs: 86400000 });
|
|
const retentionFn = jest.fn<any>().mockResolvedValue(undefined);
|
|
const helper = {
|
|
getCleanupOrphanedJob: jest.fn<any>().mockReturnValue(jest.fn<any>().mockResolvedValue(undefined)),
|
|
getCleanupRetentionJob: jest.fn<any>().mockReturnValue(retentionFn),
|
|
};
|
|
await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, helper } });
|
|
|
|
expect(retentionFn).toHaveBeenCalled();
|
|
});
|
|
|
|
it("geo-check job runs the geo pipeline (no evaluate handoff)", async () => {
|
|
const job = makeJob({ id: "geo-check:m1", type: "geo-check", refId: "m1", intervalMs: 300000 });
|
|
const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) } } });
|
|
|
|
expect(mocks.geoCheckPipeline.run).toHaveBeenCalled();
|
|
expect(mocks.dispatcher.dispatch).not.toHaveBeenCalled();
|
|
expect(mocks.jobsRepository.recordSuccess).toHaveBeenCalledWith(job.id, job.nextScheduledAt, job.intervalMs, expect.any(Number));
|
|
});
|
|
|
|
it("renews the lock while a slow job is still running", async () => {
|
|
const job = makeJob({ type: "check" });
|
|
const gate = deferred<{ status: unknown; check: unknown }>();
|
|
const slowProducer = { produce: jest.fn<any>().mockReturnValue(gate.promise) };
|
|
const { mocks } = await start({
|
|
mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: slowProducer },
|
|
});
|
|
|
|
// Job is parked on produce(); push past one renewal interval.
|
|
await jest.advanceTimersByTimeAsync(LOCK_RENEW_MS);
|
|
expect(mocks.jobsRepository.renewLocks).toHaveBeenCalledWith([job.id], expect.any(Number));
|
|
|
|
// Let it finish; success is still recorded.
|
|
gate.resolve({ status: { status: "up" }, check: { id: "c1" } });
|
|
await jest.advanceTimersByTimeAsync(1);
|
|
expect(mocks.jobsRepository.recordSuccess).toHaveBeenCalled();
|
|
});
|
|
|
|
it("stops renewing once the lock is lost to another worker", async () => {
|
|
const job = makeJob({ type: "check" });
|
|
const gate = deferred<{ status: unknown; check: unknown }>();
|
|
const slowProducer = { produce: jest.fn<any>().mockReturnValue(gate.promise) };
|
|
const jobsRepository = {
|
|
...createWorker().mocks.jobsRepository,
|
|
claimDueBatch: claimOnce(job),
|
|
renewLocks: jest.fn<any>().mockResolvedValue(0),
|
|
};
|
|
const { mocks } = await start({ mocks: { jobsRepository, checkProducer: slowProducer } });
|
|
|
|
await jest.advanceTimersByTimeAsync(LOCK_RENEW_MS);
|
|
expect(mocks.jobsRepository.renewLocks).toHaveBeenCalledTimes(1); // lease lost → renewal timer stops
|
|
expect(mocks.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Lost lock") }));
|
|
|
|
// A second interval passes without another renewal attempt.
|
|
await jest.advanceTimersByTimeAsync(LOCK_RENEW_MS);
|
|
expect(mocks.jobsRepository.renewLocks).toHaveBeenCalledTimes(1);
|
|
|
|
gate.resolve({ status: { status: "up" }, check: { id: "c1" } });
|
|
await jest.advanceTimersByTimeAsync(1);
|
|
});
|
|
});
|
|
|
|
// ── monitor lifecycle delegation ────────────────────────────────────────────
|
|
|
|
describe("job lifecycle methods", () => {
|
|
it("addJob seeds a check row, plus a geo row for geo-enabled monitors", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
await worker.addJob("m1", makeMonitor({ geoCheckEnabled: true }));
|
|
|
|
const types = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
|
|
expect(types).toContain("check");
|
|
expect(types).toContain("geo-check");
|
|
});
|
|
|
|
it("addJob seeds only a check row for non-geo monitors", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
await worker.addJob("m1", makeMonitor({ geoCheckEnabled: false }));
|
|
|
|
const types = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
|
|
expect(types).toEqual(["check"]);
|
|
});
|
|
|
|
it("deleteJob removes every row for the monitor", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
await worker.deleteJob(makeMonitor());
|
|
expect(mocks.jobsRepository.deleteById).toHaveBeenCalledWith("m1");
|
|
});
|
|
|
|
it("pauseJob and resumeJob flip the active flag", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
await worker.pauseJob(makeMonitor());
|
|
expect(mocks.jobsRepository.setActiveById).toHaveBeenCalledWith("m1", false);
|
|
await worker.resumeJob(makeMonitor());
|
|
expect(mocks.jobsRepository.setActiveById).toHaveBeenCalledWith("m1", true);
|
|
});
|
|
|
|
it("updateJob reschedules the check row and drops the geo row when geo is disabled", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
await worker.updateJob(makeMonitor({ geoCheckEnabled: false, interval: 30000 }));
|
|
expect(mocks.jobsRepository.updateScheduleById).toHaveBeenCalledWith("m1", "check", 30000);
|
|
expect(mocks.jobsRepository.deleteByIdAndType).toHaveBeenCalledWith("m1", "geo-check");
|
|
});
|
|
|
|
it("updateJob upserts the geo row when geo is enabled", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
await worker.updateJob(makeMonitor({ geoCheckEnabled: true }));
|
|
const types = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
|
|
expect(types).toContain("geo-check");
|
|
expect(mocks.jobsRepository.deleteByIdAndType).not.toHaveBeenCalled();
|
|
});
|
|
});
|
|
|
|
// ── shutdown ────────────────────────────────────────────────────────────────
|
|
|
|
describe("shutdown", () => {
|
|
it("deregisters the worker and stops the polling loops", async () => {
|
|
const { worker, mocks } = createWorker({ queueMode: "worker" });
|
|
await worker.init();
|
|
await jest.advanceTimersByTimeAsync(1);
|
|
const claimsAfterInit = mocks.jobsRepository.claimDueBatch.mock.calls.length;
|
|
|
|
await worker.shutdown();
|
|
expect(mocks.queueWorkersRepository.deleteById).toHaveBeenCalledWith("worker-1");
|
|
|
|
// Loops must not re-arm after shutdown.
|
|
await jest.advanceTimersByTimeAsync(POLL_MS * 4);
|
|
expect(mocks.jobsRepository.claimDueBatch.mock.calls.length).toBe(claimsAfterInit);
|
|
// already shut down; keep afterEach from double-shutting-down
|
|
active = null;
|
|
});
|
|
});
|
|
|
|
// ── flushQueues ───────────────────────────────────────────────────────────────
|
|
|
|
describe("flushQueues", () => {
|
|
it("restarts the worker and reports success", async () => {
|
|
const { worker, mocks } = createWorker({ queueMode: "worker" });
|
|
active = worker;
|
|
await worker.init();
|
|
await jest.advanceTimersByTimeAsync(1);
|
|
|
|
const result = await worker.flushQueues();
|
|
await jest.advanceTimersByTimeAsync(1);
|
|
|
|
expect(result).toEqual({ success: true });
|
|
expect(mocks.queueWorkersRepository.deleteById).toHaveBeenCalledWith("worker-1"); // shutdown leg
|
|
expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledTimes(2); // init + re-init
|
|
});
|
|
});
|
|
|
|
// ── getMetrics ──────────────────────────────────────────────────────────────
|
|
|
|
describe("getMetrics", () => {
|
|
it("aggregates run/fail counts, active leases, and failing jobs across all rows", async () => {
|
|
const now = Date.now();
|
|
const rows = [
|
|
makeJob({ id: "a", runCount: 5, failCount: 0, lockedBy: "w", lockedUntil: now + 10000 }), // active lease
|
|
makeJob({ id: "b", runCount: 2, failCount: 3, lastFailReason: "boom", lockedBy: null, lockedUntil: null }), // failing
|
|
makeJob({ id: "c", runCount: 1, failCount: 0, lockedBy: "w", lockedUntil: now - 10000 }), // expired lease, not active
|
|
];
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
mocks.jobsRepository.findAll.mockResolvedValue(rows);
|
|
mocks.queueWorkersRepository.findRecent.mockResolvedValue([{ workerId: "w" }]);
|
|
|
|
const metrics = await worker.getMetrics();
|
|
|
|
expect(metrics.jobs).toBe(3);
|
|
expect(metrics.totalRuns).toBe(8);
|
|
expect(metrics.totalFailures).toBe(3);
|
|
expect(metrics.activeJobs).toBe(1);
|
|
expect(metrics.failingJobs).toBe(1);
|
|
expect(metrics.jobsWithFailures).toHaveLength(1);
|
|
expect(metrics.jobsWithFailures[0]).toEqual(expect.objectContaining({ failCount: 3, failReason: "boom" }));
|
|
expect(metrics.workers).toEqual([{ workerId: "w" }]);
|
|
});
|
|
});
|
|
|
|
// ── getJobs ─────────────────────────────────────────────────────────────────
|
|
|
|
describe("getJobs", () => {
|
|
it("maps rows to summaries and passes the total count through", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
mocks.jobsRepository.findPage.mockResolvedValue({
|
|
jobs: [makeJob({ id: "check:m1", refId: "m1", type: "check", intervalMs: 60000 })],
|
|
count: 1,
|
|
});
|
|
|
|
const page = await worker.getJobs({ page: 0, rowsPerPage: 10 });
|
|
|
|
expect(page.count).toBe(1);
|
|
expect(page.jobs).toHaveLength(1);
|
|
expect(page.jobs[0]).toEqual(expect.objectContaining({ monitorId: "m1", monitorType: "check", monitorInterval: 60000, repeat: 60000 }));
|
|
});
|
|
|
|
it("falls back to the job id for monitorId when refId is null (global jobs)", async () => {
|
|
const { worker, mocks } = createWorker();
|
|
active = worker;
|
|
mocks.jobsRepository.findPage.mockResolvedValue({
|
|
jobs: [makeJob({ id: "cleanup-orphaned", refId: null, type: "cleanup-orphaned" })],
|
|
count: 1,
|
|
});
|
|
|
|
const page = await worker.getJobs({ page: 0, rowsPerPage: 10 });
|
|
expect(page.jobs[0].monitorId).toBe("cleanup-orphaned");
|
|
});
|
|
});
|
|
});
|