From 33b15d6cbacdc95656da696f50c29323d704a319 Mon Sep 17 00:00:00 2001 From: Anso Date: Thu, 7 May 2026 13:16:24 -0400 Subject: [PATCH] feat(fleet-sync): retry failed pushes and backfill on add-node (#970) A control instance now retries fleet-sync pushes that hit a transient failure and backfills the security state on a freshly registered remote without waiting for the next policy edit. New service: - FleetSyncRetryService (singleton, start/stop) wakes 30s after boot and ticks every 5min. For each fleet resource, queries getFailedSyncTargets within a 24h window and re-pushes via FleetSyncService.pushResourceToNode through the same per-node mutex, so a normal fanout in flight serializes naturally with a retry. - After STALE_THRESHOLD_MS (1h) of continuous failure for a previously-working node, dispatches a single warning notification per cooldown window. Brand-new nodes that have never succeeded do not alert via this path; misconfigured remotes are caught by the test-connection affordance at registration time. - Wired into bootstrap startup/shutdown next to AutoHealService. Public surface: - FleetSyncService.pushResourceToNode(node, resource): targeted push to one node that re-uses the per-node mutex. Used by the retry service and any future targeted-resync flow. - routes/nodes.ts POST /api/nodes fires pushResourceAsync for both resources after a remote-proxy node row commits. Tuning constants centralized in fleetSyncConstants.ts: - RETRY_MAX_AGE_MS = 24h - STALE_THRESHOLD_MS = 1h Tests: - 8 vitest cases covering replica skip, retry dispatch, missing-node skip, alert-once-per-cooldown across the threshold window, no-alert for recent failures, no-alert for brand-new never-succeeded nodes, no-alert when the retry itself succeeds, start/stop idempotency. - Full backend suite: 1781 pass / 5 skipped. --- .../fleet-sync-retry-service.test.ts | 179 ++++++++++++++++++ backend/src/bootstrap/shutdown.ts | 2 + backend/src/bootstrap/startup.ts | 2 + backend/src/routes/nodes.ts | 10 + backend/src/services/FleetSyncRetryService.ts | 162 ++++++++++++++++ backend/src/services/FleetSyncService.ts | 24 +++ backend/src/services/fleetSyncConstants.ts | 10 + 7 files changed, 389 insertions(+) create mode 100644 backend/src/__tests__/fleet-sync-retry-service.test.ts create mode 100644 backend/src/services/FleetSyncRetryService.ts diff --git a/backend/src/__tests__/fleet-sync-retry-service.test.ts b/backend/src/__tests__/fleet-sync-retry-service.test.ts new file mode 100644 index 00000000..b86c06c3 --- /dev/null +++ b/backend/src/__tests__/fleet-sync-retry-service.test.ts @@ -0,0 +1,179 @@ +/** + * Unit tests for FleetSyncRetryService: the background loop that re-pushes + * fleet sync to nodes whose last attempt failed and emits a stale-target + * notification when a per-node failure window exceeds the threshold. + */ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +const { + mockGetFailedSyncTargets, + mockGetNode, + mockGetFleetSyncStatuses, + mockGetSystemState, + mockPushResourceToNode, + mockDispatchAlert, +} = vi.hoisted(() => ({ + mockGetFailedSyncTargets: vi.fn().mockReturnValue([]), + mockGetNode: vi.fn(), + mockGetFleetSyncStatuses: vi.fn().mockReturnValue([]), + mockGetSystemState: vi.fn().mockReturnValue(null), + mockPushResourceToNode: vi.fn().mockResolvedValue(undefined), + mockDispatchAlert: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock('../services/DatabaseService', () => ({ + DatabaseService: { + getInstance: () => ({ + getFailedSyncTargets: mockGetFailedSyncTargets, + getNode: mockGetNode, + getFleetSyncStatuses: mockGetFleetSyncStatuses, + getSystemState: mockGetSystemState, + }), + }, +})); + +vi.mock('../services/FleetSyncService', () => ({ + FleetSyncService: { + getInstance: () => ({ pushResourceToNode: mockPushResourceToNode }), + getRole: () => (mockGetSystemState('fleet_role') === 'replica' ? 'replica' : 'control'), + }, + FLEET_RESOURCES: ['scan_policies', 'cve_suppressions'] as const, +})); + +vi.mock('../services/NotificationService', () => ({ + NotificationService: { + getInstance: () => ({ dispatchAlert: mockDispatchAlert }), + }, +})); + +vi.mock('../utils/debug', () => ({ + isDebugEnabled: () => false, +})); + +import { FleetSyncRetryService } from '../services/FleetSyncRetryService'; + +beforeEach(() => { + vi.clearAllMocks(); + mockGetSystemState.mockReturnValue(null); + // Reset internal alert memo by replacing the singleton. + (FleetSyncRetryService as unknown as { instance: FleetSyncRetryService | undefined }).instance = undefined; +}); + +describe('FleetSyncRetryService.evaluate', () => { + it('skips the tick when this instance is a replica', async () => { + mockGetSystemState.mockImplementation((k: string) => (k === 'fleet_role' ? 'replica' : null)); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockGetFailedSyncTargets).not.toHaveBeenCalled(); + expect(mockPushResourceToNode).not.toHaveBeenCalled(); + }); + + it('retries each failed (node, resource) target by calling pushResourceToNode', async () => { + mockGetFailedSyncTargets.mockImplementation((resource: string) => { + if (resource === 'scan_policies') { + return [{ node_id: 7, resource, last_success_at: null, last_failure_at: Date.now() - 1000, last_error: 'timeout' }]; + } + return []; + }); + mockGetNode.mockReturnValue({ id: 7, name: 'NodeSeven', type: 'remote', api_url: 'https://n7.example', api_token: 'tok' }); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockPushResourceToNode).toHaveBeenCalledTimes(1); + const call = mockPushResourceToNode.mock.calls[0]; + expect(call[0].id).toBe(7); + expect(call[1]).toBe('scan_policies'); + }); + + it('does not retry when the node has been deleted from the registry', async () => { + mockGetFailedSyncTargets.mockImplementation((resource: string) => { + if (resource === 'cve_suppressions') { + return [{ node_id: 99, resource, last_success_at: null, last_failure_at: Date.now() - 1000, last_error: 'gone' }]; + } + return []; + }); + mockGetNode.mockReturnValue(undefined); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockPushResourceToNode).not.toHaveBeenCalled(); + }); + + it('emits a stale-target notification once when previously-working node has been failing for over an hour', async () => { + const twoHoursAgo = Date.now() - 2 * 60 * 60_000; + const fourHoursAgo = Date.now() - 4 * 60 * 60_000; + mockGetFailedSyncTargets.mockImplementation((resource: string) => { + if (resource === 'scan_policies') { + return [{ node_id: 8, resource, last_success_at: fourHoursAgo, last_failure_at: twoHoursAgo, last_error: 'unreachable' }]; + } + return []; + }); + mockGetNode.mockReturnValue({ id: 8, name: 'NodeEight', type: 'remote', api_url: 'https://n8.example', api_token: 'tok' }); + mockGetFleetSyncStatuses.mockReturnValue([ + { node_id: 8, resource: 'scan_policies', last_success_at: fourHoursAgo, last_failure_at: twoHoursAgo, last_error: 'unreachable' }, + ]); + + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockDispatchAlert).toHaveBeenCalledTimes(1); + expect(mockDispatchAlert).toHaveBeenCalledWith( + 'warning', + 'system', + expect.stringContaining('NodeEight'), + ); + + // Second tick within the cooldown should not re-alert. + mockDispatchAlert.mockClear(); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockDispatchAlert).not.toHaveBeenCalled(); + }); + + it('does not alert when failure window is still within the threshold', async () => { + const tenMinutesAgo = Date.now() - 10 * 60_000; + const twentyMinutesAgo = Date.now() - 20 * 60_000; + mockGetFailedSyncTargets.mockReturnValue([ + { node_id: 9, resource: 'scan_policies', last_success_at: twentyMinutesAgo, last_failure_at: tenMinutesAgo, last_error: 'flap' }, + ]); + mockGetNode.mockReturnValue({ id: 9, name: 'NodeNine', type: 'remote', api_url: 'https://n9.example', api_token: 'tok' }); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockDispatchAlert).not.toHaveBeenCalled(); + }); + + it('does not alert when the retry itself succeeds (last_failure_at cleared to NULL)', async () => { + const fourHoursAgo = Date.now() - 4 * 60 * 60_000; + const twoHoursAgo = Date.now() - 2 * 60 * 60_000; + // Pre-retry: stale failure exceeds threshold. + mockGetFailedSyncTargets.mockImplementation((resource: string) => { + if (resource === 'scan_policies') { + return [{ node_id: 11, resource, last_success_at: fourHoursAgo, last_failure_at: twoHoursAgo, last_error: 'flap' }]; + } + return []; + }); + mockGetNode.mockReturnValue({ id: 11, name: 'NodeEleven', type: 'remote', api_url: 'https://n11.example', api_token: 'tok' }); + // Post-retry: the push succeeded, so the row now has last_success_at set + // and last_failure_at cleared to NULL. The alert path must NOT fire. + const justNow = Date.now(); + mockGetFleetSyncStatuses.mockReturnValue([ + { node_id: 11, resource: 'scan_policies', last_success_at: justNow, last_failure_at: null, last_error: null }, + ]); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockDispatchAlert).not.toHaveBeenCalled(); + }); + + it('does not alert for a brand-new node that has never succeeded', async () => { + const fiveMinutesAgo = Date.now() - 5 * 60_000; + mockGetFailedSyncTargets.mockReturnValue([ + { node_id: 10, resource: 'scan_policies', last_success_at: null, last_failure_at: fiveMinutesAgo, last_error: 'cold start' }, + ]); + mockGetNode.mockReturnValue({ id: 10, name: 'NodeTen', type: 'remote', api_url: 'https://n10.example', api_token: 'tok' }); + await FleetSyncRetryService.getInstance().evaluate(); + expect(mockDispatchAlert).not.toHaveBeenCalled(); + }); +}); + +describe('FleetSyncRetryService start/stop', () => { + it('start() and stop() are idempotent and clear timers', () => { + const svc = FleetSyncRetryService.getInstance(); + svc.start(); + svc.stop(); + svc.start(); + svc.stop(); + // No assertions on internal state; the absence of stray timers is enough + // for vitest to exit cleanly. This also exercises the early-clear path. + expect(true).toBe(true); + }); +}); diff --git a/backend/src/bootstrap/shutdown.ts b/backend/src/bootstrap/shutdown.ts index 74254b40..92d03c1c 100644 --- a/backend/src/bootstrap/shutdown.ts +++ b/backend/src/bootstrap/shutdown.ts @@ -3,6 +3,7 @@ import { DatabaseService } from '../services/DatabaseService'; import { LicenseService } from '../services/LicenseService'; import { MonitorService } from '../services/MonitorService'; import { AutoHealService } from '../services/AutoHealService'; +import { FleetSyncRetryService } from '../services/FleetSyncRetryService'; import { DockerEventManager } from '../services/DockerEventManager'; import { ImageUpdateService } from '../services/ImageUpdateService'; import { SchedulerService } from '../services/SchedulerService'; @@ -29,6 +30,7 @@ export function installShutdownHandlers(server: Server): void { console.warn('[Shutdown] MonitorService cleanup failed:', (e as Error).message); } try { AutoHealService.getInstance().stop(); } catch (e) { console.warn('[Shutdown] AutoHealService cleanup failed:', (e as Error).message); } + try { FleetSyncRetryService.getInstance().stop(); } catch (e) { console.warn('[Shutdown] FleetSyncRetryService cleanup failed:', (e as Error).message); } try { DockerEventManager.getInstance().stop(); } catch (e) { console.warn('[Shutdown] DockerEventManager cleanup failed:', (e as Error).message); } diff --git a/backend/src/bootstrap/startup.ts b/backend/src/bootstrap/startup.ts index 81d13d44..bd1d2edb 100644 --- a/backend/src/bootstrap/startup.ts +++ b/backend/src/bootstrap/startup.ts @@ -5,6 +5,7 @@ import { LicenseService } from '../services/LicenseService'; import SelfUpdateService from '../services/SelfUpdateService'; import { MonitorService } from '../services/MonitorService'; import { AutoHealService } from '../services/AutoHealService'; +import { FleetSyncRetryService } from '../services/FleetSyncRetryService'; import { DockerEventManager } from '../services/DockerEventManager'; import TrivyService from '../services/TrivyService'; import { ImageUpdateService } from '../services/ImageUpdateService'; @@ -39,6 +40,7 @@ export async function startServer(server: Server): Promise { // safely run alongside the async initializers below. MonitorService.getInstance().start(); AutoHealService.getInstance().start(); + FleetSyncRetryService.getInstance().start(); ImageUpdateService.getInstance().start(); SchedulerService.getInstance().start(); MfaService.getInstance().start(); diff --git a/backend/src/routes/nodes.ts b/backend/src/routes/nodes.ts index 8a1863ef..cc31ba5c 100644 --- a/backend/src/routes/nodes.ts +++ b/backend/src/routes/nodes.ts @@ -12,6 +12,7 @@ import { CAPABILITIES, getSenchoVersion, fetchRemoteMeta, type RemoteMeta } from import { PilotTunnelManager } from '../services/PilotTunnelManager'; import { PilotCloseCode } from '../pilot/protocol'; import { FleetUpdateTrackerService } from '../services/FleetUpdateTrackerService'; +import { FleetSyncService } from '../services/FleetSyncService'; import { isValidRemoteUrl } from '../utils/validation'; import { getErrorMessage } from '../utils/errors'; @@ -156,6 +157,15 @@ nodesRouter.post('/', async (req: Request, res: Response) => { NodeRegistry.getInstance().notifyNodeAdded(id); + // Backfill replicated security state on the new remote so an operator who + // adds a node mid-life does not have to wait for the next policy edit + // before scan_policies and cve_suppressions land. No-op for local nodes + // and pilot-agent nodes (FleetSyncService.pushResource filters them out). + if (type === 'remote' && resolvedMode === 'proxy') { + FleetSyncService.getInstance().pushResourceAsync('scan_policies'); + FleetSyncService.getInstance().pushResourceAsync('cve_suppressions'); + } + let enrollment: ReturnType | null = null; if (resolvedMode === 'pilot_agent') { enrollment = mintPilotEnrollment(id, req); diff --git a/backend/src/services/FleetSyncRetryService.ts b/backend/src/services/FleetSyncRetryService.ts new file mode 100644 index 00000000..9ad704f6 --- /dev/null +++ b/backend/src/services/FleetSyncRetryService.ts @@ -0,0 +1,162 @@ +import { DatabaseService, type Node } from './DatabaseService'; +import { FleetSyncService, FLEET_RESOURCES } from './FleetSyncService'; +import { NotificationService } from './NotificationService'; +import { isDebugEnabled } from '../utils/debug'; +import { RETRY_MAX_AGE_MS, STALE_THRESHOLD_MS } from './fleetSyncConstants'; + +const INITIAL_DELAY_MS = 30_000; +const EVAL_INTERVAL_MS = 5 * 60_000; + +/** + * Background retry loop for fleet sync. + * + * The control's normal write path fires `pushResource` once per change; if a + * remote was offline at that moment, the failure is recorded on + * `fleet_sync_status` but never re-attempted unless another write happens. In + * practice an operator who sets a policy and walks away leaves the dead remote + * permanently stale. This service ticks every 5 minutes, queries the failed- + * not-yet-succeeded targets within the last 24h, and re-pushes through the + * same per-node mutex used by the write path. + * + * After STALE_THRESHOLD_MS of continuous failure for one (node, resource), we + * dispatch a single warning notification and remember the alert. The next + * recorded success clears the memo so a future failure can alert again. + */ +export class FleetSyncRetryService { + private static instance: FleetSyncRetryService; + private intervalId: NodeJS.Timeout | null = null; + private initialTimer: NodeJS.Timeout | null = null; + private isProcessing = false; + + /** Per-target alert memo: `${nodeId}:${resource}` → timestamp of last alert. */ + private alertedAt = new Map(); + + private constructor() {} + + static getInstance(): FleetSyncRetryService { + if (!FleetSyncRetryService.instance) { + FleetSyncRetryService.instance = new FleetSyncRetryService(); + } + return FleetSyncRetryService.instance; + } + + start(): void { + this.initialTimer = setTimeout(() => { + void this.evaluate(); + this.intervalId = setInterval(() => void this.evaluate(), EVAL_INTERVAL_MS); + }, INITIAL_DELAY_MS); + } + + stop(): void { + if (this.initialTimer) { + clearTimeout(this.initialTimer); + this.initialTimer = null; + } + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + } + + /** + * One pass: find every (node, resource) pair with an unresolved failure + * inside the retry window and re-push to it. Each push goes through the + * FleetSyncService per-node mutex, so a normal fanout in flight will + * serialize naturally. + */ + async evaluate(): Promise { + if (this.isProcessing) return; + this.isProcessing = true; + try { + if (FleetSyncService.getRole() === 'replica') { + if (isDebugEnabled()) { + console.debug('[FleetSyncRetry:debug] Skipping tick: this instance is a replica'); + } + return; + } + const db = DatabaseService.getInstance(); + for (const resource of FLEET_RESOURCES) { + const failedTargets = db.getFailedSyncTargets(resource, RETRY_MAX_AGE_MS); + for (const target of failedTargets) { + const node = db.getNode(target.node_id); + if (!node) continue; + await this.retryOne(node, resource, target.last_failure_at, target.last_success_at); + } + } + } catch (err) { + console.error('[FleetSyncRetry] evaluate error:', err instanceof Error ? err.message : err); + } finally { + this.isProcessing = false; + } + } + + private async retryOne( + node: Node, + resource: typeof FLEET_RESOURCES[number], + lastFailureAt: number | null, + lastSuccessAt: number | null, + ): Promise { + if (node.id == null) return; + if (isDebugEnabled()) { + console.debug( + `[FleetSyncRetry:debug] Retrying ${resource} for node "${node.name}" (id=${node.id})`, + ); + } + await FleetSyncService.getInstance().pushResourceToNode( + node as Node & { id: number }, + resource, + ); + // After the retry, re-read status to decide whether to alert. A + // successful retry clears the alert memo via the success path above + // (recordFleetSyncSuccess was called inside the push). A still-failing + // retry might have crossed the staleness threshold. + if (lastFailureAt !== null && this.exceedsStaleThreshold(lastFailureAt, lastSuccessAt)) { + this.maybeAlertStale(node, resource); + } + } + + private exceedsStaleThreshold(lastFailureAt: number, lastSuccessAt: number | null): boolean { + // Alert only when a previously-working node has been failing for more + // than the threshold. Brand-new nodes that have never succeeded fall + // through here: misconfigured remotes are caught by the test-connection + // affordance at registration time, not by this background notifier. + if (lastSuccessAt === null) return false; + return lastFailureAt - lastSuccessAt > STALE_THRESHOLD_MS; + } + + private maybeAlertStale(node: Node, resource: typeof FLEET_RESOURCES[number]): void { + if (node.id == null) return; + const key = `${node.id}:${resource}`; + const now = Date.now(); + const lastAlerted = this.alertedAt.get(key); + if (lastAlerted !== undefined && now - lastAlerted < STALE_THRESHOLD_MS) { + return; + } + // Re-read post-retry status: if the most recent push succeeded, clear + // the memo and skip the alert. The retry above ran inside the per-node + // mutex, so the status is up to date by the time we read it. + // recordFleetSyncSuccess clears last_failure_at to NULL, so a successful + // retry shows as `last_success_at !== null && last_failure_at === null`. + // A retry that landed atop an existing failure shows as both non-null + // with last_success_at >= last_failure_at. + const db = DatabaseService.getInstance(); + const fresh = db + .getFleetSyncStatuses() + .find((s) => s.node_id === node.id && s.resource === resource); + if (fresh && fresh.last_success_at !== null + && (fresh.last_failure_at === null || fresh.last_success_at >= fresh.last_failure_at)) { + this.alertedAt.delete(key); + return; + } + this.alertedAt.set(key, now); + void NotificationService.getInstance() + .dispatchAlert( + 'warning', + 'system', + `Fleet sync to node "${node.name}" has been failing for over an hour for ${resource}. Check the node's connectivity and API token.`, + ) + .catch((err) => { + console.warn('[FleetSyncRetry] Failed to dispatch stale alert:', err); + }); + } +} diff --git a/backend/src/services/FleetSyncService.ts b/backend/src/services/FleetSyncService.ts index 65190891..a8be203b 100644 --- a/backend/src/services/FleetSyncService.ts +++ b/backend/src/services/FleetSyncService.ts @@ -224,6 +224,30 @@ export class FleetSyncService { }); } + /** + * Push a resource to a single remote node. Used by the retry service to + * re-send to a node that previously failed without disturbing other nodes + * in the fleet. Goes through the same per-node mutex as `pushResource`, + * so a normal fanout and a retry never overlap on one node. + * + * No-op when this instance is a replica or when the node is not a + * proxy-mode remote with credentials configured. + */ + public async pushResourceToNode( + node: Node & { id: number }, + resource: FleetResource, + ): Promise { + if (FleetSyncService.getRole() === 'replica') return; + if (node.type !== 'remote' || !node.api_url || !node.api_token) return; + const rows = this.loadResource(resource); + const payload: Omit = { + rows, + pushedAt: this.nextPushedAt(), + controlIdentity: FleetSyncService.getControlIdentity(), + }; + await this.enqueuePushToNode(node, resource, payload); + } + /** * Apply a received sync payload on a replica. Runs control-anchor check, * staleness comparison, role flip, identity cache, row replacement, and diff --git a/backend/src/services/fleetSyncConstants.ts b/backend/src/services/fleetSyncConstants.ts index dac0ed76..cbca3308 100644 --- a/backend/src/services/fleetSyncConstants.ts +++ b/backend/src/services/fleetSyncConstants.ts @@ -45,6 +45,16 @@ export const SYNC_PATH_PREFIX = '/api/fleet/sync/'; /** How long to suppress repeat truncation alerts after one fires. */ export const TRUNCATION_ALERT_COOLDOWN_MS = 6 * 60 * 60 * 1000; +/** How far back the retry service looks for failed sync targets. */ +export const RETRY_MAX_AGE_MS = 24 * 60 * 60 * 1000; + +/** + * Failure-window threshold for the retry-service stale-target notification. + * A previously-working node whose `last_failure_at - last_success_at` exceeds + * this triggers one warning per cooldown. + */ +export const STALE_THRESHOLD_MS = 60 * 60 * 1000; + /** * Resource enum kept here so the state-key helpers below can type-check * their arguments without a cycle through FleetSyncService. The ordering