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.
This commit is contained in:
Anso
2026-05-07 13:16:24 -04:00
committed by GitHub
parent 7dde257e1f
commit 33b15d6cba
7 changed files with 389 additions and 0 deletions
@@ -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);
});
});
+2
View File
@@ -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);
}
+2
View File
@@ -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<void> {
// 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();
+10
View File
@@ -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<typeof mintPilotEnrollment> | null = null;
if (resolvedMode === 'pilot_agent') {
enrollment = mintPilotEnrollment(id, req);
@@ -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<string, number>();
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<void> {
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<void> {
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);
});
}
}
+24
View File
@@ -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<void> {
if (FleetSyncService.getRole() === 'replica') return;
if (node.type !== 'remote' || !node.api_url || !node.api_token) return;
const rows = this.loadResource(resource);
const payload: Omit<FleetSyncPayload, 'targetIdentity'> = {
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
@@ -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