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
+      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
+        combined.push(...rows)
+        count++
+      }
+      repos.addStatsRollups(tx, combined)
+      return count
     })
-      if (claimed && ++processed % 10 === 0) await new Promise<void>((resolve) => setImmediate(resolve))
-    }
+    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) }
 }

docs/performance-maintenance.md

@@ -0,0 +1,22 @@
+# Statistics maintenance resource correction
+
+The production report showed sustained CPU and substantial disk IO after the update. Inspection found that startup immediately drained every pending legacy statistics row: each sample opened a transaction, changed its pending index and rewrote additive rollup buckets. `setImmediate` yielded to requests but did not throttle this work. Existing logs do not establish that this is the sole production cause; no production CPU/IO profile was obtained.
+
+The correction processes durable, atomic batches and merges observations sharing an agent/hour/day bucket before writing. Default maintenance processes at most 1000 samples or approximately one second per invocation, with up to 50 samples per transaction, a cooperative 10ms computation check and real 25ms inter-batch pauses. The computation check is not a hard deadline for synchronous SQLite writes. Cleanup performs at most two deletion batches per tier per invocation.
+
+The application waits 30 seconds before its first maintenance tick and resumes every 30 seconds with one shared in-flight promise. Backfill/retention also reject overlapping passes for the same DB object. Shutdown aborts waits and awaits the active pass before DB closure. Pending rows remain durable between ticks. Processed raw predecessors are retained until the entire backfill backlog drains, so resumed legacy intervals retain their counter baseline; raw deletion still requires processed status. Marks and rollup writes commit or roll back together, preventing replay double-counting.
+
+## Local before/after observation
+
+The same Windows Node/SQLite file-backed fixture seeded 4000 legacy samples across eight agents, eight days old. After seeding it truncated the WAL and disabled automatic checkpointing, making generated WAL bytes observable. Each profile drained all pending work, checked exact hourly/daily dropped and accepted totals, and verified that a second pass processed zero rows. No wall-time assertion or production capacity claim is made.
+
+| Observation | Previous implementation | Corrected implementation |
+| --- | ---: | ---: |
+| Wall time | 7341.96ms | 3877.81ms |
+| Process CPU | 5641ms | 515ms |
+| Generated WAL bytes | 79561352 | 3860472 |
+| SQLite row changes | 12000 | 5380 |
+
+The corrected wall time includes real inter-batch pauses. These were isolated local observations, not production measurements; WAL auto-checkpointing was deliberately disabled for both. The profile drains successive bounded passes immediately for comparison, while the actual application spaces passes 30 seconds apart. Its maximum sample budget therefore allows at most about 33 legacy samples/second averaged across ticks; actual progress depends on the one-second budget and host load. Progress logs report processed/deleted counts and whether more pending history remains. A large backlog is resumed rather than silently discarded.
+
+Focused regressions cover predecessor preservation across partial backfill/retention/resume, transaction rollback, exact-once retries, abort/resume, overlapping passes and one-write bucket coalescing with preserved flags/counters. Root's complete API suite passed 285 tests and shared/DB/API production builds passed. A full-suite concurrent profile also retained the reduction (3889312 WAL bytes); its CPU/timing is not substituted for the isolated comparison above. Astra review found no confirmed P1/P2 issue. Production CPU, IO and backlog progress still need observation after rollout.

packages/db/src/repositories/analytics.ts

@@ -6,6 +6,18 @@ export type StatsRollupInsert = typeof agentStatsRollups.$inferInsert
 
 /** Called inside the ingest transaction; replay checks precede this additive write. */
 export function addStatsRollups(db: Db, rows: StatsRollupInsert[]) {
+  // Backfill observations often share hour/day buckets. Merge them before
+  // UPSERT rather than rewriting the same SQLite pages per observation.
+  const merged = new Map<string, StatsRollupInsert>()
+  for (const input of rows) {
+    const key = `${input.agentId}\0${input.resolution}\0${input.bucketStart}`
+    const row = merged.get(key)
+    if (!row) { merged.set(key, { ...input }); continue }
+    for (const field of ['packetsDropped', 'packetsAccepted', 'droppedCoverageMs', 'acceptedCoverageMs', 'lostReports'] as const) row[field] = (row[field] ?? 0) + (input[field] ?? 0)
+    for (const field of ['partial', 'truncated', 'estimated'] as const) row[field] = Math.max(row[field] ?? 0, input[field] ?? 0)
+    row.unsupportedJson = JSON.stringify([...new Set([...(JSON.parse(row.unsupportedJson ?? '[]') as string[]), ...(JSON.parse(input.unsupportedJson ?? '[]') as string[])])].slice(0, 16))
+  }
+  rows = [...merged.values()]
   for (let offset = 0; offset < rows.length; offset += 100) {
     db.insert(agentStatsRollups).values(rows.slice(offset, offset + 100)).onConflictDoUpdate({
       target: [agentStatsRollups.agentId, agentStatsRollups.resolution, agentStatsRollups.bucketStart],