update tests

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();
+		});
+	});
 });

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 = <T>() => {
 	let resolve!: (value: T) => void;
@@ -91,6 +93,14 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 		toStatusResponse: jest.fn<any>().mockReturnValue({ status: "up" }),
 		toLastEvaluatedAt: jest.fn<any>().mockReturnValue(12345),
 	};
+	const bufferService = {
+		addToBuffer: jest.fn<any>(),
+		addGeoCheckToBuffer: jest.fn<any>(),
+		scheduleNextFlush: jest.fn<any>(),
+		flushBuffer: jest.fn<any>().mockResolvedValue(undefined),
+		flushGeoBuffer: jest.fn<any>().mockResolvedValue(undefined),
+		shutdown: jest.fn<any>().mockResolvedValue(undefined),
+	};
 	const checkProducer = {
 		produce: jest.fn<any>().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<any>(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<any>().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<any>().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", () => {