EXPERIMENTAL: new stats logic

server/services/stats.js

@@ -1,22 +1,269 @@
 /**
  * Container stats collection and broadcast.
  *
- * Docker stats streams are NDJSON and chunks may contain partial lines or
- * multiple JSON objects — we buffer and parse line-by-line so CPU/memory
- * actually update (JSON.parse on raw chunks often fails silently).
+ * Uses **one-shot** Docker stats (stream:false) with bounded concurrency —
+ * not perpetual `stats({ stream: true })` attachments.
+ *
+ * Why: each live stats stream forces dockerd to sample cgroups ~1 Hz for that
+ * container for as long as the stream is open. With N running containers that
+ * is N concurrent streams and is a well-known host CPU / load-average killer.
+ * One-shot polls only work when we ask, and we can throttle concurrency.
+ *
+ * Collection only runs while peers.size > 0 (paused when idle).
  */
-import { docker, extractIpAddress } from './docker.js'
+import fs from 'fs'
+import path from 'path'
+import { docker } from './docker.js'
 import { peers } from '../core/peer-registry.js'
 import { Pushes } from '../../shared/protocol.js'
 import { recordSample, pruneMissing } from './stats-history.js'
 import { CONFIG } from '../../config.js'
 import logger from '../utils/logger.js'
 
