diff --git a/server/openapi.json b/server/openapi.json index d0ae81f26..32f5013a1 100644 --- a/server/openapi.json +++ b/server/openapi.json @@ -13303,8 +13303,7 @@ "items": { "type": "string", "minLength": 1, - "maxLength": 253, - "pattern": "^(https?:\\/\\/\\S+|[A-Za-z0-9.\\-:[\\]]+)$" + "maxLength": 253 }, "maxItems": 10 }, @@ -13317,7 +13316,7 @@ "type": "array", "items": { "type": "string", - "pattern": "^[a-f\\d]{24}$/i" + "pattern": "^[0-9a-fA-F]{24}$" } } } diff --git a/server/src/api/controllers/settingsController.ts b/server/src/api/controllers/settingsController.ts index d02c3886a..cd36467e2 100644 --- a/server/src/api/controllers/settingsController.ts +++ b/server/src/api/controllers/settingsController.ts @@ -6,6 +6,7 @@ import { AppError } from "@/utils/AppError.js"; import { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { IEmailService } from "@/service/emailService.js"; import { IProxiesService } from "@/domain/proxies/proxy.service.js"; +import { IEgressStateService } from "@/domain/egress/egress-state.service.js"; import { Settings } from "@/domain/app-settings/app-settings.type.js"; export interface ISettingsController { @@ -18,10 +19,17 @@ class SettingsController implements ISettingsController { private settingsService: ISettingsService; private emailService: IEmailService; private proxiesService: IProxiesService; - constructor(settingsService: ISettingsService, emailService: IEmailService, proxiesService: IProxiesService) { + private egressStateService: IEgressStateService; + constructor( + settingsService: ISettingsService, + emailService: IEmailService, + proxiesService: IProxiesService, + egressStateService: IEgressStateService + ) { this.settingsService = settingsService; this.emailService = emailService; this.proxiesService = proxiesService; + this.egressStateService = egressStateService; } buildAppSettings = async (dbSettings: Settings) => { @@ -70,7 +78,14 @@ class SettingsController implements ISettingsController { } } + const previousSettings = await this.settingsService.getDBSettings(); const updatedSettings = await this.settingsService.updateDbSettings(validatedBody); + + // Switching the egress check on or off starts from a clean state: no degraded episode, no pending recovery job + if (validatedBody.egressCheckEnabled !== undefined && validatedBody.egressCheckEnabled !== previousSettings.egressCheckEnabled) { + await this.egressStateService.reset(); + } + const returnSettings = await this.buildAppSettings(updatedSettings); return res.status(200).json({ success: true, diff --git a/server/src/api/validation/settingsValidation.ts b/server/src/api/validation/settingsValidation.ts index 33fe433fc..5bf635054 100644 --- a/server/src/api/validation/settingsValidation.ts +++ b/server/src/api/validation/settingsValidation.ts @@ -1,14 +1,16 @@ import { CHECK_TTL_SENTINEL } from "@/domain/checks/check.type.js"; import { MAX_EGRESS_POLL_INTERVAL_SECONDS, MAX_EGRESS_TARGETS, MIN_EGRESS_POLL_INTERVAL_SECONDS } from "@/domain/egress/egress.type.js"; +import { parseEgressTarget } from "@/utils/egressTarget.js"; import { z } from "zod"; // A reliability target is a bare host/IP (ICMP ping), host:port (TCP connect) or an http(s) URL. +// Parsed with the same function the probe uses, so nothing the validator accepts can fail to probe. const egressTargetValidation = z .string() .trim() .min(1) .max(253) - .regex(/^(https?:\/\/\S+|[A-Za-z0-9.\-:[\]]+)$/, "Invalid egress target"); + .refine((target) => parseEgressTarget(target) !== null, "Enter a hostname or IP address, host:port, or http(s) URL"); //**************************************** // Settings Validations @@ -55,7 +57,7 @@ export const updateAppSettingsBodyValidation = z egressCheckEnabled: z.boolean().optional(), egressCheckTargets: z.array(egressTargetValidation).max(MAX_EGRESS_TARGETS).optional(), // empty list falls back to DEFAULT_EGRESS_TARGETS egressPollIntervalSeconds: z.number().int().min(MIN_EGRESS_POLL_INTERVAL_SECONDS).max(MAX_EGRESS_POLL_INTERVAL_SECONDS).optional(), - egressNotifications: z.array(z.string().regex(/^[a-f\d]{24}$/i, "Invalid notification id")).optional(), + egressNotifications: z.array(z.string().regex(/^[0-9a-fA-F]{24}$/, "Invalid notification id")).optional(), }) .strip() .superRefine((body, ctx) => { diff --git a/server/src/config/controllers.ts b/server/src/config/controllers.ts index 7787a389e..ba98030bb 100644 --- a/server/src/config/controllers.ts +++ b/server/src/config/controllers.ts @@ -36,7 +36,12 @@ export const initializeControllers = (apiServices: ApiServices): InitializedCont return { authController: new AuthController(apiServices.userService), monitorController: new MonitorController(apiServices.monitorService, apiServices.notificationsService), - settingsController: new SettingsController(apiServices.settingsService, apiServices.emailService, apiServices.proxiesService), + settingsController: new SettingsController( + apiServices.settingsService, + apiServices.emailService, + apiServices.proxiesService, + apiServices.egressStateService + ), checkController: new CheckController(apiServices.checkService), geoCheckController: new GeoCheckController(apiServices.geoChecksService), inviteController: new InviteController(apiServices.inviteService), diff --git a/server/src/config/services.api.ts b/server/src/config/services.api.ts index e929fbfe8..68b722f0a 100644 --- a/server/src/config/services.api.ts +++ b/server/src/config/services.api.ts @@ -103,7 +103,7 @@ export const buildApi = (shared: SharedServices, jobScheduler: IJobScheduler): A const tagsService = new TagsService(tagsRepository, monitorsRepository); const diagnosticService = new DiagnosticService(db); const proxiesService = new ProxiesService(proxiesRepository, monitorsRepository, settingsService); - const egressStateService = new EgressStateService(egressStateRepository); + const egressStateService = new EgressStateService(egressStateRepository, jobsRepository); return { ...shared, worker: jobScheduler, diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index b97bb56f3..e15e1cbf5 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, jobsRepository, networkService, logger); + const egressService = new EgressService(settingsService, egressStateRepository, jobsRepository, networkService, proxyResolver, logger); // *********************** // Reactors and dispatcher @@ -149,7 +149,8 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig checksRepository, incidentsRepository, geoChecksRepository, - dockerLogsRepository + dockerLogsRepository, + egressService ); const worker = await DBQueueWorker.create({ @@ -165,7 +166,6 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig geoCheckPipeline, dispatcher: reactorDispatcher, helper: workerHelper, - egressService, queueWorkersRepository, queueMode: envSettings.queueMode, queuePrimaryProcesses: envSettings.queuePrimaryProcesses, diff --git a/server/src/domain/egress/egress-state.repository.interface.ts b/server/src/domain/egress/egress-state.repository.interface.ts index 533aad656..c718d4078 100644 --- a/server/src/domain/egress/egress-state.repository.interface.ts +++ b/server/src/domain/egress/egress-state.repository.interface.ts @@ -2,9 +2,10 @@ import type { EgressProbeResult, EgressState } from "@/domain/egress/egress.type export interface IEgressStateRepository { // read - findSingleton(): Promise; // creates the "ok" singleton on first call + findSingleton(): Promise; // creates the "ok" singleton on first call, a plain read after that // update recordProbe(results: EgressProbeResult[], now: Date): Promise; + reset(): Promise; // back to "ok" with no episode; used when the feature is toggled // Atomic transitions. Each returns the updated state when this caller performed the transition, // or null when the state was already on the target side (another worker got there first). markDegraded(results: EgressProbeResult[], now: Date): Promise; diff --git a/server/src/domain/egress/egress-state.repository.mongo.ts b/server/src/domain/egress/egress-state.repository.mongo.ts index 606551e5a..a5bb110c3 100644 --- a/server/src/domain/egress/egress-state.repository.mongo.ts +++ b/server/src/domain/egress/egress-state.repository.mongo.ts @@ -27,6 +27,9 @@ class MongoEgressStateRepository implements IEgressStateRepository { }; findSingleton = async () => { + // Read first: with timestamps on, every findOneAndUpdate would $set updatedAt, so a read would write. + const existing = await EgressStateModel.findOne({ singleton: true }).lean(); + if (existing) return this.toEntity(existing); const doc = await EgressStateModel.findOneAndUpdate( { singleton: true }, { $setOnInsert: { singleton: true, status: "ok" } }, @@ -35,6 +38,15 @@ class MongoEgressStateRepository implements IEgressStateRepository { return this.toEntity(doc); }; + reset = async () => { + const doc = await EgressStateModel.findOneAndUpdate( + { singleton: true }, + { $set: { status: "ok", degradedSince: null, lastProbeAt: null, lastProbeResults: [] }, $setOnInsert: { singleton: true } }, + { upsert: true, new: true, setDefaultsOnInsert: true } + ).lean(); + return this.toEntity(doc); + }; + recordProbe = async (results: EgressProbeResult[], now: Date) => { const doc = await EgressStateModel.findOneAndUpdate( { singleton: true }, diff --git a/server/src/domain/egress/egress-state.service.ts b/server/src/domain/egress/egress-state.service.ts index 35800c7ac..c4701abe9 100644 --- a/server/src/domain/egress/egress-state.service.ts +++ b/server/src/domain/egress/egress-state.service.ts @@ -1,20 +1,31 @@ import { IEgressStateRepository } from "@/domain/egress/egress-state.repository.interface.js"; +import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import type { EgressState } from "@/domain/egress/egress.type.js"; export interface IEgressStateService { getState(): Promise; + reset(): Promise; } const SERVICE_NAME = "EgressStateService"; -// Read-only view of the instance-global egress state for the API process. Transitions are -// owned by the worker (egress.service.ts); this exists so the controller never touches the repository. +// API-process view of the instance-global egress state. Probe-driven transitions are owned by the +// worker (egress.service.ts); this exists so the controller never touches the repositories. export class EgressStateService implements IEgressStateService { static SERVICE_NAME = SERVICE_NAME; - constructor(private egressStateRepository: IEgressStateRepository) {} + constructor( + private egressStateRepository: IEgressStateRepository, + private jobsRepository: IJobsRepository + ) {} getState = async (): Promise => { return this.egressStateRepository.findSingleton(); }; + + // Called when the feature is switched on or off: drop any pending recovery job and start from "ok". + reset = async (): Promise => { + await this.jobsRepository.deleteByIdAndType(null, "egress"); + return this.egressStateRepository.reset(); + }; } diff --git a/server/src/domain/egress/egress.service.ts b/server/src/domain/egress/egress.service.ts index 9ca98f4b2..40a4d4cd2 100644 --- a/server/src/domain/egress/egress.service.ts +++ b/server/src/domain/egress/egress.service.ts @@ -2,50 +2,57 @@ 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 { jobId, type Job, type JobSeed } from "@/domain/jobs/job.type.js"; import type { INetworkService } from "@/service/networkService.js"; +import type { IProxyResolver } from "@/service/network/ProxyResolver.js"; import type { ILogger } from "@/utils/logger.js"; import { DEFAULT_EGRESS_POLL_INTERVAL_SECONDS, DEFAULT_EGRESS_TARGETS, + type EgressAssessment, type EgressProbeResult, type EgressStatus, } from "@/domain/egress/egress.type.js"; import { timeRequest } from "@/service/network/utils.js"; +import { parseEgressTarget, type EgressTarget } from "@/utils/egressTarget.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; +// How long a settled assessment answers for later failures. An assessment still in flight is shared +// for as long as it takes, so this only bounds how stale a reused answer can be. +const ASSESSMENT_REUSE_MS = 5000; // Global job row (refId null) that re-probes while degraded. Inserted when egress degrades, removed on recovery. -const RECOVERY_JOB_ID = jobId("egress", null); +const RECOVERY_JOB_TYPE = "egress" as const; +const RECOVERY_JOB_ID = jobId(RECOVERY_JOB_TYPE, null); // Synthetic identity stamped on the probe monitors so provider responses are recognisable in logs. const PROBE_MONITOR_ID = "egress-probe"; const PROBE_TEAM_ID = "system"; -// "[host]:port" or "host:port" where host has no colons (so a bare IPv6 address is not mistaken for host:port). -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; // Handler for the "egress" job: re-probes while degraded and removes the job once egress is back. - checkRecovery(): Promise; + checkRecovery(job: Job): Promise; } +type CachedAssessment = { + value: Promise; + expiresAt: number | null; // null while the assessment is still in flight +}; + export class EgressService implements IEgressService { static SERVICE_NAME = SERVICE_NAME; - private assessment: { value: Promise; expiresAt: number } | null = null; + private assessment: CachedAssessment | null = null; constructor( private settingsService: ISettingsService, private egressStateRepository: IEgressStateRepository, private jobsRepository: IJobsRepository, private networkService: INetworkService, + private proxyResolver: IProxyResolver, private logger: ILogger ) {} @@ -54,46 +61,52 @@ export class EgressService implements IEgressService { // **************************** // Reuses the existing ping/port/http providers by dressing each target up as a minimal monitor. - private toProbeMonitor = (target: string): Monitor => { + // proxyMode "inherit" lets the resolver apply the global proxy to http probes, as it does for monitors. + private toProbeMonitor = (target: EgressTarget, raw: string): Monitor => { const base = { id: PROBE_MONITOR_ID, teamId: PROBE_TEAM_ID, - name: `Egress probe: ${target}`, + name: `Egress probe: ${raw}`, method: "GET", + proxyMode: "inherit", useAdvancedMatching: false, ignoreTlsErrors: false, customUpCodes: [], }; - if (/^https?:\/\//i.test(target)) { - return { ...base, type: "http", url: target } as unknown as Monitor; + switch (target.kind) { + case "http": + return { ...base, type: "http", url: target.url } as unknown as Monitor; + case "port": + return { ...base, type: "port", url: target.host, port: target.port } as unknown as Monitor; + case "ping": + return { ...base, type: "ping", url: target.host } as unknown as Monitor; } - - const hostPort = HOST_PORT_PATTERN.exec(target); - if (hostPort) { - const host = hostPort[1] ?? hostPort[2]; - return { ...base, type: "port", url: host, port: Number(hostPort[3]) } as unknown as Monitor; - } - - return { ...base, type: "ping", url: target } as unknown as Monitor; }; - private probeTarget = async (target: string): Promise => { - const monitor = this.toProbeMonitor(target); + private probeTarget = async (raw: string): Promise => { + const target = parseEgressTarget(raw); + if (!target) { + // Validation rejects these; a stored value can still be stale, and it must not count as reachable. + return { target: raw, reachable: false, responseTime: 0, message: "Invalid egress target" }; + } + + const monitor = this.toProbeMonitor(target, raw); let timer: NodeJS.Timeout | undefined; - const { response, responseTime, error } = await timeRequest(() => - Promise.race([ - this.networkService.requestStatus(monitor), + const { response, responseTime, error } = await timeRequest(async () => { + const proxyUrl = await this.proxyResolver.resolve(monitor); + return Promise.race([ + this.networkService.requestStatus(monitor, { proxyUrl }), new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`Egress probe timed out after ${PROBE_TIMEOUT_MS}ms`)), PROBE_TIMEOUT_MS); }), - ]).finally(() => clearTimeout(timer)) - ); + ]).finally(() => clearTimeout(timer)); + }); if (error || !response) { return { - target, + target: raw, reachable: false, responseTime, message: error instanceof Error ? error.message : error ? String(error) : "No response", @@ -101,7 +114,7 @@ export class EgressService implements IEgressService { } return { - target, + target: raw, reachable: response.status === true, responseTime: response.responseTime ?? responseTime, message: response.message, @@ -127,32 +140,43 @@ export class EgressService implements IEgressService { // Event-triggered assessment // **************************** - 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; + assessAfterFailure = async (): Promise => { + const assessment = await this.sharedAssessment(); + return assessment === "disabled" ? null : assessment; }; - private assess = async (): Promise => { + // When egress is down every monitor fails at once. Concurrent failures share the in-flight assessment + // however long it takes, and a settled one answers for ASSESSMENT_REUSE_MS. Same shape as ProxyResolver's cache. + private sharedAssessment = (): Promise => { + const cached = this.assessment; + if (cached && (cached.expiresAt === null || cached.expiresAt > Date.now())) { + return cached.value; + } + + const entry: CachedAssessment = { value: this.assess(), expiresAt: null }; + this.assessment = entry; + void entry.value.then((result) => { + if (this.assessment !== entry) return; + if (result === null) { + this.assessment = null; // an internal failure is not cached, so the next failing check retries + } else { + entry.expiresAt = Date.now() + ASSESSMENT_REUSE_MS; + } + }); + return entry.value; + }; + + private assess = async (): Promise => { try { const settings = await this.settingsService.getDBSettings(); if (!settings.egressCheckEnabled) { - return null; + return "disabled"; } const state = await this.egressStateRepository.findSingleton(); if (state.status === "degraded") { - // Recovery is detected by the scheduled job. Make sure it exists in case its row was lost. - await this.scheduleRecoveryCheck(settings.egressPollIntervalSeconds); + // Recovery is detected by the scheduled job. Make sure the row exists without touching its schedule. + await this.scheduleRecoveryCheck(settings.egressPollIntervalSeconds, false); return "degraded"; } @@ -173,7 +197,8 @@ export class EgressService implements IEgressService { method: "assessAfterFailure", details: { results }, }); - await this.scheduleRecoveryCheck(settings.egressPollIntervalSeconds); + // The worker that performed the transition owns the job's schedule; anyone else only makes sure the row exists. + await this.scheduleRecoveryCheck(settings.egressPollIntervalSeconds, degraded !== null); return "degraded"; } catch (error: unknown) { // The egress check must never break check production; treat an internal failure as "unknown" and carry on. @@ -191,35 +216,42 @@ export class EgressService implements IEgressService { // Recovery job // **************************** - // upsertJob only sets nextScheduledAt on insert, so re-arming while degraded never pushes a pending run back. - private scheduleRecoveryCheck = async (pollIntervalSeconds: number | undefined) => { + // reschedule=true sets nextScheduledAt (start of an episode); false only inserts the row if it is missing, + // so re-arming while degraded never pushes a pending run back. + private scheduleRecoveryCheck = async (pollIntervalSeconds: number | undefined, reschedule: boolean) => { const intervalMs = this.toIntervalMs(pollIntervalSeconds); const seed: JobSeed = { id: RECOVERY_JOB_ID, - type: "egress", + type: RECOVERY_JOB_TYPE, refId: null, isActive: true, nextScheduledAt: Date.now() + intervalMs, intervalMs, }; - await this.jobsRepository.upsertJob(seed); + if (reschedule) { + await this.jobsRepository.upsertCleanupJob(seed); // global-row upsert that also sets the schedule + } else { + await this.jobsRepository.upsertJob(seed); + } }; - private removeRecoveryJob = async () => { - await this.jobsRepository.deleteByIdAndType(null, "egress"); + // Removes the row only if nobody has rescheduled it since this run claimed it. Another worker may have seen + // egress fail again in the meantime and started a new episode; its schedule must survive. + private releaseRecoveryJob = async (job: Job) => { + await this.jobsRepository.deleteGlobalJobIfUnchanged(RECOVERY_JOB_TYPE, job.nextScheduledAt); }; // 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 => { + checkRecovery = async (job: Job): Promise => { + const settings = await this.settingsService.getDBSettings(); 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(); + if (!settings.egressCheckEnabled || state.status !== "degraded") { + // Feature switched off, recovery recorded by another worker, or state reset: nothing left to poll for. + await this.releaseRecoveryJob(job); return; } - const settings = await this.settingsService.getDBSettings(); const results = await this.probeTargets(this.resolveTargets(settings.egressCheckTargets)); const now = new Date(); @@ -229,7 +261,7 @@ export class EgressService implements IEgressService { } const recovered = await this.egressStateRepository.markRecovered(results, now); - await this.removeRecoveryJob(); + await this.releaseRecoveryJob(job); if (!recovered) { // Another process performed the transition. return; diff --git a/server/src/domain/egress/egress.type.ts b/server/src/domain/egress/egress.type.ts index 386116ef7..49c8d462f 100644 --- a/server/src/domain/egress/egress.type.ts +++ b/server/src/domain/egress/egress.type.ts @@ -3,6 +3,8 @@ export const EgressStatuses = ["ok", "degraded"] as const; export type EgressStatus = (typeof EgressStatuses)[number]; +// Outcome of one assessment: a status, or "disabled" when the feature is switched off (distinct from an internal failure). +export type EgressAssessment = EgressStatus | "disabled"; // Sensible defaults for a fresh install. Bare hosts are ICMP-pinged, host:port is a TCP connect, // http(s):// URLs are fetched. Any one target being reachable means egress is fine. diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index a352c0888..6fe6b23b9 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -61,6 +61,9 @@ export interface IJobsRepository { // Delete a single job of type. refId null addresses a global row deleteByIdAndType(refId: string | null, type: JobType): Promise; + // Delete a global row only if nobody has rescheduled it since it was claimed (nextScheduledAt unchanged) + deleteGlobalJobIfUnchanged(type: JobType, nextScheduledAt: number): 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 ce4143b2d..d545b0241 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -278,6 +278,11 @@ class MongoJobsRepository implements IJobsRepository { return res.deletedCount > 0; }; + deleteGlobalJobIfUnchanged = async (type: JobType, nextScheduledAt: number) => { + const res = await JobModel.deleteOne({ _id: jobId(type, null), nextScheduledAt }); + return res.deletedCount > 0; + }; + deleteByMonitorIdsNotIn = async (monitorIds: string[]): Promise => { // remove jobs who'se refId is not null (cleanup jobs) and not in the list of monitor IDs (orphaned monitor jobs) const res = await JobModel.deleteMany({ refId: { $ne: null, $nin: monitorIds } }); diff --git a/server/src/domain/notifications/notification.service.ts b/server/src/domain/notifications/notification.service.ts index 61d8d09e3..e270e76d4 100644 --- a/server/src/domain/notifications/notification.service.ts +++ b/server/src/domain/notifications/notification.service.ts @@ -178,7 +178,18 @@ export class NotificationsService implements INotificationsService { deleteById = async (id: string, teamId: string): Promise => { await this.monitorsRepository.removeNotificationFromMonitors(id); + await this.removeFromEgressNotifications(id); const deleted = await this.notificationsRepository.deleteById(id, teamId); return deleted; }; + + // App settings hold notification ids for the egress recovered alert; drop the deleted one so it never dangles + private removeFromEgressNotifications = async (id: string) => { + const settings = await this.settingsService.getDBSettings(); + const current = settings.egressNotifications ?? []; + const remaining = current.filter((notificationId) => notificationId !== id); + if (remaining.length !== current.length) { + await this.settingsService.updateDbSettings({ egressNotifications: remaining }); + } + }; } diff --git a/server/src/service/network/PingProvider.ts b/server/src/service/network/PingProvider.ts index 53f6aad0d..9194ed9d2 100644 --- a/server/src/service/network/PingProvider.ts +++ b/server/src/service/network/PingProvider.ts @@ -4,6 +4,7 @@ import { MonitorType, Monitor } from "@/domain/monitors/monitor.type.js"; import { MonitorStatusResponse } from "@/types/network.js"; import { AppError } from "@/utils/AppError.js"; import ping from "ping"; +import net from "net"; import { timeRequest } from "@/service/network/utils.js"; const SERVICE_NAME = "PingProvider"; @@ -19,10 +20,13 @@ export class PingProvider implements IStatusProvider { } private sanitizeHost(url: string): string { - return url - .replace(/^https?:\/\//, "") - .replace(/\/.*$/, "") - .replace(/:.*/, ""); + const host = url.replace(/^https?:\/\//, "").replace(/\/.*$/, ""); + // "[2001:db8::1]" or "[2001:db8::1]:53": the address is inside the brackets + const bracketed = /^\[([^\]]+)\](?::\d+)?$/.exec(host); + if (bracketed) return bracketed[1]!; + // A bare IPv6 address contains colons that are not a port separator + if (net.isIPv6(host)) return host; + return host.replace(/:.*/, ""); } async handle(monitor: Monitor): Promise> { diff --git a/server/src/utils/egressTarget.ts b/server/src/utils/egressTarget.ts new file mode 100644 index 000000000..240ad6c04 --- /dev/null +++ b/server/src/utils/egressTarget.ts @@ -0,0 +1,48 @@ +import { z } from "zod"; +import { dnsHostnameRegex } from "@/api/validation/shared.js"; + +// A reliability target for the egress self-check, in one of three shapes. Parsed here so the +// settings validator and the probe agree on exactly what is accepted. +export type EgressTarget = + | { kind: "http"; url: string } // http(s):// URL, fetched + | { kind: "port"; host: string; port: number } // host:port or [ipv6]:port, TCP connect + | { kind: "ping"; host: string }; // bare hostname or IP address, ICMP + +// "[ipv6]:port" or "host:port" where host has no colons, so a bare IPv6 address is never read as host:port. +const HOST_PORT_PATTERN = /^(?:\[([^\]]+)\]|([^:/\s[\]]+)):(\d{1,5})$/; +const BRACKETED_PATTERN = /^\[([^\]]+)\]$/; + +const isIpv6 = (value: string): boolean => z.ipv6().safeParse(value).success; +const isHostOrIpv4 = (value: string): boolean => value === "localhost" || z.ipv4().safeParse(value).success || dnsHostnameRegex.test(value); + +const parseHttp = (target: string): EgressTarget | null => { + try { + const { protocol, hostname } = new URL(target); + return (protocol === "http:" || protocol === "https:") && hostname.length > 0 ? { kind: "http", url: target } : null; + } catch { + return null; + } +}; + +export const parseEgressTarget = (raw: string): EgressTarget | null => { + const target = raw.trim(); + if (target.length === 0) return null; + + if (/^https?:\/\//i.test(target)) return parseHttp(target); + + const hostPort = HOST_PORT_PATTERN.exec(target); + if (hostPort) { + const port = Number(hostPort[3]); + if (port < 1 || port > 65535) return null; + const bracketedHost = hostPort[1]; + if (bracketedHost !== undefined) return isIpv6(bracketedHost) ? { kind: "port", host: bracketedHost, port } : null; + const host = hostPort[2]!; + return isHostOrIpv4(host) ? { kind: "port", host, port } : null; + } + + const bracketed = BRACKETED_PATTERN.exec(target); + if (bracketed) return isIpv6(bracketed[1]!) ? { kind: "ping", host: bracketed[1]! } : null; + + if (isIpv6(target) || isHostOrIpv4(target)) return { kind: "ping", host: target }; + return null; +}; diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 49f630b01..c076151ef 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -10,7 +10,6 @@ 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"; @@ -45,7 +44,6 @@ export interface DBQueueWorkerDependencies { geoCheckPipeline: ICheckPipeline; dispatcher: IReactorDispatcher; helper: IWorkerHelper; - egressService: IEgressService; queueWorkersRepository: IQueueWorkersRepository; // forwarded queueMode: QueueMode; // forwarded queuePrimaryProcesses: boolean; // consumed immediately @@ -74,7 +72,6 @@ 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; @@ -95,7 +92,6 @@ 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; } @@ -182,7 +178,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { 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 + await this.helper.getEgressRecoveryJob()(job); // Re-probe while degraded; removes its own row on recovery break; } diff --git a/server/src/worker/worker.helper.ts b/server/src/worker/worker.helper.ts index d8a35e0b2..9e3a6aa8e 100644 --- a/server/src/worker/worker.helper.ts +++ b/server/src/worker/worker.helper.ts @@ -11,10 +11,13 @@ import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; import { ILogger } from "@/utils/logger.js"; import { IDockerLogsRepository } from "@/domain/docker/docker-log.repository.interface.js"; +import { IEgressService } from "@/domain/egress/egress.service.js"; +import type { Job } from "@/domain/jobs/job.type.js"; export interface IWorkerHelper { getCleanupOrphanedJob(): () => Promise; getCleanupRetentionJob(): () => Promise; + getEgressRecoveryJob(): (job: Job) => Promise; } export interface MonitorActionDecision { @@ -45,6 +48,7 @@ export class WorkerHelper implements IWorkerHelper { private incidentsRepository: IIncidentsRepository; private geoChecksRepository: IGeoChecksRepository; private dockerLogsRepository: IDockerLogsRepository; + private egressService: IEgressService; constructor( logger: ILogger, @@ -57,7 +61,8 @@ export class WorkerHelper implements IWorkerHelper { checksRepository: IChecksRepository, incidentsRepository: IIncidentsRepository, geoChecksRepository: IGeoChecksRepository, - dockerLogsRepository: IDockerLogsRepository + dockerLogsRepository: IDockerLogsRepository, + egressService: IEgressService ) { this.logger = logger; this.checkService = checkService; @@ -70,8 +75,17 @@ export class WorkerHelper implements IWorkerHelper { this.incidentsRepository = incidentsRepository; this.geoChecksRepository = geoChecksRepository; this.dockerLogsRepository = dockerLogsRepository; + this.egressService = egressService; } + // Runs while instance egress is degraded; the service removes the row once egress is back. + // Errors propagate so the queue records the failure and retries with its usual backoff. + getEgressRecoveryJob = () => { + return async (job: Job) => { + await this.egressService.checkRecovery(job); + }; + }; + getCleanupOrphanedJob = () => { return async () => { try { diff --git a/server/test/unit/providers/network/pingProvider.test.ts b/server/test/unit/providers/network/pingProvider.test.ts index 04e4ad74e..6b4129850 100644 --- a/server/test/unit/providers/network/pingProvider.test.ts +++ b/server/test/unit/providers/network/pingProvider.test.ts @@ -110,6 +110,24 @@ describe("PingProvider", () => { expect(mockPing.promise.probe).toHaveBeenCalledWith("example.com"); }); + it("keeps a bare IPv6 address intact", async () => { + const mockPing = createMockPing(); + const provider = new PingProvider(mockPing as any); + + await provider.handle(makeMonitor({ url: "2606:4700:4700::1111" })); + + expect(mockPing.promise.probe).toHaveBeenCalledWith("2606:4700:4700::1111"); + }); + + it("unwraps a bracketed IPv6 address and drops its port", async () => { + const mockPing = createMockPing(); + const provider = new PingProvider(mockPing as any); + + await provider.handle(makeMonitor({ url: "[2606:4700:4700::1111]:53" })); + + expect(mockPing.promise.probe).toHaveBeenCalledWith("2606:4700:4700::1111"); + }); + it("throws AppError when url is missing", async () => { const provider = new PingProvider(createMockPing() as any); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 75acaa063..a2d532c16 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -113,8 +113,8 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses const helper = { getCleanupOrphanedJob: jest.fn().mockReturnValue(jest.fn().mockResolvedValue(undefined)), getCleanupRetentionJob: jest.fn().mockReturnValue(jest.fn().mockResolvedValue(undefined)), + getEgressRecoveryJob: 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), @@ -135,7 +135,6 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses geoCheckPipeline, dispatcher, helper, - egressService, queueWorkersRepository, logger, ...overrides?.mocks, @@ -442,11 +441,17 @@ describe("DBQueueWorker", () => { expect(retentionFn).toHaveBeenCalled(); }); - it("egress job runs the egress recovery check", async () => { + it("egress job runs the helper's egress recovery function with the claimed job", async () => { const job = makeJob({ id: "egress", type: "egress", refId: null, intervalMs: 30000 }); - const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) } } }); + const egressFn = jest.fn().mockResolvedValue(undefined); + const helper = { + getCleanupOrphanedJob: jest.fn().mockReturnValue(jest.fn().mockResolvedValue(undefined)), + getCleanupRetentionJob: jest.fn().mockReturnValue(jest.fn().mockResolvedValue(undefined)), + getEgressRecoveryJob: jest.fn().mockReturnValue(egressFn), + }; + const { mocks } = await start({ mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, helper } }); - expect(mocks.egressService.checkRecovery).toHaveBeenCalledTimes(1); + expect(egressFn).toHaveBeenCalledWith(expect.objectContaining({ id: "egress", type: "egress" })); expect(mocks.jobsRepository.recordSuccess).toHaveBeenCalledWith(job.id, job.nextScheduledAt, job.intervalMs, expect.any(Number)); }); diff --git a/server/test/unit/services/egressService.test.ts b/server/test/unit/services/egressService.test.ts index ffe124495..8aee753d7 100644 --- a/server/test/unit/services/egressService.test.ts +++ b/server/test/unit/services/egressService.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, it, jest } from "@jest/globals"; import { EgressService } from "../../../src/domain/egress/egress.service.ts"; import type { EgressState } from "../../../src/domain/egress/egress.type.ts"; +import type { Job } from "../../../src/domain/jobs/job.type.ts"; import { createMockLogger } from "../../helpers/createMockLogger.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── @@ -25,6 +26,23 @@ const makeSettings = (overrides?: Record) => ({ ...overrides, }); +// The claimed "egress" job row as the queue hands it to the handler +const makeJob = (overrides?: Partial): Job => ({ + id: "egress", + type: "egress", + refId: null, + isActive: true, + nextScheduledAt: 1_000, + intervalMs: 30_000, + lockedBy: "worker-1", + lockedUntil: 61_000, + runCount: 0, + failCount: 0, + lastFinishedAt: null, + lastFailReason: null, + ...overrides, +}); + // requestStatus mock that answers per target url const statusFor = (reachableTargets: string[]) => jest.fn().mockImplementation(async (monitor: any) => ({ @@ -37,6 +55,15 @@ const statusFor = (reachableTargets: string[]) => responseTime: 12, })); +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" })), + reset: jest.fn(), + ...overrides, +}); + const createService = (overrides?: Record) => { const defaults = { settingsService: { getDBSettings: jest.fn().mockResolvedValue(makeSettings()) }, @@ -45,12 +72,15 @@ const createService = (overrides?: Record) => { recordProbe: jest.fn().mockResolvedValue(makeState()), 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" })), + reset: jest.fn(), }, jobsRepository: { upsertJob: jest.fn().mockResolvedValue(true), - deleteByIdAndType: jest.fn().mockResolvedValue(true), + upsertCleanupJob: jest.fn().mockResolvedValue(true), + deleteGlobalJobIfUnchanged: jest.fn().mockResolvedValue(true), }, networkService: { requestStatus: statusFor(["1.1.1.1", "8.8.8.8"]) }, + proxyResolver: { resolve: jest.fn().mockResolvedValue(undefined) }, logger: createMockLogger(), ...overrides, }; @@ -59,18 +89,14 @@ const createService = (overrides?: Record) => { defaults.egressStateRepository as any, defaults.jobsRepository as any, defaults.networkService as any, + defaults.proxyResolver as any, defaults.logger as any ); return { service, defaults }; }; -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, -}); +const probedUrls = (defaults: { networkService: { requestStatus: unknown } }) => + (defaults.networkService.requestStatus as jest.Mock).mock.calls.map((call) => (call[0] as any).url); // ── Tests ──────────────────────────────────────────────────────────────────── @@ -85,7 +111,7 @@ describe("EgressService", () => { it("maps a bare host to a ping monitor, host:port to a port monitor and a URL to an http monitor", async () => { const { service, defaults } = createService(); - await service.probeTargets(["1.1.1.1", "8.8.8.8:53", "https://example.com/health", "[2606:4700::1111]:443"]); + await service.probeTargets(["1.1.1.1", "8.8.8.8:53", "https://example.com/health", "[2606:4700::1111]:443", "2606:4700::1111"]); const monitors = (defaults.networkService.requestStatus as jest.Mock).mock.calls.map((call) => call[0] as any); expect(monitors).toEqual([ @@ -93,9 +119,22 @@ describe("EgressService", () => { expect.objectContaining({ type: "port", url: "8.8.8.8", port: 53 }), expect.objectContaining({ type: "http", url: "https://example.com/health", method: "GET", useAdvancedMatching: false }), expect.objectContaining({ type: "port", url: "2606:4700::1111", port: 443 }), + expect.objectContaining({ type: "ping", url: "2606:4700::1111" }), ]); }); + it("passes the resolved proxy to the provider so http probes honour the global proxy", async () => { + const resolve = jest.fn().mockResolvedValue("http://proxy.internal:3128"); + const { service, defaults } = createService({ proxyResolver: { resolve } }); + + await service.probeTargets(["https://example.com/health"]); + + expect(resolve).toHaveBeenCalledWith(expect.objectContaining({ type: "http", proxyMode: "inherit" })); + expect(defaults.networkService.requestStatus).toHaveBeenCalledWith(expect.objectContaining({ type: "http" }), { + proxyUrl: "http://proxy.internal:3128", + }); + }); + it("reports reachable only when the provider says status is true", async () => { const { service } = createService({ networkService: { requestStatus: statusFor(["1.1.1.1"]) } }); @@ -107,6 +146,15 @@ describe("EgressService", () => { ]); }); + it("counts an unparseable stored target as unreachable without probing", async () => { + const { service, defaults } = createService(); + + const results = await service.probeTargets(["[1.1.1.1"]); + + expect(results).toEqual([expect.objectContaining({ target: "[1.1.1.1", reachable: false, message: "Invalid egress target" })]); + expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); + }); + it("counts a thrown provider error as unreachable", async () => { const { service } = createService({ networkService: { requestStatus: jest.fn().mockRejectedValue(new Error("ENETUNREACH")) } }); @@ -154,9 +202,11 @@ describe("EgressService", () => { expect.any(Date) ); expect(defaults.egressStateRepository.markDegraded).not.toHaveBeenCalled(); + expect(defaults.jobsRepository.upsertJob).not.toHaveBeenCalled(); + expect(defaults.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled(); }); - it("marks degraded, warns, inserts the recovery job and returns degraded when every target is unreachable", async () => { + it("marks degraded, warns, schedules the recovery job and returns degraded when every target is unreachable", async () => { const { service, defaults } = createService({ networkService: { requestStatus: statusFor([]) } }); const result = await service.assessAfterFailure(); @@ -168,7 +218,8 @@ describe("EgressService", () => { ); expect(defaults.egressStateRepository.recordProbe).not.toHaveBeenCalled(); expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("egress degraded") })); - expect(defaults.jobsRepository.upsertJob).toHaveBeenCalledWith({ + // The worker that performed the transition sets the schedule + expect(defaults.jobsRepository.upsertCleanupJob).toHaveBeenCalledWith({ id: "egress", type: "egress", refId: null, @@ -176,9 +227,26 @@ describe("EgressService", () => { nextScheduledAt: expect.any(Number), intervalMs: 30_000, }); + expect(defaults.jobsRepository.upsertJob).not.toHaveBeenCalled(); }); - it("returns degraded without probing when the persisted state is already degraded, re-arming the recovery job", async () => { + it("only makes sure the job row exists when another worker performed the transition", async () => { + const { service, defaults } = createService({ + networkService: { requestStatus: statusFor([]) }, + egressStateRepository: degradedRepository({ + findSingleton: jest.fn().mockResolvedValue(makeState()), + markDegraded: jest.fn().mockResolvedValue(null), + }), + }); + + const result = await service.assessAfterFailure(); + + expect(result).toBe("degraded"); + expect(defaults.jobsRepository.upsertJob).toHaveBeenCalledWith(expect.objectContaining({ id: "egress", intervalMs: 30_000 })); + expect(defaults.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled(); + }); + + it("returns degraded without probing when the persisted state is already degraded, without touching the job's schedule", async () => { const { service, defaults } = createService({ egressStateRepository: degradedRepository() }); const result = await service.assessAfterFailure(); @@ -188,6 +256,7 @@ describe("EgressService", () => { expect(defaults.egressStateRepository.markDegraded).not.toHaveBeenCalled(); expect(defaults.egressStateRepository.recordProbe).not.toHaveBeenCalled(); expect(defaults.jobsRepository.upsertJob).toHaveBeenCalledWith(expect.objectContaining({ id: "egress", intervalMs: 30_000 })); + expect(defaults.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled(); }); it("uses the default poll interval for the recovery job when the setting is missing or invalid", async () => { @@ -198,41 +267,7 @@ describe("EgressService", () => { 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); + expect(defaults.jobsRepository.upsertCleanupJob).toHaveBeenCalledWith(expect.objectContaining({ intervalMs: 30_000 })); }); it("falls back to the default targets when the configured list is empty", async () => { @@ -242,8 +277,7 @@ describe("EgressService", () => { await service.assessAfterFailure(); - const urls = (defaults.networkService.requestStatus as jest.Mock).mock.calls.map((call) => (call[0] as any).url); - expect(urls).toEqual(["1.1.1.1", "8.8.8.8"]); + expect(probedUrls(defaults)).toEqual(["1.1.1.1", "8.8.8.8"]); }); it("returns null and logs when the assessment itself fails", async () => { @@ -258,34 +292,97 @@ describe("EgressService", () => { }); }); + // ── shared assessment ───────────────────────────────────────────────────── + + describe("shared assessment", () => { + 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("keeps sharing an in-flight assessment that outlives the reuse window", async () => { + jest.useFakeTimers(); + const { service, defaults } = createService({ networkService: { requestStatus: jest.fn().mockReturnValue(new Promise(() => {})) } }); + + const first = service.assessAfterFailure(); + await jest.advanceTimersByTimeAsync(4_900); // probes still hanging, past the 5 s reuse window from the first call + const second = service.assessAfterFailure(); + await jest.advanceTimersByTimeAsync(200); // probe timeout fires, both settle + + expect(await Promise.all([first, second])).toEqual(["degraded", "degraded"]); + expect(defaults.networkService.requestStatus).toHaveBeenCalledTimes(2); // one round of two targets, not two rounds + }); + + it("reuses a settled assessment within the reuse 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(1_000); + await service.assessAfterFailure(); + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); + + await jest.advanceTimersByTimeAsync(5_000); + await service.assessAfterFailure(); + expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(2); + }); + + it("reuses a disabled result instead of re-reading settings for every failure", async () => { + const getDBSettings = jest.fn().mockResolvedValue(makeSettings({ egressCheckEnabled: false })); + const { service } = createService({ settingsService: { getDBSettings } }); + + expect(await service.assessAfterFailure()).toBeNull(); + expect(await service.assessAfterFailure()).toBeNull(); + + expect(getDBSettings).toHaveBeenCalledTimes(1); + }); + + 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); + }); + }); + // ── checkRecovery (the "egress" job) ────────────────────────────────────── describe("checkRecovery", () => { - it("marks recovered, removes the job and logs when a target becomes reachable", async () => { + it("marks recovered, releases the job and logs when a target becomes reachable", async () => { const { service, defaults } = createService({ egressStateRepository: degradedRepository(), networkService: { requestStatus: statusFor(["1.1.1.1"]) }, }); + const job = makeJob({ nextScheduledAt: 123_456 }); - await service.checkRecovery(); + await service.checkRecovery(job); 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"); + // Conditional on the schedule this run claimed, so a concurrent new episode keeps its job + expect(defaults.jobsRepository.deleteGlobalJobIfUnchanged).toHaveBeenCalledWith("egress", 123_456); expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Instance egress recovered") })); }); - it("removes the job but does not log a recovery when another process already recorded it", async () => { + it("releases the job but does not log a recovery when another process already recorded it", async () => { const { service, defaults } = createService({ egressStateRepository: degradedRepository({ markRecovered: jest.fn().mockResolvedValue(null) }), networkService: { requestStatus: statusFor(["1.1.1.1"]) }, }); - await service.checkRecovery(); + await service.checkRecovery(makeJob()); - expect(defaults.jobsRepository.deleteByIdAndType).toHaveBeenCalledWith(null, "egress"); + expect(defaults.jobsRepository.deleteGlobalJobIfUnchanged).toHaveBeenCalledWith("egress", 1_000); expect(defaults.logger.info).not.toHaveBeenCalledWith( expect.objectContaining({ message: expect.stringContaining("Instance egress recovered") }) ); @@ -297,21 +394,33 @@ describe("EgressService", () => { networkService: { requestStatus: statusFor([]) }, }); - await service.checkRecovery(); + await service.checkRecovery(makeJob()); expect(defaults.egressStateRepository.recordProbe).toHaveBeenCalledTimes(1); expect(defaults.egressStateRepository.markRecovered).not.toHaveBeenCalled(); - expect(defaults.jobsRepository.deleteByIdAndType).not.toHaveBeenCalled(); + expect(defaults.jobsRepository.deleteGlobalJobIfUnchanged).not.toHaveBeenCalled(); }); - it("removes the job without probing when the persisted state is no longer degraded", async () => { + it("releases the job without probing when the persisted state is no longer degraded", async () => { const { service, defaults } = createService(); - await service.checkRecovery(); + await service.checkRecovery(makeJob()); expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); expect(defaults.egressStateRepository.markRecovered).not.toHaveBeenCalled(); - expect(defaults.jobsRepository.deleteByIdAndType).toHaveBeenCalledWith(null, "egress"); + expect(defaults.jobsRepository.deleteGlobalJobIfUnchanged).toHaveBeenCalledWith("egress", 1_000); + }); + + it("releases the job without probing when the feature has been disabled mid-episode", async () => { + const { service, defaults } = createService({ + settingsService: { getDBSettings: jest.fn().mockResolvedValue(makeSettings({ egressCheckEnabled: false })) }, + egressStateRepository: degradedRepository(), + }); + + await service.checkRecovery(makeJob()); + + expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); + expect(defaults.jobsRepository.deleteGlobalJobIfUnchanged).toHaveBeenCalledWith("egress", 1_000); }); it("lets errors propagate so the queue records the failure and retries", async () => { @@ -319,7 +428,7 @@ describe("EgressService", () => { egressStateRepository: degradedRepository({ findSingleton: jest.fn().mockRejectedValue(new Error("db down")) }), }); - await expect(service.checkRecovery()).rejects.toThrow("db down"); + await expect(service.checkRecovery(makeJob())).rejects.toThrow("db down"); }); }); }); diff --git a/server/test/unit/services/egressStateService.test.ts b/server/test/unit/services/egressStateService.test.ts index 85c32f32d..a75c0de1f 100644 --- a/server/test/unit/services/egressStateService.test.ts +++ b/server/test/unit/services/egressStateService.test.ts @@ -9,12 +9,18 @@ const createEgressStateRepo = () => ({ recordProbe: jest.fn(), markDegraded: jest.fn(), markRecovered: jest.fn(), + reset: jest.fn(), +}); + +const createJobsRepo = () => ({ + deleteByIdAndType: jest.fn().mockResolvedValue(true), }); const createService = () => { const egressStateRepository = createEgressStateRepo(); - const service = new EgressStateService(egressStateRepository as any); - return { service, egressStateRepository }; + const jobsRepository = createJobsRepo(); + const service = new EgressStateService(egressStateRepository as any, jobsRepository as any); + return { service, egressStateRepository, jobsRepository }; }; const makeState = (overrides?: Partial): EgressState => ({ @@ -67,4 +73,28 @@ describe("EgressStateService", () => { await expect(service.getState()).rejects.toThrow("db down"); }); }); + + describe("reset", () => { + it("removes any pending recovery job and returns the state to ok", async () => { + const { service, egressStateRepository, jobsRepository } = createService(); + const reset = makeState(); + (egressStateRepository.reset as jest.Mock).mockResolvedValue(reset); + + const result = await service.reset(); + + expect(jobsRepository.deleteByIdAndType).toHaveBeenCalledWith(null, "egress"); + expect(egressStateRepository.reset).toHaveBeenCalledTimes(1); + expect(result).toEqual(reset); + }); + + it("does not touch the probe-driven transitions", async () => { + const { service, egressStateRepository } = createService(); + (egressStateRepository.reset as jest.Mock).mockResolvedValue(makeState()); + + await service.reset(); + + expect(egressStateRepository.markDegraded).not.toHaveBeenCalled(); + expect(egressStateRepository.markRecovered).not.toHaveBeenCalled(); + }); + }); }); diff --git a/server/test/unit/services/notificationsService.test.ts b/server/test/unit/services/notificationsService.test.ts index 8e1633803..f2462866b 100644 --- a/server/test/unit/services/notificationsService.test.ts +++ b/server/test/unit/services/notificationsService.test.ts @@ -26,8 +26,10 @@ const createMonitorsRepo = () => ({ removeNotificationFromMonitors: jest.fn(), }); -const createSettingsService = (clientHost = "https://app.example.com") => ({ +const createSettingsService = (clientHost = "https://app.example.com", egressNotifications: string[] = []) => ({ getSettings: jest.fn().mockReturnValue({ clientHost }), + getDBSettings: jest.fn().mockResolvedValue({ egressNotifications }), + updateDbSettings: jest.fn().mockResolvedValue({ egressNotifications: [] }), }); const createMessageBuilder = () => ({ @@ -371,5 +373,25 @@ describe("NotificationsService", () => { expect(result).toBe(deleted); expect(monitorsRepository.removeNotificationFromMonitors).toHaveBeenCalledWith("notif-1"); }); + + it("removes the notification from the egress recovered recipients in app settings", async () => { + const settingsService = createSettingsService("https://app.example.com", ["notif-0", "notif-1"]); + const { service, notificationsRepository } = createService({ settingsService }); + (notificationsRepository.deleteById as jest.Mock).mockResolvedValue(makeNotification()); + + await service.deleteById("notif-1", "team-1"); + + expect(settingsService.updateDbSettings).toHaveBeenCalledWith({ egressNotifications: ["notif-0"] }); + }); + + it("leaves app settings alone when the notification is not an egress recipient", async () => { + const settingsService = createSettingsService("https://app.example.com", ["notif-0"]); + const { service, notificationsRepository } = createService({ settingsService }); + (notificationsRepository.deleteById as jest.Mock).mockResolvedValue(makeNotification()); + + await service.deleteById("notif-1", "team-1"); + + expect(settingsService.updateDbSettings).not.toHaveBeenCalled(); + }); }); }); diff --git a/server/test/unit/services/workerHelper.test.ts b/server/test/unit/services/workerHelper.test.ts index 749804768..93e224491 100644 --- a/server/test/unit/services/workerHelper.test.ts +++ b/server/test/unit/services/workerHelper.test.ts @@ -41,6 +41,9 @@ const createHelper = (overrides?: Record) => { const checkService = { deleteOlderThan: jest.fn().mockResolvedValue(0), }; + const egressService = { + checkRecovery: jest.fn().mockResolvedValue(undefined), + }; const defaults = { logger: createMockLogger(), @@ -54,6 +57,7 @@ const createHelper = (overrides?: Record) => { incidentsRepository, geoChecksRepository, dockerLogsRepository, + egressService, ...overrides, }; @@ -68,7 +72,8 @@ const createHelper = (overrides?: Record) => { defaults.checksRepository as any, defaults.incidentsRepository as any, defaults.geoChecksRepository as any, - defaults.dockerLogsRepository as any + defaults.dockerLogsRepository as any, + defaults.egressService as any ); return { helper, defaults }; }; @@ -184,4 +189,22 @@ describe("WorkerHelper", () => { expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Unknown error", stack: undefined })); }); }); + + describe("getEgressRecoveryJob", () => { + it("returns a function that hands the claimed job to the egress service", async () => { + const { helper, defaults } = createHelper(); + const job = { id: "egress", type: "egress", refId: null, nextScheduledAt: 1000 } as any; + + await helper.getEgressRecoveryJob()(job); + + expect(defaults.egressService.checkRecovery).toHaveBeenCalledWith(job); + }); + + it("lets the egress service's error propagate so the queue records the failure", async () => { + const { helper, defaults } = createHelper(); + (defaults.egressService.checkRecovery as jest.Mock).mockRejectedValue(new Error("db down")); + + await expect(helper.getEgressRecoveryJob()({ id: "egress" } as any)).rejects.toThrow("db down"); + }); + }); }); diff --git a/server/test/unit/utils/egressTarget.test.ts b/server/test/unit/utils/egressTarget.test.ts new file mode 100644 index 000000000..22c567bae --- /dev/null +++ b/server/test/unit/utils/egressTarget.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from "@jest/globals"; +import { parseEgressTarget } from "../../../src/utils/egressTarget.ts"; + +describe("parseEgressTarget", () => { + it("parses http(s) URLs", () => { + expect(parseEgressTarget("https://example.com/health")).toEqual({ kind: "http", url: "https://example.com/health" }); + expect(parseEgressTarget("HTTP://example.com")).toEqual({ kind: "http", url: "HTTP://example.com" }); + expect(parseEgressTarget("http://[2606:4700::1111]:8080/")).toEqual({ kind: "http", url: "http://[2606:4700::1111]:8080/" }); + }); + + it("parses host:port with a hostname, IPv4 address, localhost or bracketed IPv6 address", () => { + expect(parseEgressTarget("dns.google:53")).toEqual({ kind: "port", host: "dns.google", port: 53 }); + expect(parseEgressTarget("8.8.8.8:53")).toEqual({ kind: "port", host: "8.8.8.8", port: 53 }); + expect(parseEgressTarget("localhost:8080")).toEqual({ kind: "port", host: "localhost", port: 8080 }); + expect(parseEgressTarget("[2606:4700::1111]:443")).toEqual({ kind: "port", host: "2606:4700::1111", port: 443 }); + }); + + it("parses bare hosts and addresses as ping targets", () => { + expect(parseEgressTarget("1.1.1.1")).toEqual({ kind: "ping", host: "1.1.1.1" }); + expect(parseEgressTarget("example.com")).toEqual({ kind: "ping", host: "example.com" }); + expect(parseEgressTarget("2606:4700:4700::1111")).toEqual({ kind: "ping", host: "2606:4700:4700::1111" }); + expect(parseEgressTarget("[2606:4700:4700::1111]")).toEqual({ kind: "ping", host: "2606:4700:4700::1111" }); + expect(parseEgressTarget(" 1.1.1.1 ")).toEqual({ kind: "ping", host: "1.1.1.1" }); + }); + + it("rejects anything the probe could not reach", () => { + for (const target of [ + "", + "ftp://example.com", + "http://", + "bad host", + "1.1.1.1;rm", + "[1.1.1.1", + "1.1.1.1]", + "[::1", + "::1]", + "[[::1]]", + "[not-an-address]:53", + "[1.1.1.1]:53", + "999.1.1.1", + "1.1.1.1:0", + "1.1.1.1:70000", + "1.1.1.1:", + "host-without-tld", + ]) { + expect(parseEgressTarget(target)).toBeNull(); + } + }); +}); diff --git a/server/test/unit/validation/settingsValidation.test.ts b/server/test/unit/validation/settingsValidation.test.ts index 36e25c3b2..745cbd81e 100644 --- a/server/test/unit/validation/settingsValidation.test.ts +++ b/server/test/unit/validation/settingsValidation.test.ts @@ -34,8 +34,30 @@ describe("settingsValidation", () => { expect(() => updateAppSettingsBodyValidation.parse({ egressCheckTargets: targets })).toThrow(); }); + it("accepts IPv6 targets, bare or bracketed with a port", () => { + const targets = ["2606:4700:4700::1111", "[2606:4700:4700::1111]", "[2606:4700:4700::1111]:53", "localhost", "localhost:8080"]; + expect(updateAppSettingsBodyValidation.parse({ egressCheckTargets: targets }).egressCheckTargets).toEqual(targets); + }); + it("rejects targets that are not a host, host:port or http(s) URL", () => { - for (const target of ["", " ", "ftp://example.com", "bad host", "1.1.1.1;rm"]) { + for (const target of ["", " ", "ftp://example.com", "bad host", "1.1.1.1;rm", "http://"]) { + expect(() => updateAppSettingsBodyValidation.parse({ egressCheckTargets: [target] })).toThrow(); + } + }); + + it("rejects malformed addresses that the probe could never reach", () => { + for (const target of [ + "[1.1.1.1", + "1.1.1.1]", + "[::1", + "::1]", + "[[::1]]", + "[not-an-address]:53", + "999.1.1.1", + "1.1.1.1:0", + "1.1.1.1:70000", + "[::1]:", + ]) { expect(() => updateAppSettingsBodyValidation.parse({ egressCheckTargets: [target] })).toThrow(); } }); @@ -59,6 +81,11 @@ describe("settingsValidation", () => { expect(updateAppSettingsBodyValidation.parse({ egressNotifications: [] }).egressNotifications).toEqual([]); }); + it("accepts upper-case hex notification ids", () => { + const id = "64B7F0C2A1D2E3F4A5B6C7D8"; + expect(updateAppSettingsBodyValidation.parse({ egressNotifications: [id] }).egressNotifications).toEqual([id]); + }); + it("rejects notification ids that are not 24-character hex", () => { for (const id of ["not-an-id", "64b7f0c2a1d2e3f4a5b6c7d", "zzzzzzzzzzzzzzzzzzzzzzzz", 123]) { expect(() => updateAppSettingsBodyValidation.parse({ egressNotifications: [id] })).toThrow();