From 7a82c76d5c44da2b94310e2ac687e913ac676e2f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 12:20:21 -0700 Subject: [PATCH] update tests --- .../test/unit/services/bufferService.test.ts | 28 +++++++ .../test/unit/services/dbQueueWorker.test.ts | 80 +++++++++++++++++++ 2 files changed, 108 insertions(+) diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index 7c344c086..d994e5f24 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/server/test/unit/services/bufferService.test.ts @@ -482,4 +482,32 @@ describe("BufferService", () => { ); }); }); + + // ── shutdown ───────────────────────────────────────────────────────────── + + describe("shutdown", () => { + it("stops the flush timer and flushes both buffers", async () => { + const { service, checkService, geoChecksService } = createService(); + service.addToBuffer(makeCheck()); + service.addGeoCheckToBuffer(makeGeoCheck()); + + await service.shutdown(); + + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + expect(geoChecksService.createGeoChecks).toHaveBeenCalledTimes(1); + + // Timer is cleared: advancing past the flush interval triggers no further flush. + await jest.advanceTimersByTimeAsync(60 * 1000); + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + }); + + it("is safe to call with empty buffers", async () => { + const { service, checkService, geoChecksService } = createService(); + + await service.shutdown(); + + expect(checkService.createChecks).not.toHaveBeenCalled(); + expect(geoChecksService.createGeoChecks).not.toHaveBeenCalled(); + }); + }); }); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 66ca6753d..40a208aa2 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -20,6 +20,8 @@ import { createMockLogger } from "../../helpers/createMockLogger.ts"; const POLL_MS = 250; const LOCK_RENEW_MS = LOCK_MS / 3; +const DRAIN_TIMEOUT_MS = 25_000; // mirror of DBQueueWorker DRAIN_TIMEOUT_MS +const DRAIN_POLL_MS = 200; // mirror of DBQueueWorker DRAIN_POLL_MS const deferred = () => { let resolve!: (value: T) => void; @@ -91,6 +93,14 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses toStatusResponse: jest.fn().mockReturnValue({ status: "up" }), toLastEvaluatedAt: jest.fn().mockReturnValue(12345), }; + const bufferService = { + addToBuffer: jest.fn(), + addGeoCheckToBuffer: jest.fn(), + scheduleNextFlush: jest.fn(), + flushBuffer: jest.fn().mockResolvedValue(undefined), + flushGeoBuffer: jest.fn().mockResolvedValue(undefined), + shutdown: jest.fn().mockResolvedValue(undefined), + }; const checkProducer = { produce: jest.fn().mockResolvedValue({ status: { status: "up" }, check: { id: "c1" } }), }; @@ -115,6 +125,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, @@ -131,6 +142,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses mocks.monitorsRepository as any, mocks.checksRepository as any, mocks.checkService as any, + mocks.bufferService as any, mocks.checkProducer as any, mocks.checkEvaluator as any, mocks.geoCheckPipeline as any, @@ -233,6 +245,7 @@ describe("DBQueueWorker", () => { mocks.monitorsRepository as any, mocks.checksRepository as any, mocks.checkService as any, + mocks.bufferService as any, mocks.checkProducer as any, mocks.checkEvaluator as any, mocks.geoCheckPipeline as any, @@ -619,6 +632,73 @@ describe("DBQueueWorker", () => { }); }); + // ── drain ───────────────────────────────────────────────────────────────────── + + describe("drain", () => { + // Hand the "check" loop exactly one job, then nothing. + const claimOnce = (job: Job) => { + let handed = false; + return jest.fn(async (type: string) => { + if (type === job.type && !handed) { + handed = true; + return [job]; + } + return []; + }); + }; + + it("stops the loops and flushes the buffer when nothing is in flight", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + const claimsBefore = mocks.jobsRepository.claimDueBatch.mock.calls.length; + + await worker.drain(); + + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + expect(mocks.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("drained") })); + // Loops must not claim again after draining. + await jest.advanceTimersByTimeAsync(POLL_MS * 4); + expect(mocks.jobsRepository.claimDueBatch.mock.calls.length).toBe(claimsBefore); + }); + + it("waits for an in-flight job to finish before flushing", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const slowProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker, mocks } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: slowProducer }, + }); + + // Job is parked on produce(); drain must not flush while it is still in flight. + const draining = worker.drain(); + await jest.advanceTimersByTimeAsync(DRAIN_POLL_MS * 3); + expect(mocks.bufferService.shutdown).not.toHaveBeenCalled(); + + // Let the job finish; drain observes inFlight == 0, then flushes. + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); + await jest.advanceTimersByTimeAsync(DRAIN_POLL_MS); + await draining; + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + }); + + it("times out after DRAIN_TIMEOUT_MS, warns, and still flushes", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const stuckProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker, mocks } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: stuckProducer }, + }); + + const draining = worker.drain(); + await jest.advanceTimersByTimeAsync(DRAIN_TIMEOUT_MS + DRAIN_POLL_MS); + await draining; + + expect(mocks.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("timed out") })); + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); // release the stuck job for cleanup + }); + }); + // ── flushQueues ─────────────────────────────────────────────────────────────── describe("flushQueues", () => {