-const STATS_CACHE_TTL = CONFIG.STATS?.CACHE_TTL_MS ?? 1000
-/** How often we tick collection when peers are connected */
-const STATS_COLLECT_INTERVAL = CONFIG.STATS?.ACTIVE_INTERVAL_MS ?? 1000
-/** Minimum gap between allStats broadcasts */
-const STATS_BROADCAST_INTERVAL = CONFIG.STATS?.INTERVAL_MS ?? 2000
+const STATS_FILE =
+  process.env.PEARDOCK_STATS_PATH || path.join(process.cwd(), 'peardock-stats.json')
+
+/** Clamp helpers for RPC / env / disk */
+export const STATS_LIMITS = Object.freeze({
+  collectIntervalMs: { min: 1000, max: 60_000 },
+  broadcastIntervalMs: { min: 1000, max: 60_000 },
+  concurrency: { min: 1, max: 32 },
+  cacheTtlMs: { min: 0, max: 30_000 },
+  warnContainers: { min: 10, max: 10_000 },
+})
+
+function envInt(name, fallback) {
+  const n = Number(process.env[name])
+  return Number.isFinite(n) && n > 0 ? Math.floor(n) : fallback
+}
+
+function clampInt(value, min, max, fallback) {
+  const n = Number(value)
+  if (!Number.isFinite(n)) return fallback
+  return Math.min(max, Math.max(min, Math.floor(n)))
+}
+
+/**
+ * Built-in defaults, then env overrides (bootstrap before disk / UI).
+ * @returns {{
+ *   collectIntervalMs: number,
+ *   broadcastIntervalMs: number,
+ *   concurrency: number,
+ *   cacheTtlMs: number,
+ *   warnContainers: number,
+ * }}
+ */
+export function defaultStatsConfig() {
+  const collectIntervalMs = clampInt(
+    envInt(
+      'PEARDOCK_STATS_INTERVAL_MS',
+      CONFIG.STATS?.ACTIVE_INTERVAL_MS ?? CONFIG.STATS?.INTERVAL_MS ?? 2000
+    ),
+    STATS_LIMITS.collectIntervalMs.min,
+    STATS_LIMITS.collectIntervalMs.max,
+    2000
+  )
+  const broadcastIntervalMs = clampInt(
+    envInt('PEARDOCK_STATS_BROADCAST_MS', CONFIG.STATS?.INTERVAL_MS ?? 2000),
+    STATS_LIMITS.broadcastIntervalMs.min,
+    STATS_LIMITS.broadcastIntervalMs.max,
+    2000
+  )
+  const concurrency = clampInt(
+    envInt('PEARDOCK_STATS_CONCURRENCY', CONFIG.STATS?.CONCURRENCY ?? 6),
+    STATS_LIMITS.concurrency.min,
+    STATS_LIMITS.concurrency.max,
+    6
+  )
+  const cacheTtlMs = clampInt(
+    envInt('PEARDOCK_STATS_CACHE_TTL_MS', CONFIG.STATS?.CACHE_TTL_MS ?? 1000),
+    STATS_LIMITS.cacheTtlMs.min,
+    STATS_LIMITS.cacheTtlMs.max,
+    1000
+  )
+  const warnContainers = clampInt(
+    envInt('PEARDOCK_STATS_WARN_CONTAINERS', 80),
+    STATS_LIMITS.warnContainers.min,
+    STATS_LIMITS.warnContainers.max,
+    80
+  )
+  return {
+    collectIntervalMs,
+    broadcastIntervalMs,
+    concurrency,
+    cacheTtlMs,
+    warnContainers,
+  }
+}
+
+/** @type {ReturnType<typeof defaultStatsConfig>} */
+let runtime = defaultStatsConfig()
+let diskLoaded = false
+
+function loadStatsDisk() {
+  try {
+    if (!fs.existsSync(STATS_FILE)) return null
+    const raw = JSON.parse(fs.readFileSync(STATS_FILE, 'utf8'))
+    if (!raw || typeof raw !== 'object') return null
+    return raw
+  } catch (err) {
+    logger.warn('Failed to load stats config', { error: err.message, path: STATS_FILE })
+    return null
+  }
+}
+
+function saveStatsDisk() {
+  try {
+    const payload = {
+      version: 1,
+      updatedAt: new Date().toISOString(),
+      collectIntervalMs: runtime.collectIntervalMs,
+      broadcastIntervalMs: runtime.broadcastIntervalMs,
+      concurrency: runtime.concurrency,
+      cacheTtlMs: runtime.cacheTtlMs,
+      warnContainers: runtime.warnContainers,
+    }
+    const tmp = `${STATS_FILE}.${process.pid}.tmp`
+    fs.writeFileSync(tmp, JSON.stringify(payload, null, 2), { mode: 0o600 })
+    fs.renameSync(tmp, STATS_FILE)
+  } catch (err) {
+    logger.warn('Failed to save stats config', { error: err.message, path: STATS_FILE })
+  }
+}
+
+/**
+ * Normalize a partial config object with clamps.
+ * @param {object} partial
+ * @param {ReturnType<typeof defaultStatsConfig>} [base]
+ */
+export function normalizeStatsConfig(partial = {}, base = defaultStatsConfig()) {
+  const src = partial && typeof partial === 'object' ? partial : {}
+  // Accept seconds from UI as collectIntervalSec / broadcastIntervalSec
+  let collectMs = src.collectIntervalMs
+  if (collectMs == null && src.collectIntervalSec != null) {
+    collectMs = Number(src.collectIntervalSec) * 1000
+  }
+  let broadcastMs = src.broadcastIntervalMs
+  if (broadcastMs == null && src.broadcastIntervalSec != null) {
+    broadcastMs = Number(src.broadcastIntervalSec) * 1000
+  }
+  return {
+    collectIntervalMs: clampInt(
+      collectMs ?? base.collectIntervalMs,
+      STATS_LIMITS.collectIntervalMs.min,
+      STATS_LIMITS.collectIntervalMs.max,
+      base.collectIntervalMs
+    ),
+    broadcastIntervalMs: clampInt(
+      broadcastMs ?? base.broadcastIntervalMs,
+      STATS_LIMITS.broadcastIntervalMs.min,
+      STATS_LIMITS.broadcastIntervalMs.max,
+      base.broadcastIntervalMs
+    ),
+    concurrency: clampInt(
+      src.concurrency ?? base.concurrency,
+      STATS_LIMITS.concurrency.min,
+      STATS_LIMITS.concurrency.max,
+      base.concurrency
+    ),
+    cacheTtlMs: clampInt(
+      src.cacheTtlMs ?? base.cacheTtlMs,
+      STATS_LIMITS.cacheTtlMs.min,
+      STATS_LIMITS.cacheTtlMs.max,
+      base.cacheTtlMs
+    ),
+    warnContainers: clampInt(
+      src.warnContainers ?? base.warnContainers,
+      STATS_LIMITS.warnContainers.min,
+      STATS_LIMITS.warnContainers.max,
+      base.warnContainers
+    ),
+  }
+}
+
+/**
+ * Ensure disk overlay is applied once (env defaults first, then peardock-stats.json).
+ */
+export function ensureStatsConfigLoaded() {
+  if (diskLoaded) return runtime
+  diskLoaded = true
+  runtime = defaultStatsConfig()
+  const disk = loadStatsDisk()
+  if (disk) {
+    runtime = normalizeStatsConfig(disk, runtime)
+    logger.debug('Stats config loaded from disk', {
+      path: STATS_FILE,
+      collectIntervalMs: runtime.collectIntervalMs,
+      concurrency: runtime.concurrency,
+    })
+  }
+  return runtime
+}
+
+/**
+ * Public config snapshot for RPC / Settings UI.
+ * @returns {object}
+ */
+export function getStatsConfig() {
+  ensureStatsConfigLoaded()
+  return {
+    collectIntervalMs: runtime.collectIntervalMs,
+    broadcastIntervalMs: runtime.broadcastIntervalMs,
+    concurrency: runtime.concurrency,
+    cacheTtlMs: runtime.cacheTtlMs,
+    warnContainers: runtime.warnContainers,
+    // Convenience for UI
+    collectIntervalSec: Math.round(runtime.collectIntervalMs / 1000),
+    broadcastIntervalSec: Math.round(runtime.broadcastIntervalMs / 1000),
+    active: Boolean(intervalHandle),
+    peers: peers.size,
+    limits: { ...STATS_LIMITS },
+    path: STATS_FILE,
+  }
+}
+
+/**
+ * Apply partial config live (no process restart). Optionally persist to disk.
+ * @param {object} partial
+ * @param {{ persist?: boolean, replace?: boolean }} [opts]
+ * @returns {object} getStatsConfig()
+ */
+export function updateStatsConfig(partial = {}, opts = {}) {
+  ensureStatsConfigLoaded()
+  const base = opts.replace ? defaultStatsConfig() : { ...runtime }
+  const next = normalizeStatsConfig(partial, base)
+  const intervalChanged = next.collectIntervalMs !== runtime.collectIntervalMs
+  runtime = next
+  if (opts.persist !== false) saveStatsDisk()
+  if (intervalChanged && intervalHandle) {
+    rearmCollectInterval()
+  }
+  logger.info('Stats config updated', {
+    collectIntervalMs: runtime.collectIntervalMs,
+    broadcastIntervalMs: runtime.broadcastIntervalMs,
+    concurrency: runtime.concurrency,
+    cacheTtlMs: runtime.cacheTtlMs,
+    persist: opts.persist !== false,
+  })
+  return getStatsConfig()
+}
+
+/** Rearm setInterval after collect interval change (while peers connected). */
+function rearmCollectInterval() {
+  if (!intervalHandle) return
+  clearInterval(intervalHandle)
+  intervalHandle = setInterval(() => {
+    statsTick().catch(() => {})
+  }, runtime.collectIntervalMs)
+  if (typeof intervalHandle.unref === 'function') intervalHandle.unref()
+  logger.debug('Stats collect interval rearmed', {
+    intervalMs: runtime.collectIntervalMs,
+  })
+}
+
+export function getStatsFilePath() {
+  return STATS_FILE
+}
 
 /** @type {Record<string, object>} */
 const containerStats = {}
