fix(api): bound analytics backfill and reduce SQLite write amplification

apps/api/src/app.ts

@@ -159,24 +159,19 @@ export async function buildApp(opts: BuildAppOptions = {}) {
     // Raw stats samples grow ~1 row per agent per sync interval; drop old
     // ones daily. Lifetime totals live on agents; aggregates are kept.
     const retentionAbort = new AbortController()
-    app.addHook('onClose', async () => { retentionAbort.abort() })
-    const runRetention = async () => {
-      const deleted = await runStatsRetention(app.db, new Date().toISOString(), 1000, retentionAbort.signal)
-      if (deleted > 0) app.log.info({ deleted }, 'stats retention completed')
+    let retentionRunning: Promise<void> | null = null
+    const runRetention = () => {
+      if (retentionRunning) return retentionRunning
+      retentionRunning = runStatsRetention(app.db, new Date().toISOString(), 1000, retentionAbort.signal, {
+        onProgress: (progress) => { if (progress.processed || progress.deleted || progress.pending) app.log.info(progress, 'bounded stats maintenance completed') },
+      }).then(() => {}).catch(err => { app.log.warn({ err }, 'stats retention failed') }).finally(() => { retentionRunning = null })
+      return retentionRunning
     }
-    app.scheduler.addCronJob(
-      new CronJob(
-        { cronExpression: '17 * * * *' },
-        new AsyncTask('stats-retention', runRetention, (err) => {
-          app.log.warn({ err }, 'stats retention failed')
-        }),
-        { preventOverrun: true },
-      ),
-    )
-    // First cleanup right at startup, not only after the next 03:17.
-    runRetention().catch((err) =>
-      app.log.warn({ err }, 'stats retention failed'),
-    )
+    // Delay startup work, then resume bounded durable batches. One promise owns
+    // every tick and shutdown, so an initial run can never overlap a later tick.
+    const retentionTimer = setInterval(() => { void runRetention() }, 30_000)
+    retentionTimer.unref()
+    app.addHook('onClose', async () => { clearInterval(retentionTimer); retentionAbort.abort(); await retentionRunning })
   }
 
   return app

apps/api/src/services/analytics-maintenance.test.ts

@@ -0,0 +1,40 @@
+import { describe, expect, it } from 'vitest'
+import { createDb, runMigrations, repos } from '@evofw/db'
+import { mkdtempSync, rmSync, statSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { backfillStatsAnalytics } from './analytics.js'
+
+describe('file-backed analytics maintenance', () => {
+  it('profiles a legacy history with WAL autocheckpoint disabled and preserves all deltas', async () => {
+    const directory = mkdtempSync(join(tmpdir(), 'evofw-analytics-'))
+    const path = join(directory, 'history.db')
+    const { db, sqlite } = createDb(`sqlite:${path}`)
+    try {
+      runMigrations(sqlite)
+      const now = Date.parse('2026-10-26T12:00:00Z'), start = now - 8 * 86400_000
+      const agents = 8, samples = 500
+      for (let i = 0; i < agents; i++) repos.insertAgent(db, { id: `a${i}`, name: `a${i}`, tokenHash: `hash${i}`, tokenPrefix: `t${i}`, status: 'approved' })
+      const insert = sqlite.prepare('INSERT INTO agent_stats_samples(id,agent_id,packets_dropped,packets_accepted,recorded_at) VALUES (?,?,?,?,?)')
+      sqlite.transaction(() => {
+        for (let i = 0; i < samples; i++) for (let a = 0; a < agents; a++) insert.run(`${i}-${a}`, `a${a}`, i * 10, i * 20, new Date(start + i * 60_000).toISOString())
+      })()
+      sqlite.pragma('wal_checkpoint(TRUNCATE)')
+      sqlite.pragma('wal_autocheckpoint = 0')
+      const before = sqlite.prepare('SELECT total_changes() AS n').get() as { n: number }
+      const cpu = process.cpuUsage(), began = performance.now()
+      let processed = 0
+      while (repos.listPendingAnalytics(db, 1).length) processed += await backfillStatsAnalytics(db, new Date(now).toISOString())
+      const elapsedMs = performance.now() - began, used = process.cpuUsage(cpu)
+      const after = sqlite.prepare('SELECT total_changes() AS n').get() as { n: number }
+      expect(processed).toBe(agents * samples)
+      for (const resolution of ['1h', '1d']) {
+        const totals = sqlite.prepare('SELECT sum(packets_dropped) AS dropped,sum(packets_accepted) AS accepted FROM agent_stats_rollups WHERE resolution=?').get(resolution) as { dropped: number; accepted: number }
+        expect(totals.dropped).toBeCloseTo(agents * (samples - 1) * 10)
+        expect(totals.accepted).toBeCloseTo(agents * (samples - 1) * 20)
+      }
+      expect(await backfillStatsAnalytics(db, new Date(now).toISOString())).toBe(0)
+      console.info(JSON.stringify({ samples: agents * samples, elapsedMs, cpuMs: (used.user + used.system) / 1000, walBytes: statSync(`${path}-wal`).size, sqliteChanges: after.n - before.n }))
+    } finally { sqlite.close(); rmSync(directory, { recursive: true, force: true }) }
+  }, 60_000)
+})

apps/api/src/services/analytics.test.ts

@@ -25,6 +25,74 @@ function report(db: ReturnType<typeof memory>['db'], id: string, sequence: numbe
 const series = (db: ReturnType<typeof memory>['db'], ids = 'a', seconds = 180) => queryTimeseries(db, { from: at(0), to: at(seconds), resolution: '1m', agent_ids: ids }, at(seconds))
 
 describe('bounded statistics analytics', () => {
+  it('coalesces repeated rollup buckets into one write while preserving counters, flags and lost-report counts', () => {
+    const { db, sqlite } = memory()
+    const before = (sqlite.prepare('SELECT total_changes() AS n').get() as { n: number }).n
+    repos.addStatsRollups(db, [
+      { agentId: 'a', resolution: '1h', bucketStart: at(0), packetsDropped: 10, packetsAccepted: 20, droppedCoverageMs: 1000, acceptedCoverageMs: 1000, partial: 1, unsupportedJson: '["legacy"]', lostReports: 2 },
+      { agentId: 'a', resolution: '1h', bucketStart: at(0), packetsDropped: 30, packetsAccepted: 40, droppedCoverageMs: 2000, acceptedCoverageMs: 3000, estimated: 1, unsupportedJson: '["gap","legacy"]', lostReports: 3 },
+    ])
+    expect((sqlite.prepare('SELECT total_changes() AS n').get() as { n: number }).n - before).toBe(1)
+    const row = sqlite.prepare('SELECT * FROM agent_stats_rollups').get() as Record<string, unknown>
+    expect(row).toMatchObject({ packets_dropped: 40, packets_accepted: 60, dropped_coverage_ms: 3000, accepted_coverage_ms: 4000, partial: 1, estimated: 1, lost_reports: 5 })
+    expect(JSON.parse(String(row.unsupported_json)).sort()).toEqual(['gap', 'legacy'])
+  })
+
+  it('allows only one retention pass per DB while leaving the deferred backlog resumable', async () => {
+    const { db } = memory()
+    for (let i = 0; i < 6; i++) repos.insertStatsSample(db, { id: `flight-${i}`, agentId: 'a', packetsDropped: i * 10, packetsAccepted: 0, recordedAt: at(i * 60) })
+    let progress = 0
+    const running = runStatsRetention(db, at(2 * 86400), 1000, undefined, { maxSamples: 1, onProgress: () => { progress++ } })
+    expect(await runStatsRetention(db, at(2 * 86400))).toBe(0)
+    await running
+    expect(progress).toBe(1)
+    expect(repos.listStatsSamples(db, 'a')).toHaveLength(6)
+    await runStatsRetention(db, at(2 * 86400), 1000, undefined, { pauseMs: 0 })
+    expect(repos.listStatsSamples(db, 'a')).toHaveLength(0)
+  })
+
+  it('keeps processed predecessors until a bounded backlog fully drains, then prunes without losing its final delta', async () => {
+    const { db, sqlite } = memory()
+    for (let i = 0; i < 4; i++) repos.insertStatsSample(db, { id: `bounded-${i}`, agentId: 'a', packetsDropped: i * 10, packetsAccepted: i * 20, recordedAt: at(i * 60) })
+    const progress: { processed: number; deleted: number; pending: boolean }[] = []
+    const options = { maxSamples: 2, batchSize: 2, pauseMs: 0, onProgress: (value: typeof progress[number]) => progress.push(value) }
+    await runStatsRetention(db, at(2 * 86400), 1000, undefined, options)
+    expect(progress[0]).toMatchObject({ processed: 2, pending: true })
+    expect(repos.listStatsSamples(db, 'a')).toHaveLength(4)
+    expect((sqlite.prepare('SELECT count(*) AS n FROM agent_stats_samples WHERE analytics_processed=0').get() as { n: number }).n).toBe(2)
+    await runStatsRetention(db, at(2 * 86400), 1000, undefined, options)
+    expect(progress[1]).toMatchObject({ processed: 2, pending: false })
+    expect(repos.listStatsSamples(db, 'a')).toHaveLength(0)
+    const row = sqlite.prepare("SELECT sum(packets_dropped) AS dropped,sum(packets_accepted) AS accepted FROM agent_stats_rollups WHERE resolution='1h'").get()
+    expect(row).toEqual({ dropped: 30, accepted: 60 })
+  })
+
+  it('rolls back an entire claimed backfill batch on write failure and retries exactly once', async () => {
+    const { db, sqlite } = memory()
+    for (let i = 0; i < 3; i++) repos.insertStatsSample(db, { id: `rollback-${i}`, agentId: 'a', packetsDropped: i * 10, packetsAccepted: 0, recordedAt: at(i * 60) })
+    sqlite.exec("CREATE TRIGGER fail_backfill BEFORE INSERT ON agent_stats_rollups BEGIN SELECT RAISE(ABORT, 'backfill failed'); END")
+    await expect(backfillStatsAnalytics(db, at(120))).rejects.toThrow('backfill failed')
+    expect((sqlite.prepare('SELECT sum(analytics_processed) AS n FROM agent_stats_samples').get() as { n: number }).n).toBe(0)
+    sqlite.exec('DROP TRIGGER fail_backfill')
+    expect(await backfillStatsAnalytics(db, at(120))).toBe(3)
+    expect(await backfillStatsAnalytics(db, at(120))).toBe(0)
+    expect(series(db, 'a', 120).items.reduce((sum, row) => sum + (row.packets_dropped ?? 0), 0)).toBe(20)
+  })
+
+  it('aborts real inter-batch sleep promptly, retains pending rows and resumes without double rollups', async () => {
+    const { db } = memory()
+    for (let i = 0; i < 20; i++) repos.insertStatsSample(db, { id: `abort-${i}`, agentId: 'a', packetsDropped: i * 10, packetsAccepted: 0, recordedAt: at(i * 60) })
+    const controller = new AbortController()
+    const timer = setTimeout(() => controller.abort(), 5)
+    const count = await backfillStatsAnalytics(db, at(1200), controller.signal, { batchSize: 2, pauseMs: 500 })
+    clearTimeout(timer)
+    expect(count).toBeGreaterThan(0)
+    expect(count).toBeLessThan(20)
+    const remaining = await backfillStatsAnalytics(db, at(1200), undefined, { pauseMs: 0 })
+    expect(count + remaining).toBe(20)
+    expect(series(db, 'a', 1200).items.reduce((sum, row) => sum + (row.packets_dropped ?? 0), 0)).toBe(190)
+  })
+
   it('sums contained exact minute intervals into an exact hour and flags only proportional cross-bucket splits', () => {
     const { db } = memory()
     report(db, 'a', 1, 0, 0, 0)

apps/api/src/services/analytics.ts

@@ -1,6 +1,7 @@
 import { repos, type Db, type StatsRollupInsert } from '@evofw/db'
 import { type StatsResolution, type TelemetryQuality, type TimeseriesResponse } from '@evofw/shared'
 import { AppError } from '../plugins/error-handler.js'
+import { setTimeout as sleep } from 'node:timers/promises'
 
 export const RESOLUTION_MS: Record<StatsResolution, number> = { '1m': 60_000, '1h': 3_600_000, '1d': 86_400_000 }
 export const RETENTION_MS: Record<StatsResolution, number> = { '1m': 86_400_000, '1h': 30 * 86_400_000, '1d': 90 * 86_400_000 }
@@ -93,17 +94,44 @@ export function queryTimeseries(db: Db, query: TimeseriesQuery, now = new Date()
   return { from: iso(from), to: iso(to), resolution: selectedResolution, items, agent_ids: ids }
 }
 
-/** Old samples become explicitly estimated once, before raw history is removed. */
-export async function backfillStatsAnalytics(db: Db, now = new Date().toISOString(), signal?: AbortSignal) {
+export interface AnalyticsMaintenanceOptions {
+  maxSamples?: number
+  batchSize?: number
+  pauseMs?: number
+  maxDurationMs?: number
+  onProgress?: (value: { processed: number; deleted: number; pending: boolean }) => void
+}
+function maintenanceOptions(options: AnalyticsMaintenanceOptions) {
+  const result = { maxSamples: options.maxSamples ?? 1000, batchSize: options.batchSize ?? 50, pauseMs: options.pauseMs ?? 25, maxDurationMs: options.maxDurationMs ?? 1000 }
+  if (!Number.isInteger(result.maxSamples) || result.maxSamples < 1 || result.maxSamples > 5000 || !Number.isInteger(result.batchSize) || result.batchSize < 1 || result.batchSize > 100 || !Number.isInteger(result.pauseMs) || result.pauseMs < 0 || result.pauseMs > 1000 || !Number.isFinite(result.maxDurationMs) || result.maxDurationMs < 1 || result.maxDurationMs > 10_000) throw new Error('Invalid analytics maintenance budget')
+  return result
+}
+async function pause(ms: number, signal?: AbortSignal) {
+  try { await sleep(ms, undefined, { signal }) } catch (error) { if (!signal?.aborted) throw error }
+}
+const backfills = new WeakSet<Db>()
+const retentions = new WeakSet<Db>()
+
+/** Atomic bounded legacy batches; pending rows remain durable between ticks. */
+export async function backfillStatsAnalytics(db: Db, now = new Date().toISOString(), signal?: AbortSignal, options: AnalyticsMaintenanceOptions = {}) {
+  const budget = maintenanceOptions(options)
+  if (!Number.isFinite(Date.parse(now))) throw new Error('Invalid analytics timestamp')
+  if (backfills.has(db)) return 0
+  backfills.add(db)
+  const started = performance.now()
   let processed = 0
+  try {
   for (;;) {
-    if (signal?.aborted) return processed
-    const pending = repos.listPendingAnalytics(db, 100)
+    if (signal?.aborted || processed >= budget.maxSamples || performance.now() - started >= budget.maxDurationMs) return processed
+    const pending = repos.listPendingAnalytics(db, Math.min(budget.batchSize, budget.maxSamples - processed))
     if (!pending.length) return processed
-    for (const sample of pending) {
-      if (signal?.aborted) return processed
-      const claimed = db.transaction((tx) => {
-        if (!repos.markAnalyticsProcessed(tx, sample.id)) return false
+    const claimed = db.transaction((tx) => {
+      let count = 0
+      const combined: StatsRollupInsert[] = []
+      const batchStarted = performance.now()
+      for (const sample of pending) {
+        if (signal?.aborted || (count > 0 && (performance.now() - batchStarted >= 10 || combined.length >= 5000))) break
+        if (!repos.markAnalyticsProcessed(tx, sample.id)) continue
         const previous = repos.previousStatsSample(tx, sample)
         const delta = (current: number, before: number | undefined, present: number | null) => present === 0 || before === undefined ? null : current >= before ? current - before : current
         // An explicit v2 null is an unknown epoch/baseline interval, never an
@@ -114,31 +142,45 @@ export async function backfillStatsAnalytics(db: Db, now = new Date().toISOStrin
           { start: sample.droppedIntervalStart ?? previous?.recordedAt ?? null, end: sample.recordedAt, delta: dropped },
           { start: sample.acceptedIntervalStart ?? previous?.recordedAt ?? null, end: sample.recordedAt, delta: accepted },
           { partial: true, truncated: false, unsupported: ['legacy_estimated_backfill'], lost_reports: 0 }, now, true)
-        repos.addStatsRollups(tx, rows)
-        return true
-      })
-      if (claimed && ++processed % 10 === 0) await new Promise<void>((resolve) => setImmediate(resolve))
-    }
+        combined.push(...rows)
+        count++
+      }
+      repos.addStatsRollups(tx, combined)
+      return count
+    })
+    processed += claimed
+    if (!claimed) return processed
+    await pause(budget.pauseMs, signal)
   }
+  } finally { backfills.delete(db) }
 }
 
-export async function runStatsRetention(db: Db, now = new Date().toISOString(), batchSize = 1000, signal?: AbortSignal) {
+export async function runStatsRetention(db: Db, now = new Date().toISOString(), batchSize = 1000, signal?: AbortSignal, options: AnalyticsMaintenanceOptions = {}) {
   if (!Number.isInteger(batchSize) || batchSize < 1 || batchSize > 5000 || !Number.isFinite(Date.parse(now))) throw new Error('Invalid retention parameters')
+  const budget = maintenanceOptions(options)
+  if (retentions.has(db)) return 0
+  retentions.add(db)
   let deleted = 0
-  await backfillStatsAnalytics(db, now, signal)
+  try {
+  const processed = await backfillStatsAnalytics(db, now, signal, options)
+  const pending = repos.listPendingAnalytics(db, 1).length > 0 || backfills.has(db)
   const nowMs = Date.parse(now)
   const cutoffs = [
     ['raw', iso(nowMs - 86_400_000)], ['ip', iso(nowMs - 7 * 86_400_000)], ['port', iso(nowMs - 7 * 86_400_000)],
     ...(['1m', '1h', '1d'] as const).map((resolution) => [resolution, iso(floor(nowMs - RETENTION_MS[resolution], RESOLUTION_MS[resolution]))]),
   ] as const
   for (const [kind, cutoff] of cutoffs) {
-    for (;;) {
+    // Pending legacy rows still need their processed raw predecessors.
+    if (kind === 'raw' && pending) continue
+    for (let batch = 0; batch < 2; batch++) {
       if (signal?.aborted) return deleted
       const count = repos.deleteAnalyticsBatch(db, kind as 'raw' | 'ip' | 'port' | StatsResolution, cutoff, batchSize)
       deleted += count
       if (count < batchSize) break
-      await new Promise<void>((resolve) => setImmediate(resolve))
+      await pause(budget.pauseMs, signal)
     }
   }
+  options.onProgress?.({ processed, deleted, pending })
   return deleted
+  } finally { retentions.delete(db) }
 }