From 6b728599c471034761d6b3e48772eedc899d08eb Mon Sep 17 00:00:00 2001 From: Anso Date: Sun, 17 May 2026 01:55:17 -0400 Subject: [PATCH] fix(mesh): persist PilotMetrics counters across central restart (#1078) The 12 PilotMetrics counters (proxy_dials_failed, proxy_bridges_total, proxy_idle_closes, the pilot-tunnel counters, and the peer-to-central callback counters) live in a singleton holding scalar fields, so process restart wipes them. Operators investigating rotation- or dial-failure trends across a central restart lost the aggregate metric. Persist the snapshot under a pilot_metrics_counters JSON row on the existing system_state KV table. Hydrate via PilotMetrics.load() in bootstrap/startup.ts before any service that increments runs; final-flush in bootstrap/shutdown.ts before the SQLite handle closes. Between those boundaries a buffered flush mirrors the audit-log pattern: every 1 s or every 100 increments, whichever fires first. Hot increment path stays a single property write plus a pending-counter check. Defensive parse on the row: object check + numeric filter, so an operator hand-edit or schema drift across releases degrades to zero state with a warn rather than crashing boot. Schema additions back-fill missing keys with zero so a release that adds a counter does not need a migration. Closes F-R-4 from the mesh tracker. --- backend/src/__tests__/pilot-metrics.test.ts | 187 ++++++++++++++++++++ backend/src/bootstrap/shutdown.ts | 4 + backend/src/bootstrap/startup.ts | 10 ++ backend/src/services/DatabaseService.ts | 35 ++++ backend/src/services/PilotMetrics.ts | 156 +++++++++++++--- 5 files changed, 369 insertions(+), 23 deletions(-) create mode 100644 backend/src/__tests__/pilot-metrics.test.ts diff --git a/backend/src/__tests__/pilot-metrics.test.ts b/backend/src/__tests__/pilot-metrics.test.ts new file mode 100644 index 00000000..ca8b6738 --- /dev/null +++ b/backend/src/__tests__/pilot-metrics.test.ts @@ -0,0 +1,187 @@ +/** + * PilotMetrics persistence behavior. Verifies the F-R-4 contract: + * + * - cold start: zero counters, no DB row + * - cold start with a persisted blob: snapshot reflects persisted values + * - increments queue writes that drain on the threshold or interval + * - explicit flush() persists the current snapshot + * - malformed JSON in the DB row degrades gracefully (warn, zero state) + * - persisted blob missing a counter back-fills that field with zero + * (schema-drift scenario for releases that add a new counter) + * - stop() cancels a pending interval flush without writing + */ +import { describe, it, expect, beforeEach, afterEach, beforeAll, afterAll, vi } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let PilotMetrics: typeof import('../services/PilotMetrics').PilotMetrics; +let PILOT_METRICS_COUNTERS_KEY: string; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService, PILOT_METRICS_COUNTERS_KEY } = await import('../services/DatabaseService')); + ({ PilotMetrics } = await import('../services/PilotMetrics')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +beforeEach(() => { + // Clear any persisted blob from a prior test and reset the singleton. + const db = DatabaseService.getInstance(); + db.getDb().prepare('DELETE FROM system_state WHERE key = ?').run(PILOT_METRICS_COUNTERS_KEY); + PilotMetrics.resetForTests(); +}); + +afterEach(() => { + PilotMetrics.stop(); + vi.useRealTimers(); +}); + +describe('PilotMetrics persistence', () => { + it('cold start: load with no row returns zeros and writes nothing', () => { + const db = DatabaseService.getInstance(); + const setSpy = vi.spyOn(db, 'setPilotMetricsCounters'); + PilotMetrics.load(db); + const snap = PilotMetrics.snapshot(); + expect(snap.proxy_dials_failed).toBe(0); + expect(snap.proxy_bridges_total).toBe(0); + expect(snap.proxy_idle_closes).toBe(0); + expect(snap.tunnels_total).toBe(0); + PilotMetrics.flush(); + expect(setSpy).not.toHaveBeenCalled(); + setSpy.mockRestore(); + }); + + it('cold start with persisted blob: snapshot reflects persisted values', () => { + const db = DatabaseService.getInstance(); + db.setPilotMetricsCounters({ + tunnels_total: 11, + tunnels_replaced: 1, + tunnels_rejected_capacity: 0, + enroll_acks: 0, + frame_decode_errors: 0, + proxy_bridges_total: 7, + proxy_dials_failed: 5, + proxy_idle_closes: 2, + proxy_bridges_peer_initiated_total: 3, + mesh_central_bootstraps_total: 4, + mesh_callback_dials_failed_total: 0, + mesh_callback_auth_failures_total: 0, + }); + PilotMetrics.load(db); + const snap = PilotMetrics.snapshot(); + expect(snap.tunnels_total).toBe(11); + expect(snap.proxy_dials_failed).toBe(5); + expect(snap.proxy_bridges_total).toBe(7); + expect(snap.mesh_central_bootstraps_total).toBe(4); + }); + + it('threshold flush: increments past the threshold trigger a single persist', () => { + const db = DatabaseService.getInstance(); + const setSpy = vi.spyOn(db, 'setPilotMetricsCounters'); + PilotMetrics.load(db, { threshold: 3, intervalMs: 60_000 }); + PilotMetrics.increment('proxy_dials_failed'); + PilotMetrics.increment('proxy_dials_failed'); + expect(setSpy).not.toHaveBeenCalled(); + PilotMetrics.increment('proxy_dials_failed'); + expect(setSpy).toHaveBeenCalledTimes(1); + const written = setSpy.mock.calls[0][0]; + expect(written.proxy_dials_failed).toBe(3); + setSpy.mockRestore(); + }); + + it('interval flush: a single increment persists after the interval fires', () => { + vi.useFakeTimers(); + const db = DatabaseService.getInstance(); + const setSpy = vi.spyOn(db, 'setPilotMetricsCounters'); + PilotMetrics.load(db, { threshold: 1000, intervalMs: 1_000 }); + PilotMetrics.increment('proxy_bridges_total'); + expect(setSpy).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1_000); + expect(setSpy).toHaveBeenCalledTimes(1); + expect(setSpy.mock.calls[0][0].proxy_bridges_total).toBe(1); + setSpy.mockRestore(); + }); + + it('explicit flush with no pending writes is a no-op', () => { + const db = DatabaseService.getInstance(); + const setSpy = vi.spyOn(db, 'setPilotMetricsCounters'); + PilotMetrics.load(db); + PilotMetrics.flush(); + expect(setSpy).not.toHaveBeenCalled(); + setSpy.mockRestore(); + }); + + it('explicit flush after one increment persists immediately', () => { + const db = DatabaseService.getInstance(); + const setSpy = vi.spyOn(db, 'setPilotMetricsCounters'); + PilotMetrics.load(db, { threshold: 1000, intervalMs: 60_000 }); + PilotMetrics.increment('proxy_idle_closes'); + PilotMetrics.flush(); + expect(setSpy).toHaveBeenCalledTimes(1); + expect(setSpy.mock.calls[0][0].proxy_idle_closes).toBe(1); + setSpy.mockRestore(); + }); + + it('malformed JSON in the persisted row degrades to zero state', () => { + const db = DatabaseService.getInstance(); + db.setSystemState(PILOT_METRICS_COUNTERS_KEY, '{not json'); + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + PilotMetrics.load(db); + const snap = PilotMetrics.snapshot(); + expect(snap.proxy_dials_failed).toBe(0); + expect(warnSpy).toHaveBeenCalled(); + PilotMetrics.increment('proxy_dials_failed'); + expect(PilotMetrics.snapshot().proxy_dials_failed).toBe(1); + warnSpy.mockRestore(); + }); + + it('schema drift: persisted blob missing a counter back-fills with zero', () => { + const db = DatabaseService.getInstance(); + db.setSystemState( + PILOT_METRICS_COUNTERS_KEY, + JSON.stringify({ proxy_dials_failed: 9 }), + ); + PilotMetrics.load(db); + const snap = PilotMetrics.snapshot(); + expect(snap.proxy_dials_failed).toBe(9); + expect(snap.proxy_bridges_total).toBe(0); + expect(snap.tunnels_total).toBe(0); + expect(snap.mesh_callback_auth_failures_total).toBe(0); + }); + + it('stop() cancels a pending interval flush without persisting', () => { + vi.useFakeTimers(); + const db = DatabaseService.getInstance(); + const setSpy = vi.spyOn(db, 'setPilotMetricsCounters'); + PilotMetrics.load(db, { threshold: 1000, intervalMs: 1_000 }); + PilotMetrics.increment('proxy_dials_failed'); + PilotMetrics.stop(); + vi.advanceTimersByTime(5_000); + expect(setSpy).not.toHaveBeenCalled(); + setSpy.mockRestore(); + }); + + it('flush() catches DB write errors without losing pending count', () => { + const db = DatabaseService.getInstance(); + const setSpy = vi + .spyOn(db, 'setPilotMetricsCounters') + .mockImplementationOnce(() => { throw new Error('disk full'); }) + .mockImplementationOnce(() => undefined); + const errSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + PilotMetrics.load(db, { threshold: 1000, intervalMs: 60_000 }); + PilotMetrics.increment('proxy_dials_failed'); + PilotMetrics.flush(); + expect(setSpy).toHaveBeenCalledTimes(1); + expect(errSpy).toHaveBeenCalled(); + // Counter still in memory; next successful flush persists the value. + PilotMetrics.flush(); + expect(setSpy).toHaveBeenCalledTimes(2); + expect(setSpy.mock.calls[1][0].proxy_dials_failed).toBe(1); + setSpy.mockRestore(); + errSpy.mockRestore(); + }); +}); diff --git a/backend/src/bootstrap/shutdown.ts b/backend/src/bootstrap/shutdown.ts index 92d03c1c..d7f2ebe1 100644 --- a/backend/src/bootstrap/shutdown.ts +++ b/backend/src/bootstrap/shutdown.ts @@ -10,6 +10,7 @@ import { SchedulerService } from '../services/SchedulerService'; import { MfaService } from '../services/MfaService'; import { MeshService } from '../services/MeshService'; import { BlueprintReconciler } from '../services/BlueprintReconciler'; +import { PilotMetrics } from '../services/PilotMetrics'; /** * Wire graceful shutdown handlers. Docker sends SIGTERM when the container @@ -49,6 +50,9 @@ export function installShutdownHandlers(server: Server): void { try { BlueprintReconciler.getInstance().stop(); } catch (e) { console.warn('[Shutdown] BlueprintReconciler cleanup failed:', (e as Error).message); } + try { PilotMetrics.flush(); } catch (e) { + console.warn('[Shutdown] PilotMetrics flush failed:', (e as Error).message); + } try { DatabaseService.getInstance().flushAuditLogBuffer(); } catch (e) { console.warn('[Shutdown] Audit log flush failed:', (e as Error).message); } diff --git a/backend/src/bootstrap/startup.ts b/backend/src/bootstrap/startup.ts index c4e07410..765eac1b 100644 --- a/backend/src/bootstrap/startup.ts +++ b/backend/src/bootstrap/startup.ts @@ -17,6 +17,7 @@ import { MeshService } from '../services/MeshService'; import { BlueprintReconciler } from '../services/BlueprintReconciler'; import { applyPilotModeCapabilityFilter } from '../services/CapabilityRegistry'; import { PilotTunnelManager } from '../services/PilotTunnelManager'; +import { PilotMetrics } from '../services/PilotMetrics'; import { invalidateRemoteMetaCache } from '../helpers/cacheInvalidation'; import { sweepStaleTempDirs as sweepStaleGitTempDirs } from '../services/GitSourceService'; import { PORT } from '../helpers/constants'; @@ -70,6 +71,15 @@ export async function startServer(server: Server): Promise { applyPilotModeCapabilityFilter(); } + // Hydrate pilot/mesh counters from the persisted snapshot before any + // service that increments them (MeshService, PilotTunnelManager) starts. + // Failures fall through to zero-initialized counters; do not block boot. + try { + PilotMetrics.load(DatabaseService.getInstance()); + } catch (err) { + console.warn('[Startup] PilotMetrics load failed:', (err as Error).message); + } + // Initialize the license service before any tier-gated code can run. LicenseService.getInstance().initialize(); diff --git a/backend/src/services/DatabaseService.ts b/backend/src/services/DatabaseService.ts index 3bcebcb0..293839ae 100644 --- a/backend/src/services/DatabaseService.ts +++ b/backend/src/services/DatabaseService.ts @@ -605,6 +605,8 @@ export interface ScanSummary { const AUDIT_LOG_FLUSH_INTERVAL_MS = 1_000; const AUDIT_LOG_FLUSH_THRESHOLD = 100; +export const PILOT_METRICS_COUNTERS_KEY = 'pilot_metrics_counters'; + export class DatabaseService { private static instance: DatabaseService; private db: Database.Database; @@ -1775,6 +1777,39 @@ export class DatabaseService { this.db.prepare('INSERT OR REPLACE INTO system_state (key, value) VALUES (?, ?)').run(key, value); } + /** + * Persisted PilotMetrics counters. Returns the parsed JSON object (a + * numeric record keyed by counter name) or null when the row is missing + * or unparseable. Callers (PilotMetrics.load) handle per-field defaulting + * so a missing counter in the persisted blob does not break a new release + * that added the counter. + * + * This is the first JSON blob stored in `system_state`; mirror this + * parse/validate shape (object check + numeric filter) for any future + * JSON-shaped system_state row so an operator-edited row cannot crash + * boot. + */ + public getPilotMetricsCounters(): Record | null { + const raw = this.getSystemState(PILOT_METRICS_COUNTERS_KEY); + if (raw === null) return null; + try { + const parsed = JSON.parse(raw) as unknown; + if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) return null; + const out: Record = {}; + for (const [k, v] of Object.entries(parsed as Record)) { + if (typeof v === 'number' && Number.isFinite(v)) out[k] = v; + } + return out; + } catch (err) { + console.warn('[DatabaseService] pilot_metrics_counters JSON parse failed:', (err as Error).message); + return null; + } + } + + public setPilotMetricsCounters(counters: Record): void { + this.setSystemState(PILOT_METRICS_COUNTERS_KEY, JSON.stringify(counters)); + } + /** * Run `fn` inside a single SQLite transaction. better-sqlite3 promotes a * nested call to a SAVEPOINT, so callers can compose this with methods diff --git a/backend/src/services/PilotMetrics.ts b/backend/src/services/PilotMetrics.ts index d9ae187d..118881af 100644 --- a/backend/src/services/PilotMetrics.ts +++ b/backend/src/services/PilotMetrics.ts @@ -1,16 +1,24 @@ /** - * PilotMetrics: in-memory counters for the pilot-agent reverse-tunnel - * subsystem. Strictly process-local — Sencho does not export metrics to any - * external sink (privacy posture, see CLAUDE.md). Counters reset on process - * restart by design; their purpose is operator support and debug, not - * long-term trend analysis. + * PilotMetrics: process-local counters for the pilot-agent reverse-tunnel + * subsystem and the proxy-mode mesh dialer. Strictly process-local. Sencho + * does not export metrics to any external sink (privacy posture, see + * CLAUDE.md). Their purpose is operator support and debug, surfaced via + * GET /api/system/pilot-tunnels (admin only). * - * No general in-process metrics facility exists in the backend today, so this - * is a per-feature pattern. When a shared facility lands, this module should - * be replaced by an instance of it rather than grown. + * Counters are hydrated from SQLite on startup via load() and persisted via + * a buffered flush that mirrors the audit-log buffer pattern in + * DatabaseService: one flush per PILOT_METRICS_FLUSH_INTERVAL_MS or + * PILOT_METRICS_FLUSH_THRESHOLD increments, whichever comes first, plus an + * explicit flush() in the shutdown handler before the DB closes. + * + * No general in-process metrics facility exists in the backend today, so + * this is a per-feature pattern. When a shared facility lands, this module + * should be replaced by an instance of it rather than grown. */ -interface Counters { +import type { DatabaseService } from './DatabaseService'; + +export interface Counters { tunnels_total: number; tunnels_replaced: number; tunnels_rejected_capacity: number; @@ -32,29 +40,131 @@ interface Counters { mesh_callback_auth_failures_total: number; } +const ZERO_COUNTERS: Counters = { + tunnels_total: 0, + tunnels_replaced: 0, + tunnels_rejected_capacity: 0, + enroll_acks: 0, + frame_decode_errors: 0, + proxy_bridges_total: 0, + proxy_dials_failed: 0, + proxy_idle_closes: 0, + proxy_bridges_peer_initiated_total: 0, + mesh_central_bootstraps_total: 0, + mesh_callback_dials_failed_total: 0, + mesh_callback_auth_failures_total: 0, +}; + +export const PILOT_METRICS_FLUSH_INTERVAL_MS = 1_000; +export const PILOT_METRICS_FLUSH_THRESHOLD = 100; + +export interface PilotMetricsTestOverrides { + intervalMs?: number; + threshold?: number; +} + class PilotMetricsImpl { - private counters: Counters = { - tunnels_total: 0, - tunnels_replaced: 0, - tunnels_rejected_capacity: 0, - enroll_acks: 0, - frame_decode_errors: 0, - proxy_bridges_total: 0, - proxy_dials_failed: 0, - proxy_idle_closes: 0, - proxy_bridges_peer_initiated_total: 0, - mesh_central_bootstraps_total: 0, - mesh_callback_dials_failed_total: 0, - mesh_callback_auth_failures_total: 0, - }; + private counters: Counters = { ...ZERO_COUNTERS }; + private db: DatabaseService | null = null; + private pendingWrites = 0; + private flushTimer: ReturnType | null = null; + private intervalMs: number = PILOT_METRICS_FLUSH_INTERVAL_MS; + private threshold: number = PILOT_METRICS_FLUSH_THRESHOLD; + + /** + * Hydrate counters from persisted JSON (if any) and wire periodic flush + * to disk. Idempotent: a second call rebinds the DB reference and + * re-hydrates from it but does not double-schedule the timer. + */ + public load(db: DatabaseService, overrides?: PilotMetricsTestOverrides): void { + this.db = db; + if (overrides?.intervalMs !== undefined) this.intervalMs = overrides.intervalMs; + if (overrides?.threshold !== undefined) this.threshold = overrides.threshold; + const persisted = db.getPilotMetricsCounters(); + const next: Counters = { ...ZERO_COUNTERS }; + if (persisted) { + for (const key of Object.keys(next) as Array) { + const value = persisted[key]; + if (typeof value === 'number' && Number.isFinite(value)) { + next[key] = value; + } + } + } + this.counters = next; + this.pendingWrites = 0; + } public increment(name: K): void { this.counters[name] += 1; + if (!this.db) return; + this.pendingWrites += 1; + if (this.pendingWrites >= this.threshold) { + this.flush(); + return; + } + if (!this.flushTimer) { + this.flushTimer = setTimeout(() => { + this.flushTimer = null; + this.flush(); + }, this.intervalMs); + // Allow process exit even when this timer is pending; shutdown + // calls flush() explicitly and a stuck DB write must not block + // SIGTERM. + if (typeof this.flushTimer.unref === 'function') { + this.flushTimer.unref(); + } + } } public snapshot(): Counters { return { ...this.counters }; } + + /** + * Persist the current counter snapshot to SQLite. Safe to call from any + * path; the shutdown handler invokes this before closing the DB. A flush + * with no pending writes is a no-op (avoids writing on every shutdown + * even when nothing changed since load). + */ + public flush(): void { + if (this.flushTimer) { + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + if (!this.db || this.pendingWrites === 0) return; + try { + this.db.setPilotMetricsCounters({ ...this.counters }); + this.pendingWrites = 0; + } catch (err) { + // Keep counters in memory so the next flush retries with the + // accumulated value. Do NOT zero pendingWrites here. + console.error('[PilotMetrics] Failed to persist counters:', (err as Error).message); + } + } + + /** + * Cancel any pending interval flush without writing. Intended for tests; + * the production shutdown path uses flush() instead so in-memory writes + * survive the restart. + */ + public stop(): void { + if (this.flushTimer) { + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + } + + /** + * Reset all in-memory state. Tests only. + */ + public resetForTests(): void { + this.stop(); + this.counters = { ...ZERO_COUNTERS }; + this.db = null; + this.pendingWrites = 0; + this.intervalMs = PILOT_METRICS_FLUSH_INTERVAL_MS; + this.threshold = PILOT_METRICS_FLUSH_THRESHOLD; + } } export const PilotMetrics = new PilotMetricsImpl();