From 46eb0b4063b26d20176cd2250b33457dff68eccf Mon Sep 17 00:00:00 2001 From: Craig Lawson Date: Mon, 14 Sep 2026 20:41:58 +0100 Subject: [PATCH] Replace the egress recovery timer with a scheduled job (#3937) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failed check still calls assessAfterFailure, which now shares one in-flight assessment between concurrent failures (same cache shape as ProxyResolver). When egress degrades it upserts a global "egress" job; DBQueueWorker runs checkRecovery for that type at the configured interval, and the job deletes its own row once any target answers. The unref'd timer, startRecoveryPolling, stop and init are gone: a degraded episode survives a restart through the jobs collection. deleteByIdAndType accepts a null refId for global rows. 🔧 --- server/src/config/services.worker.ts | 6 +- server/src/domain/egress/egress.service.ts | 197 +++++-------- .../domain/jobs/job.repository.interface.ts | 4 +- .../src/domain/jobs/job.repository.mongo.ts | 4 +- server/src/domain/jobs/job.type.ts | 4 +- server/src/worker/worker.db-queue.ts | 9 + server/src/worker/worker.job-scheduler.ts | 1 + .../test/unit/services/dbQueueWorker.test.ts | 14 +- .../test/unit/services/egressService.test.ts | 273 +++++++----------- 9 files changed, 203 insertions(+), 309 deletions(-) diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index 83a875f05..b97bb56f3 100644 --- a/server/src/config/services.worker.ts +++ b/server/src/config/services.worker.ts @@ -106,7 +106,7 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig const bufferService = new BufferService(logger, checkService, geoChecksService, dockerLogsService, settingsService, jobsRepository); const statusService = new StatusService(logger, monitorsRepository, monitorStatsRepository); const monitorStatusPolicy = new MonitorStatusPolicy(); - const egressService = new EgressService(settingsService, egressStateRepository, networkService, logger); + const egressService = new EgressService(settingsService, egressStateRepository, jobsRepository, networkService, logger); // *********************** // Reactors and dispatcher @@ -165,6 +165,7 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig geoCheckPipeline, dispatcher: reactorDispatcher, helper: workerHelper, + egressService, queueWorkersRepository, queueMode: envSettings.queueMode, queuePrimaryProcesses: envSettings.queuePrimaryProcesses, @@ -189,8 +190,5 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig }); } - // Resume recovery polling if the instance was mid-episode when it last stopped - await egressService.init(); - return { worker, networkService, bufferService, statusService, egressService }; }; diff --git a/server/src/domain/egress/egress.service.ts b/server/src/domain/egress/egress.service.ts index cb3b79ebb..9ca98f4b2 100644 --- a/server/src/domain/egress/egress.service.ts +++ b/server/src/domain/egress/egress.service.ts @@ -1,6 +1,8 @@ import type { Monitor } from "@/domain/monitors/monitor.type.js"; import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import type { IEgressStateRepository } from "@/domain/egress/egress-state.repository.interface.js"; +import type { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { jobId, type JobSeed } from "@/domain/jobs/job.type.js"; import type { INetworkService } from "@/service/networkService.js"; import type { ILogger } from "@/utils/logger.js"; import { @@ -13,6 +15,11 @@ import { timeRequest } from "@/service/network/utils.js"; const SERVICE_NAME = "EgressService"; const PROBE_TIMEOUT_MS = 5000; +// When egress is down every monitor fails at once, so concurrent failures share one assessment +// rather than each probing. Same in-flight cache shape as ProxyResolver. +const ASSESSMENT_TTL_MS = PROBE_TIMEOUT_MS; +// Global job row (refId null) that re-probes while degraded. Inserted when egress degrades, removed on recovery. +const RECOVERY_JOB_ID = jobId("egress", null); // Synthetic identity stamped on the probe monitors so provider responses are recognisable in logs. const PROBE_MONITOR_ID = "egress-probe"; @@ -23,21 +30,21 @@ const HOST_PORT_PATTERN = /^(?:\[([^\]]+)\]|([^:/\s]+)):(\d{1,5})$/; export interface IEgressService { probeTargets(targets: string[]): Promise; + // Called by the check producer when a check fails. Null when the feature is off or the assessment itself failed. assessAfterFailure(): Promise; - startRecoveryPolling(): void; - stop(): void; - init(): Promise; + // Handler for the "egress" job: re-probes while degraded and removes the job once egress is back. + checkRecovery(): Promise; } export class EgressService implements IEgressService { static SERVICE_NAME = SERVICE_NAME; - private polling = false; - private pollTimer: NodeJS.Timeout | null = null; + private assessment: { value: Promise; expiresAt: number } | null = null; constructor( private settingsService: ISettingsService, private egressStateRepository: IEgressStateRepository, + private jobsRepository: IJobsRepository, private networkService: INetworkService, private logger: ILogger ) {} @@ -111,11 +118,31 @@ export class EgressService implements IEgressService { return targets.length > 0 ? targets : [...DEFAULT_EGRESS_TARGETS]; }; + private toIntervalMs = (seconds: number | undefined): number => { + const valid = typeof seconds === "number" && Number.isFinite(seconds) && seconds > 0; + return (valid ? seconds : DEFAULT_EGRESS_POLL_INTERVAL_SECONDS) * 1000; + }; + // **************************** // Event-triggered assessment // **************************** - assessAfterFailure = async (): Promise => { + assessAfterFailure = (): Promise => { + const now = Date.now(); + if (this.assessment && this.assessment.expiresAt > now) { + return this.assessment.value; + } + // Cache the in-flight promise so concurrent failures share one probe + const value = this.assess(); + this.assessment = { value, expiresAt: now + ASSESSMENT_TTL_MS }; + // Don't cache an internal failure, so the next failing check retries + void value.then((status) => { + if (status === null && this.assessment?.value === value) this.assessment = null; + }); + return value; + }; + + private assess = async (): Promise => { try { const settings = await this.settingsService.getDBSettings(); if (!settings.egressCheckEnabled) { @@ -124,9 +151,8 @@ export class EgressService implements IEgressService { const state = await this.egressStateRepository.findSingleton(); if (state.status === "degraded") { - // Recovery detection belongs to the poll loop. Make sure one is running in this process in case - // the worker that entered the degraded state is no longer around. - this.startRecoveryPolling(); + // Recovery is detected by the scheduled job. Make sure it exists in case its row was lost. + await this.scheduleRecoveryCheck(settings.egressPollIntervalSeconds); return "degraded"; } @@ -147,7 +173,7 @@ export class EgressService implements IEgressService { method: "assessAfterFailure", details: { results }, }); - this.startRecoveryPolling(); + await this.scheduleRecoveryCheck(settings.egressPollIntervalSeconds); return "degraded"; } catch (error: unknown) { // The egress check must never break check production; treat an internal failure as "unknown" and carry on. @@ -162,129 +188,58 @@ export class EgressService implements IEgressService { }; // **************************** - // Recovery poll loop + // Recovery job // **************************** - startRecoveryPolling = () => { - if (this.polling) { - return; - } - this.polling = true; - this.logger.info({ message: "Starting egress recovery polling", service: SERVICE_NAME, method: "startRecoveryPolling" }); - void this.scheduleNextTick(); + // upsertJob only sets nextScheduledAt on insert, so re-arming while degraded never pushes a pending run back. + private scheduleRecoveryCheck = async (pollIntervalSeconds: number | undefined) => { + const intervalMs = this.toIntervalMs(pollIntervalSeconds); + const seed: JobSeed = { + id: RECOVERY_JOB_ID, + type: "egress", + refId: null, + isActive: true, + nextScheduledAt: Date.now() + intervalMs, + intervalMs, + }; + await this.jobsRepository.upsertJob(seed); }; - stop = () => { - this.polling = false; - if (this.pollTimer) { - clearTimeout(this.pollTimer); - this.pollTimer = null; - } + private removeRecoveryJob = async () => { + await this.jobsRepository.deleteByIdAndType(null, "egress"); }; - private readPollIntervalMs = async () => { - try { - const settings = await this.settingsService.getDBSettings(); - const seconds = settings.egressPollIntervalSeconds; - return (Number.isFinite(seconds) && seconds > 0 ? seconds : DEFAULT_EGRESS_POLL_INTERVAL_SECONDS) * 1000; - } catch (error: unknown) { - this.logger.warn({ - message: `Could not read egress poll interval, using default: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - method: "readPollIntervalMs", - }); - return DEFAULT_EGRESS_POLL_INTERVAL_SECONDS * 1000; - } - }; - - private scheduleNextTick = async () => { - if (!this.polling) { - return; - } - const intervalMs = await this.readPollIntervalMs(); - if (!this.polling) { - return; - } - this.pollTimer = setTimeout(() => void this.tick(), intervalMs); - // Never hold the process open for this loop; the state is persisted and init() resumes it on restart. - this.pollTimer.unref(); - }; - - private tick = async () => { - this.pollTimer = null; - if (!this.polling) { + // Runs on the queue at the configured interval while degraded. Errors propagate so the queue records + // the failure and retries with its usual backoff. + checkRecovery = async (): Promise => { + const state = await this.egressStateRepository.findSingleton(); + if (state.status !== "degraded") { + // Another worker recorded the recovery, or the state was reset; nothing left to poll for. + await this.removeRecoveryJob(); return; } - try { - const state = await this.egressStateRepository.findSingleton(); - if (state.status !== "degraded") { - this.logger.info({ message: "Egress no longer degraded, stopping recovery polling", service: SERVICE_NAME, method: "tick" }); - this.stop(); - return; - } + const settings = await this.settingsService.getDBSettings(); + const results = await this.probeTargets(this.resolveTargets(settings.egressCheckTargets)); + const now = new Date(); - const settings = await this.settingsService.getDBSettings(); - const results = await this.probeTargets(this.resolveTargets(settings.egressCheckTargets)); - const now = new Date(); - - if (!results.some((result) => result.reachable)) { - await this.egressStateRepository.recordProbe(results, now); - return; - } - - const recovered = await this.egressStateRepository.markRecovered(results, now); - this.stop(); - if (!recovered) { - // Another process performed the transition. - return; - } - - this.logger.info({ - message: `Instance egress recovered (degraded since ${recovered.degradedSince ?? "unknown"})`, - service: SERVICE_NAME, - method: "tick", - details: { results }, - }); - } catch (error: unknown) { - this.logger.error({ - message: error instanceof Error ? error.message : String(error), - service: SERVICE_NAME, - method: "tick", - stack: error instanceof Error ? error.stack : undefined, - }); - } finally { - // No-op once stop() has been called; otherwise keep the loop alive through errors. - await this.scheduleNextTick(); + if (!results.some((result) => result.reachable)) { + await this.egressStateRepository.recordProbe(results, now); + return; } - }; - // **************************** - // Lifecycle - // **************************** - - init = async () => { - try { - const settings = await this.settingsService.getDBSettings(); - if (!settings.egressCheckEnabled) { - return; - } - const state = await this.egressStateRepository.findSingleton(); - if (state.status === "degraded") { - this.logger.warn({ - message: `Instance egress was degraded at startup (since ${state.degradedSince ?? "unknown"}), resuming recovery polling`, - service: SERVICE_NAME, - method: "init", - }); - this.startRecoveryPolling(); - } - } catch (error: unknown) { - this.logger.error({ - message: error instanceof Error ? error.message : String(error), - service: SERVICE_NAME, - method: "init", - stack: error instanceof Error ? error.stack : undefined, - }); + const recovered = await this.egressStateRepository.markRecovered(results, now); + await this.removeRecoveryJob(); + if (!recovered) { + // Another process performed the transition. + return; } + + this.logger.info({ + message: `Instance egress recovered (degraded since ${recovered.degradedSince ?? "unknown"})`, + service: SERVICE_NAME, + method: "checkRecovery", + details: { results }, + }); }; } diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index d253523fa..a352c0888 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -58,8 +58,8 @@ export interface IJobsRepository { // Delete, drop all rows deleteById(refId: string): Promise; - // Delete a single job of type - deleteByIdAndType(refId: string, type: JobType): Promise; + // Delete a single job of type. refId null addresses a global row + deleteByIdAndType(refId: string | null, type: JobType): Promise; // Delete jobs that reference monitors that no longer exist deleteByMonitorIdsNotIn(monitorIds: string[]): Promise; diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index 60bd4cd77..ce4143b2d 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -273,8 +273,8 @@ class MongoJobsRepository implements IJobsRepository { return res.deletedCount > 0; }; - deleteByIdAndType = async (refId: string, type: JobType) => { - const res = await JobModel.deleteOne({ refId, type }); // Delete a single typed row, e.g. just the geo row + deleteByIdAndType = async (refId: string | null, type: JobType) => { + const res = await JobModel.deleteOne({ refId, type }); // Delete a single typed row, e.g. just the geo row or the global egress row return res.deletedCount > 0; }; diff --git a/server/src/domain/jobs/job.type.ts b/server/src/domain/jobs/job.type.ts index 70d635f81..ae8fbddc3 100644 --- a/server/src/domain/jobs/job.type.ts +++ b/server/src/domain/jobs/job.type.ts @@ -1,10 +1,10 @@ -export const JOB_TYPES = ["check", "geo-check", "evaluate", "cleanup-orphaned", "cleanup-retention"] as const; +export const JOB_TYPES = ["check", "geo-check", "evaluate", "cleanup-orphaned", "cleanup-retention", "egress"] as const; export type JobType = (typeof JOB_TYPES)[number]; export const LOCK_MS = 60_000; export const BACKOFF_MS = 5_000; export const PARKED = Number.MAX_SAFE_INTEGER; -// Canonical _id scheme: monitor-bound rows are `${type}:${refId}`; global rows (refId null) are just the type. +// Canonical _id scheme: monitor-bound rows are `${type}:${refId}`; global rows (refId null: cleanup, egress) are just the type. export const jobId = (type: JobType, refId: string | null): string => (refId === null ? type : `${type}:${refId}`); export type Job = { diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 6f4f31353..49f630b01 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -10,6 +10,7 @@ import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js"; import { ICheckPipeline } from "@/worker/worker.check-pipeline.js"; import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; import { IWorkerHelper } from "@/worker/worker.helper.js"; +import { IEgressService } from "@/domain/egress/egress.service.js"; import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; import { WORKER_STALE_MS } from "@/domain/queue-workers/queue-worker.model.js"; import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; @@ -28,6 +29,7 @@ const CONCURRENCY: Record = { evaluate: 20, "cleanup-orphaned": 1, "cleanup-retention": 1, + egress: 1, }; export interface DBQueueWorkerDependencies { @@ -43,6 +45,7 @@ export interface DBQueueWorkerDependencies { geoCheckPipeline: ICheckPipeline; dispatcher: IReactorDispatcher; helper: IWorkerHelper; + egressService: IEgressService; queueWorkersRepository: IQueueWorkersRepository; // forwarded queueMode: QueueMode; // forwarded queuePrimaryProcesses: boolean; // consumed immediately @@ -61,6 +64,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { evaluate: 0, "cleanup-orphaned": 0, "cleanup-retention": 0, + egress: 0, }; private checksRepository: IChecksRepository; @@ -70,6 +74,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { private dispatcher: IReactorDispatcher; private geoCheckPipeline: ICheckPipeline; private helper: IWorkerHelper; + private egressService: IEgressService; private bufferService: IBufferService; private isDbConnected: () => boolean; @@ -90,6 +95,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { this.dispatcher = dependencies.dispatcher; this.geoCheckPipeline = dependencies.geoCheckPipeline; this.helper = dependencies.helper; + this.egressService = dependencies.egressService; this.bufferService = dependencies.bufferService; this.isDbConnected = dependencies.isDbConnected; } @@ -175,6 +181,9 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { case "cleanup-retention": await this.helper.getCleanupRetentionJob()(); // Get job and execute break; + case "egress": + await this.egressService.checkRecovery(); // Re-probe while degraded; removes its own row on recovery + break; } if (job.intervalMs === null) { diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts index aae8a489b..557a2f8e2 100644 --- a/server/src/worker/worker.job-scheduler.ts +++ b/server/src/worker/worker.job-scheduler.ts @@ -41,6 +41,7 @@ export class JobScheduler implements IJobScheduler { evaluate: POLL_MS, "cleanup-orphaned": POLL_MS, "cleanup-retention": POLL_MS, + egress: POLL_MS, }; // Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd. protected toCheckJob = (monitor: MonitorScheduleFields, now: number, immediate = false): JobSeed => ({ diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 8b13287d4..75acaa063 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -9,7 +9,7 @@ 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 +// sized to free capacity, runs jobs (check / evaluate / geo-check / cleanup / egress) 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. @@ -114,6 +114,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses getCleanupOrphanedJob: jest.fn().mockReturnValue(jest.fn().mockResolvedValue(undefined)), getCleanupRetentionJob: jest.fn().mockReturnValue(jest.fn().mockResolvedValue(undefined)), }; + const egressService = { checkRecovery: jest.fn().mockResolvedValue(undefined) }; const queueWorkersRepository = { upsert: jest.fn().mockResolvedValue(undefined), deleteById: jest.fn().mockResolvedValue(undefined), @@ -134,6 +135,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses geoCheckPipeline, dispatcher, helper, + egressService, queueWorkersRepository, logger, ...overrides?.mocks, @@ -226,7 +228,7 @@ describe("DBQueueWorker", () => { 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(claimedTypes).toEqual(expect.arrayContaining(["check", "geo-check", "evaluate", "cleanup-orphaned", "cleanup-retention", "egress"])); 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)); @@ -440,6 +442,14 @@ describe("DBQueueWorker", () => { expect(retentionFn).toHaveBeenCalled(); }); + it("egress job runs the egress recovery check", async () => { + const job = makeJob({ id: "egress", type: "egress", refId: null, intervalMs: 30000 }); + const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) } } }); + + expect(mocks.egressService.checkRecovery).toHaveBeenCalledTimes(1); + expect(mocks.jobsRepository.recordSuccess).toHaveBeenCalledWith(job.id, job.nextScheduledAt, job.intervalMs, expect.any(Number)); + }); + 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) } } }); diff --git a/server/test/unit/services/egressService.test.ts b/server/test/unit/services/egressService.test.ts index 8c4fbf183..ffe124495 100644 --- a/server/test/unit/services/egressService.test.ts +++ b/server/test/unit/services/egressService.test.ts @@ -46,6 +46,10 @@ const createService = (overrides?: Record) => { markDegraded: jest.fn().mockResolvedValue(makeState({ status: "degraded", degradedSince: "2026-01-01T10:00:00.000Z" })), markRecovered: jest.fn().mockResolvedValue(makeState({ lastRecoveredAt: "2026-01-01T10:05:00.000Z" })), }, + jobsRepository: { + upsertJob: jest.fn().mockResolvedValue(true), + deleteByIdAndType: jest.fn().mockResolvedValue(true), + }, networkService: { requestStatus: statusFor(["1.1.1.1", "8.8.8.8"]) }, logger: createMockLogger(), ...overrides, @@ -53,17 +57,20 @@ const createService = (overrides?: Record) => { const service = new EgressService( defaults.settingsService as any, defaults.egressStateRepository as any, + defaults.jobsRepository as any, defaults.networkService as any, defaults.logger as any ); return { service, defaults }; }; -// Lets the setTimeout chain in startRecoveryPolling settle (settings read, then timer armed) before advancing time. -const runOneTick = async (intervalMs: number) => { - await jest.advanceTimersByTimeAsync(0); - await jest.advanceTimersByTimeAsync(intervalMs); -}; +const degradedRepository = (overrides?: Record) => ({ + findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded", degradedSince: "2026-01-01T10:00:00.000Z" })), + recordProbe: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), + markDegraded: jest.fn().mockResolvedValue(null), + markRecovered: jest.fn().mockResolvedValue(makeState({ lastRecoveredAt: "2026-01-01T10:05:00.000Z", degradedSince: "2026-01-01T10:00:00.000Z" })), + ...overrides, +}); // ── Tests ──────────────────────────────────────────────────────────────────── @@ -149,8 +156,7 @@ describe("EgressService", () => { expect(defaults.egressStateRepository.markDegraded).not.toHaveBeenCalled(); }); - it("marks degraded, warns and returns degraded when every target is unreachable", async () => { - jest.useFakeTimers(); + it("marks degraded, warns, inserts the recovery job and returns degraded when every target is unreachable", async () => { const { service, defaults } = createService({ networkService: { requestStatus: statusFor([]) } }); const result = await service.assessAfterFailure(); @@ -162,20 +168,18 @@ describe("EgressService", () => { ); expect(defaults.egressStateRepository.recordProbe).not.toHaveBeenCalled(); expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("egress degraded") })); - expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Starting egress recovery polling" })); - service.stop(); + expect(defaults.jobsRepository.upsertJob).toHaveBeenCalledWith({ + id: "egress", + type: "egress", + refId: null, + isActive: true, + nextScheduledAt: expect.any(Number), + intervalMs: 30_000, + }); }); - it("returns degraded without probing when the persisted state is already degraded", async () => { - jest.useFakeTimers(); - const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn(), - markDegraded: jest.fn(), - markRecovered: jest.fn(), - }, - }); + it("returns degraded without probing when the persisted state is already degraded, re-arming the recovery job", async () => { + const { service, defaults } = createService({ egressStateRepository: degradedRepository() }); const result = await service.assessAfterFailure(); @@ -183,7 +187,52 @@ describe("EgressService", () => { expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); expect(defaults.egressStateRepository.markDegraded).not.toHaveBeenCalled(); expect(defaults.egressStateRepository.recordProbe).not.toHaveBeenCalled(); - service.stop(); + expect(defaults.jobsRepository.upsertJob).toHaveBeenCalledWith(expect.objectContaining({ id: "egress", intervalMs: 30_000 })); + }); + + it("uses the default poll interval for the recovery job when the setting is missing or invalid", async () => { + const { service, defaults } = createService({ + settingsService: { getDBSettings: jest.fn().mockResolvedValue(makeSettings({ egressPollIntervalSeconds: undefined })) }, + networkService: { requestStatus: statusFor([]) }, + }); + + await service.assessAfterFailure(); + + expect(defaults.jobsRepository.upsertJob).toHaveBeenCalledWith(expect.objectContaining({ intervalMs: 30_000 })); + }); + + it("shares one probe between concurrent failures", async () => { + const { service, defaults } = createService({ networkService: { requestStatus: statusFor(["8.8.8.8"]) } }); + + const results = await Promise.all([service.assessAfterFailure(), service.assessAfterFailure(), service.assessAfterFailure()]); + + expect(results).toEqual(["ok", "ok", "ok"]); + expect(defaults.settingsService.getDBSettings).toHaveBeenCalledTimes(1); + expect(defaults.networkService.requestStatus).toHaveBeenCalledTimes(2); // one call per target, once + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); + }); + + it("reuses a settled assessment within the cache window and probes again after it", async () => { + jest.useFakeTimers(); + const { service, defaults } = createService({ networkService: { requestStatus: statusFor(["8.8.8.8"]) } }); + + await service.assessAfterFailure(); + await jest.advanceTimersByTimeAsync(1000); + await service.assessAfterFailure(); + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); + + await jest.advanceTimersByTimeAsync(5000); + await service.assessAfterFailure(); + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(2); + }); + + it("does not cache an internal failure", async () => { + const getDBSettings = jest.fn().mockRejectedValueOnce(new Error("db down")).mockResolvedValue(makeSettings()); + const { service, defaults } = createService({ settingsService: { getDBSettings } }); + + expect(await service.assessAfterFailure()).toBeNull(); + expect(await service.assessAfterFailure()).toBe("ok"); + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); }); it("falls back to the default targets when the configured list is empty", async () => { @@ -209,196 +258,68 @@ describe("EgressService", () => { }); }); - // ── recovery polling ────────────────────────────────────────────────────── + // ── checkRecovery (the "egress" job) ────────────────────────────────────── - describe("recovery polling", () => { - it("marks recovered once and stops polling when a target becomes reachable", async () => { - jest.useFakeTimers(); - const recovered = makeState({ lastRecoveredAt: "2026-01-01T10:05:00.000Z", degradedSince: "2026-01-01T10:00:00.000Z" }); + describe("checkRecovery", () => { + it("marks recovered, removes the job and logs when a target becomes reachable", async () => { const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn(), - markDegraded: jest.fn(), - markRecovered: jest.fn().mockResolvedValue(recovered), - }, + egressStateRepository: degradedRepository(), networkService: { requestStatus: statusFor(["1.1.1.1"]) }, }); - service.startRecoveryPolling(); - await runOneTick(30_000); + await service.checkRecovery(); - expect(defaults.egressStateRepository.markRecovered).toHaveBeenCalledTimes(1); + expect(defaults.egressStateRepository.markRecovered).toHaveBeenCalledWith( + [expect.objectContaining({ target: "1.1.1.1", reachable: true }), expect.objectContaining({ target: "8.8.8.8", reachable: false })], + expect.any(Date) + ); + expect(defaults.jobsRepository.deleteByIdAndType).toHaveBeenCalledWith(null, "egress"); expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Instance egress recovered") })); - - // Loop has stopped: a further interval produces no more work - await runOneTick(30_000); - expect(defaults.egressStateRepository.markRecovered).toHaveBeenCalledTimes(1); }); - it("does not log a recovery when another process already recorded it", async () => { - jest.useFakeTimers(); + it("removes the job but does not log a recovery when another process already recorded it", async () => { const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn(), - markDegraded: jest.fn(), - markRecovered: jest.fn().mockResolvedValue(null), - }, + egressStateRepository: degradedRepository({ markRecovered: jest.fn().mockResolvedValue(null) }), networkService: { requestStatus: statusFor(["1.1.1.1"]) }, }); - service.startRecoveryPolling(); - await runOneTick(30_000); + await service.checkRecovery(); - expect(defaults.egressStateRepository.markRecovered).toHaveBeenCalledTimes(1); + expect(defaults.jobsRepository.deleteByIdAndType).toHaveBeenCalledWith(null, "egress"); expect(defaults.logger.info).not.toHaveBeenCalledWith( expect.objectContaining({ message: expect.stringContaining("Instance egress recovered") }) ); }); - it("records the probe and keeps polling while every target stays unreachable", async () => { - jest.useFakeTimers(); + it("records the probe and keeps the job while every target stays unreachable", async () => { const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - markDegraded: jest.fn(), - markRecovered: jest.fn(), - }, + egressStateRepository: degradedRepository(), networkService: { requestStatus: statusFor([]) }, }); - service.startRecoveryPolling(); - await runOneTick(30_000); - await runOneTick(30_000); + await service.checkRecovery(); - expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(2); + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); expect(defaults.egressStateRepository.markRecovered).not.toHaveBeenCalled(); - service.stop(); + expect(defaults.jobsRepository.deleteByIdAndType).not.toHaveBeenCalled(); }); - it("stops polling when the persisted state is no longer degraded", async () => { - jest.useFakeTimers(); - const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "ok" })), - recordProbe: jest.fn(), - markDegraded: jest.fn(), - markRecovered: jest.fn(), - }, - }); - - service.startRecoveryPolling(); - await runOneTick(30_000); - - expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); - await runOneTick(30_000); - expect(defaults.egressStateRepository.findSingleton).toHaveBeenCalledTimes(1); - }); - - it("survives a failing tick and reschedules", async () => { - jest.useFakeTimers(); - const findSingleton = jest - .fn() - .mockRejectedValueOnce(new Error("db blip")) - .mockResolvedValue(makeState({ status: "degraded" })); - const { service, defaults } = createService({ - egressStateRepository: { - findSingleton, - recordProbe: jest.fn().mockResolvedValue(makeState()), - markDegraded: jest.fn(), - markRecovered: jest.fn(), - }, - networkService: { requestStatus: statusFor([]) }, - }); - - service.startRecoveryPolling(); - await runOneTick(30_000); - expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "db blip", method: "tick" })); - - await runOneTick(30_000); - expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); - service.stop(); - }); - - it("re-reads the poll interval from settings on each tick", async () => { - jest.useFakeTimers(); - const { service, defaults } = createService({ - settingsService: { getDBSettings: jest.fn().mockResolvedValue(makeSettings({ egressPollIntervalSeconds: 5 })) }, - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - markDegraded: jest.fn(), - markRecovered: jest.fn(), - }, - networkService: { requestStatus: statusFor([]) }, - }); - - service.startRecoveryPolling(); - await runOneTick(5_000); - - expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); - service.stop(); - }); - - it("is idempotent: a second start does not arm a second loop", async () => { - jest.useFakeTimers(); - const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - markDegraded: jest.fn(), - markRecovered: jest.fn(), - }, - networkService: { requestStatus: statusFor([]) }, - }); - - service.startRecoveryPolling(); - service.startRecoveryPolling(); - await runOneTick(30_000); - - expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); - service.stop(); - }); - }); - - // ── init ────────────────────────────────────────────────────────────────── - - describe("init", () => { - it("resumes polling when the feature is enabled and the persisted state is degraded", async () => { - jest.useFakeTimers(); - const { service, defaults } = createService({ - egressStateRepository: { - findSingleton: jest.fn().mockResolvedValue(makeState({ status: "degraded" })), - recordProbe: jest.fn(), - markDegraded: jest.fn(), - markRecovered: jest.fn().mockResolvedValue(makeState()), - }, - }); - - await service.init(); - - expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Starting egress recovery polling" })); - service.stop(); - }); - - it("does nothing when the persisted state is ok", async () => { + it("removes the job without probing when the persisted state is no longer degraded", async () => { const { service, defaults } = createService(); - await service.init(); + await service.checkRecovery(); - expect(defaults.logger.info).not.toHaveBeenCalled(); + expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); + expect(defaults.egressStateRepository.markRecovered).not.toHaveBeenCalled(); + expect(defaults.jobsRepository.deleteByIdAndType).toHaveBeenCalledWith(null, "egress"); }); - it("does nothing when the feature is disabled", async () => { - const { service, defaults } = createService({ - settingsService: { getDBSettings: jest.fn().mockResolvedValue(makeSettings({ egressCheckEnabled: false })) }, + it("lets errors propagate so the queue records the failure and retries", async () => { + const { service } = createService({ + egressStateRepository: degradedRepository({ findSingleton: jest.fn().mockRejectedValue(new Error("db down")) }), }); - await service.init(); - - expect(defaults.egressStateRepository.findSingleton).not.toHaveBeenCalled(); + await expect(service.checkRecovery()).rejects.toThrow("db down"); }); }); });