@@ -33,10 +280,11 @@ let dockerDownLogged = false
 let dockerBackoffUntil = 0
 /** Serialize ticks so overlapping async collects cannot pile up */
 let tickInFlight = false
+let largeFleetWarned = false
 
 /**
  * Docker Engine CPU % (same idea as `docker stats`).
- * First sample often has empty precpu_stats → 0 until the next tick.
+ * One-shot stats populate precpu_stats after ~1s wait inside the engine.
  */
 function calculateCPUPercent(stats) {
   try {
@@ -157,101 +405,52 @@ function isContainerActive(statsData) {
 }
 
 /**
- * Attach a stats stream with NDJSON line buffering.
- * @param {object} statsData
- * @param {import('dockerode').Container} container
+ * IP from listContainers payload (avoids N× inspect).
+ * @param {object} containerInfo
+ * @returns {string}
  */
-function attachStatsStream(statsData, container) {
-  let buf = ''
-
-  const onChunk = (chunk) => {
-    try {
-      buf += chunk.toString('utf8')
-      // Docker may send one or more JSON objects per chunk, newline-delimited
-      let nl
-      while ((nl = buf.indexOf('\n')) >= 0) {
-        const line = buf.slice(0, nl).trim()
-        buf = buf.slice(nl + 1)
-        if (!line) continue
-        try {
-          applyDockerStatsSample(statsData, JSON.parse(line))
-        } catch {
-          // incomplete / bad line — drop
-        }
-      }
-      // Also try parse if buffer is a complete single JSON object without trailing NL yet
-      if (buf.length > 2 && buf.startsWith('{')) {
-        try {
-          const sample = JSON.parse(buf)
-          buf = ''
-          applyDockerStatsSample(statsData, sample)
-        } catch {
-          // wait for more data
-          if (buf.length > 2_000_000) buf = '' // safety
-        }
-      }
-    } catch (err) {
-      logger.debug('stats chunk parse failed', { id: statsData.id, error: err.message })
+function ipFromListContainer(containerInfo) {
+  const nets = containerInfo?.NetworkSettings?.Networks
+  if (nets && typeof nets === 'object') {
+    for (const net of Object.values(nets)) {
+      if (net?.IPAddress) return net.IPAddress
     }
   }
-
-  container
-    .stats({ stream: true })
-    .then((statsStream) => {
-      statsData.stream = statsStream
-      statsStream.on('data', onChunk)
-      statsStream.on('error', (err) => {
-        logger.error('Stats stream error', { id: statsData.id, error: err.message })
-        statsData.stream = null
-      })
-      statsStream.on('close', () => {
-        statsData.stream = null
-      })
-      statsStream.on('end', () => {
-        statsData.stream = null
-      })
-    })
-    .catch((err) => {
-      logger.error('Failed to start stats stream', { id: statsData.id, error: err.message })
-      statsData.stream = null
-    })
+  if (containerInfo?.NetworkSettings?.IPAddress) {
+    return containerInfo.NetworkSettings.IPAddress
+  }
+  return 'No IP Assigned'
 }
 
-async function initializeContainerStats(containerInfo) {
-  const container = docker.getContainer(containerInfo.Id)
-  let ipAddress = 'No IP Assigned'
-  try {
-    const details = await container.inspect()
-    ipAddress = extractIpAddress(details)
-  } catch (err) {
-    logger.debug('inspect failed for stats', { id: containerInfo.Id, error: err.message })
+function ensureStatsEntry(containerInfo) {
+  const id = containerInfo.Id
+  let statsData = containerStats[id]
+  if (!statsData) {
+    statsData = {
+      id,
+      name: containerInfo.Names?.[0]?.replace(/^\//, '') || 'Unknown',
+      cpu: 0,
+      memory: 0,
+      memoryLimit: 0,
+      netRxRate: 0,
+      netTxRate: 0,
+      blkReadRate: 0,
+      blkWriteRate: 0,
+      netRxTotal: 0,
+      netTxTotal: 0,
+      blkReadTotal: 0,
+      blkWriteTotal: 0,
+      ip: ipFromListContainer(containerInfo),
+      updatedAt: 0,
+    }
+    containerStats[id] = statsData
+  } else {
+    // Refresh name / IP from list (no inspect)
+    const name = containerInfo.Names?.[0]?.replace(/^\//, '')
+    if (name) statsData.name = name
+    const ip = ipFromListContainer(containerInfo)
+    if (ip && ip !== 'No IP Assigned') statsData.ip = ip
   }
-
-  const statsData = {
-    id: containerInfo.Id,
-    name: containerInfo.Names?.[0]?.replace(/^\//, '') || 'Unknown',
-    cpu: 0,
-    memory: 0,
-    memoryLimit: 0,
-    netRxRate: 0,
-    netTxRate: 0,
-    blkReadRate: 0,
-    blkWriteRate: 0,
-    netRxTotal: 0,
-    netTxTotal: 0,
-    blkReadTotal: 0,
-    blkWriteTotal: 0,
-    ip: ipAddress,
-    stream: null,
-    updatedAt: 0,
-  }
-
-  // Only running containers have meaningful live stats streams
-  const state = String(containerInfo.State || '').toLowerCase()
-  if (state === 'running') {
-    attachStatsStream(statsData, container)
-  }
-
   return statsData
 }
 
@@ -265,23 +464,104 @@ function applyDockerStatsSample(statsData, sample) {
   statsData.updatedAt = now
 }
 
+/**
+ * One-shot container stats. dockerode may return a Buffer/string or a brief stream;
+ * normalize to a single JSON object.
+ * @param {string} id
+ * @returns {Promise<object|null>}
+ */
+async function fetchOneShotStats(id) {
+  const container = docker.getContainer(id)
+  const result = await container.stats({ stream: false })
+
+  // Most dockerode versions return a Promise resolving to the stats object or Buffer
+  if (result && typeof result === 'object' && !Buffer.isBuffer(result) && result.cpu_stats) {
+    return result
+  }
+
+  // Stream-like (Node Readable) — rare with stream:false, but handle safely
+  if (result && typeof result.on === 'function') {
+    return await new Promise((resolve, reject) => {
+      let buf = ''
+      const timeout = setTimeout(() => {
+        try {
+          result.destroy?.()
+        } catch {
+          // ignore
+        }
+        reject(new Error('one-shot stats timeout'))
+      }, 15_000)
+      result.on('data', (chunk) => {
+        buf += chunk.toString('utf8')
+        // First complete JSON object is enough
+        const nl = buf.indexOf('\n')
+        const slice = nl >= 0 ? buf.slice(0, nl).trim() : buf.trim()
+        if (!slice.startsWith('{')) return
+        try {
+          const sample = JSON.parse(slice)
+          clearTimeout(timeout)
+          try {
+            result.destroy?.()
+          } catch {
+            // ignore
+          }
+          resolve(sample)
+        } catch {
+          // wait for more
+        }
+      })
+      result.on('end', () => {
+        clearTimeout(timeout)
+        try {
+          resolve(JSON.parse(buf.trim()))
+        } catch (err) {
+          reject(err)
+        }
+      })
+      result.on('error', (err) => {
+        clearTimeout(timeout)
+        reject(err)
+      })
+    })
+  }
+
+  if (Buffer.isBuffer(result) || typeof result === 'string') {
+    const text = String(result).trim().split('\n').find(Boolean)
+    if (text) return JSON.parse(text)
+  }
+
+  return result && typeof result === 'object' ? result : null
+}
+
+/**
+ * Run async work over items with a fixed worker pool.
+ * @template T
+ * @param {T[]} items
+ * @param {number} concurrency
+ * @param {(item: T) => Promise<void>} worker
+ */
+async function mapPool(items, concurrency, worker) {
+  if (!items.length) return
+  let i = 0
+  const n = Math.min(concurrency, items.length)
+  async function run() {
+    while (i < items.length) {
+      const idx = i++
+      await worker(items[idx])
+    }
+  }
+  await Promise.all(Array.from({ length: n }, () => run()))
+}
+
 function destroyStatsEntry(id) {
   const statsData = containerStats[id]
   if (!statsData) return
-  if (statsData.stream) {
-    try {
-      statsData.stream.destroy()
-    } catch {
-      // ignore
-    }
-  }
   delete containerStats[id]
   statsCache.delete(id)
 }
 
 /**
- * Drop live stats stream for a container before stop/remove so Docker is not
- * held open by NDJSON stats attachments (can delay force-remove).
+ * Drop in-memory stats for a container before stop/remove.
  * @param {string} id
  */
 export function destroyStatsForContainer(id) {
@@ -299,54 +579,72 @@ export function destroyStatsForContainer(id) {
   }
 }
 
+/**
+ * List once, sample running containers with bounded one-shot stats.
+ * No perpetual streams — dockerd only does work while we request samples.
+ */
 async function collectContainerStats() {
-  // Running only for streams; we still list all so stopped ids are cleaned up
-  const running = await docker.listContainers({ all: false })
+  // Single list (all) — derive running from State; was 2× listContainers/tick
   const all = await docker.listContainers({ all: true })
-  const runningIds = new Set(running.map((c) => c.Id))
+  const running = all.filter((c) => String(c.State || '').toLowerCase() === 'running')
   const allIds = new Set(all.map((c) => c.Id))
+  const runningIds = new Set(running.map((c) => c.Id))
 
-  for (const containerInfo of running) {
-    const existing = containerStats[containerInfo.Id]
-    if (!existing) {
-      try {
-        containerStats[containerInfo.Id] = await initializeContainerStats(containerInfo)
-      } catch (err) {
-        logger.error('Failed to init stats', { id: containerInfo.Id, error: err.message })
+  if (running.length >= runtime.warnContainers && !largeFleetWarned) {
+    largeFleetWarned = true
+    logger.warn(
+      'Large running fleet for stats collection — dockerd load scales with container count',
+      {
+        running: running.length,
+        concurrency: runtime.concurrency,
+        intervalMs: runtime.collectIntervalMs,
+        hint: 'Settings → Performance, or PEARDOCK_STATS_INTERVAL_MS / PEARDOCK_STATS_CONCURRENCY',
       }
-    } else if (!existing.stream) {
-      // Was stopped / stream died — reattach
-      try {
-        attachStatsStream(existing, docker.getContainer(containerInfo.Id))
-      } catch (err) {
-        logger.debug('reattach stats failed', { id: containerInfo.Id, error: err.message })
-      }
-    }
+    )
+  }
+
+  // Ensure entries + drop gone / zero-out stopped
+  for (const containerInfo of running) {
+    ensureStatsEntry(containerInfo)
   }
 
-  // Drop stats for containers that no longer exist, or stop streams for exited ones
   for (const id of Object.keys(containerStats)) {
     if (!allIds.has(id)) {
       destroyStatsEntry(id)
       continue
     }
-    if (!runningIds.has(id) && containerStats[id]?.stream) {
-      try {
-        containerStats[id].stream.destroy()
-      } catch {
-        // ignore
+    if (!runningIds.has(id)) {
+      const s = containerStats[id]
+      if (s) {
+        s.cpu = 0
+        s.memory = 0
+        s.netRxRate = 0
+        s.netTxRate = 0
+        s.blkReadRate = 0
+        s.blkWriteRate = 0
       }
-      containerStats[id].stream = null
-      containerStats[id].cpu = 0
-      // keep last memory sample or zero for stopped
-      containerStats[id].cpu = 0
-      containerStats[id].memory = 0
     }
   }
+
+  // Bounded parallel one-shot samples (only running)
+  await mapPool(running, runtime.concurrency, async (containerInfo) => {
+    if (peers.size === 0) return
+    const statsData = containerStats[containerInfo.Id]
+    if (!statsData) return
+    try {
+      const sample = await fetchOneShotStats(containerInfo.Id)
+      if (sample) applyDockerStatsSample(statsData, sample)
+    } catch (err) {
+      logger.debug('one-shot stats failed', {
+        id: containerInfo.Id?.slice?.(0, 12),
+        error: err.message,
+      })
+    }
+  })
 }
 
 /**
- * Tear down all live stats streams and in-memory collection state.
+ * Tear down in-memory collection state.
  * Used when the last peer disconnects so Docker is not polled idle.
  */
 function destroyAllStatsEntries() {
@@ -355,10 +653,11 @@ function destroyAllStatsEntries() {
   }
   statsCache.clear()
   containerActivity.clear()
+  largeFleetWarned = false
 }
 
 /**
- * Pause collection: stop interval and drop Docker stats streams.
+ * Pause collection: stop interval and drop in-memory state.
  * Safe to call when already paused.
  */
 export function pauseStatsCollection() {
@@ -391,14 +690,14 @@ async function statsTick() {
       return
     }
 
-    if (now - lastBroadcast < STATS_BROADCAST_INTERVAL) return
+    if (now - lastBroadcast < runtime.broadcastIntervalMs) return
 
     const aggregatedStats = []
     for (const [containerId, statsData] of Object.entries(containerStats)) {
       const cached = statsCache.get(containerId)
       if (
         cached &&
-        now - cached.timestamp < STATS_CACHE_TTL &&
+        now - cached.timestamp < runtime.cacheTtlMs &&
         !isContainerActive(statsData)
       ) {
         aggregatedStats.push(cached.data)
@@ -443,8 +742,9 @@ async function statsTick() {
       lastBroadcast = now
     }
 
+    const cacheTtl = Math.max(runtime.cacheTtlMs, 100)
     for (const [id, cached] of statsCache.entries()) {
-      if (now - cached.timestamp > STATS_CACHE_TTL * 10) statsCache.delete(id)
+      if (now - cached.timestamp > cacheTtl * 10) statsCache.delete(id)
     }
     for (const [id, ts] of containerActivity.entries()) {
       if (now - ts > 60000) containerActivity.delete(id)
@@ -479,11 +779,15 @@ export function resumeStatsCollection() {
   if (!intervalHandle) {
     intervalHandle = setInterval(() => {
       statsTick().catch(() => {})
-    }, STATS_COLLECT_INTERVAL)
+    }, runtime.collectIntervalMs)
     if (typeof intervalHandle.unref === 'function') intervalHandle.unref()
-    logger.debug('Stats collection resumed', { peers: peers.size })
+    logger.debug('Stats collection resumed', {
+      peers: peers.size,
+      intervalMs: runtime.collectIntervalMs,
+      concurrency: runtime.concurrency,
+    })
   }
-  // Immediate kick so first connected client gets streams ASAP
+  // Immediate kick so first connected client gets samples ASAP
   lastBroadcast = 0
   statsTick().catch(() => {})
 }
@@ -503,6 +807,7 @@ function onPeerCountChange(size, prevSize) {
  */
 export function startStatsBroadcast() {
   if (serviceArmed) return
+  ensureStatsConfigLoaded()
   serviceArmed = true
   lastBroadcast = 0
   dockerDownLogged = false
@@ -515,7 +820,7 @@ export function startStatsBroadcast() {
   if (peers.size > 0) {
     resumeStatsCollection()
   } else {
-    // Explicit idle: no interval, no Docker stats streams
+    // Explicit idle: no interval, no Docker stats requests
     pauseStatsCollection()
     logger.debug('Stats service armed (idle until first peer)')
   }
@@ -541,3 +846,25 @@ export function stopStatsBroadcast() {
 export function isStatsCollectionActive() {
   return Boolean(intervalHandle)
 }
+
+/** Test/ops helpers — same shape as getStatsConfig subset */
+export function getStatsRuntimeConfig() {
+  ensureStatsConfigLoaded()
+  return {
+    collectIntervalMs: runtime.collectIntervalMs,
+    broadcastIntervalMs: runtime.broadcastIntervalMs,
+    concurrency: runtime.concurrency,
+    cacheTtlMs: runtime.cacheTtlMs,
+    warnContainers: runtime.warnContainers,
+  }
+}
+
+/**
+ * Test helper: reset runtime to defaults and clear disk-loaded flag.
+ * Does not delete the on-disk file.
+ */
+export function resetStatsConfigForTests() {
+  diskLoaded = false
+  runtime = defaultStatsConfig()
+  largeFleetWarned = false
+}