fix(#146) E: log/write self-protection — coalesced logs, batched telemetry, bounded maps

Don't let telemetry/logging cook the loop under a storm.

- lib/log-coalescer.js: dedup+count high-frequency lines, flush ONE summarized line per
  key per window ("[loop-lag] band=critical (x47 in 30s)"). Bounded buffer (auto-flush
  at MAX_KEYS). Applied to the loop-lag "still loaded" line (band CHANGES stay immediate),
  the per-request OTA check line, and "Device reconnected".
- loop-lag: event_loop_lag rows are BUFFERED and batch-inserted on a flush interval
  (was a synchronous INSERT per sample); the buffer is bounded (drop-oldest). Its
  retention prune now rides the Item-A chunkedDelete so this table can never repeat the
  status_log bloat-then-freeze. /api/status still reads in-memory current (real-time
  band unaffected).
- Bounded the previously un-evicted per-device Maps: content-ack limiter gets an idle
  sweep (started in server.js); status-log-writer.lastWritten is capped (drop-oldest;
  it only suppresses a redundant consecutive row, so eviction is safe).

Tests: N identical lines -> one counted line; single line verbatim; coalescer buffer
bounded under a distinct-key flood; content-ack Map swept of idle buckets.
loop-lag-integration updated for the batched-insert cadence. Suite 266/266.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/config.js

@@ -184,6 +184,10 @@ module.exports = {
   otaDownloadMaxPerWindow: parseInt(process.env.OTA_DOWNLOAD_MAX_PER_WINDOW) || 120,
   otaDownloadWindowMs: parseInt(process.env.OTA_DOWNLOAD_WINDOW_MS) || 60000,
   otaApkRefreshMs: parseInt(process.env.OTA_APK_REFRESH_MS) || 60000,
+  // #146 Item E — coalescing log flush + batched event_loop_lag telemetry.
+  logCoalesceFlushMs: parseInt(process.env.LOG_COALESCE_FLUSH_MS) || 30000,
+  lagFlushMs: parseInt(process.env.LAG_FLUSH_MS) || 10000,
+  lagBufferMax: parseInt(process.env.LAG_BUFFER_MAX) || 2000,
   // #146 device_status_log write batching (lib/status-log-writer.js). Status
   // transitions are buffered and coalesced to the NET state per device per flush,
   // so a flapping device writes ~1 row/flush instead of a row per transition —

server/lib/content-ack-limiter.js

@@ -60,5 +60,22 @@ function check(deviceId, contentId, status, band = 'normal', now = Date.now()) {
   return { action: 'pass' };
 }
 
+// #146 Item E: evict idle per-device buckets so this Map can't grow unbounded over
+// churned device_ids (a SNAT flood minting provisioning ids inflated every un-swept
+// per-device Map). Keyed by winStart age.
+function sweep(now = Date.now()) {
+  let n = 0;
+  const idle = config.contentAckRateWindowMs * 4;
+  for (const [k, s] of state) if (now - s.winStart > idle) { state.delete(k); n++; }
+  return n;
+}
+let sweepTimer = null;
+function startSweep() {
+  if (sweepTimer) return sweepTimer;
+  sweepTimer = setInterval(() => sweep(), config.contentAckRateWindowMs * 4);
+  if (sweepTimer.unref) sweepTimer.unref();
+  return sweepTimer;
+}
+
 function reset() { state.clear(); } // tests
-module.exports = { check, reset };
+module.exports = { check, reset, sweep, startSweep, _size: () => state.size };

server/lib/log-coalescer.js

@@ -0,0 +1,44 @@
+'use strict';
+// #146 Item E — coalescing log buffer. Under a storm, high-frequency lines (loop-lag
+// band, per-request OTA checks, repetitive "Device reconnected") turn console.log —
+// which is a SYNCHRONOUS stdout write — into its own event-loop hog. This dedups by key
+// + counts, flushing ONE summarized line per key per interval:
+//   `[loop-lag] band=critical (x47 in 30s)`
+// Trading a little bounded RAM for loop safety is explicitly desired. The buffer is
+// BOUNDED (MAX_KEYS): if it fills, we flush immediately rather than grow.
+
+const MAX_KEYS = 500;
+
+// key -> { count, sample (the latest full line), warn }
+const buf = new Map();
+let flushMs = 30000;
+let timer = null;
+
+// record(key, line, {warn}) — `key` collapses repeats; `line` is the human text to emit.
+function record(key, line, opts = {}) {
+  let e = buf.get(key);
+  if (!e) {
+    if (buf.size >= MAX_KEYS) flush();           // bounded: never grow past MAX_KEYS
+    e = { count: 0, sample: line, warn: !!opts.warn };
+    buf.set(key, e);
+  }
+  e.count += 1;
+  e.sample = line;                                // keep the most recent detail
+  e.warn = e.warn || !!opts.warn;
+}
+
+function flush() {
+  for (const [, e] of buf) {
+    const line = e.count > 1 ? `${e.sample} (x${e.count} in ${Math.round(flushMs / 1000)}s)` : e.sample;
+    (e.warn ? console.warn : console.log)(line);
+  }
+  buf.clear();
+}
+
+function start(ms) {
+  if (ms) flushMs = ms;
+  if (!timer) { timer = setInterval(flush, flushMs); if (timer.unref) timer.unref(); }
+}
+
+function reset() { buf.clear(); }                 // tests
+module.exports = { record, flush, start, reset, _size: () => buf.size };

server/lib/status-log-writer.js

@@ -58,6 +58,14 @@ function flush() {
       }
     });
     writeAll(batch);
+    // #146 Item E: bound lastWritten so it can't grow unbounded over churned device_ids.
+    // It only suppresses a redundant consecutive same-status row, so evicting the oldest
+    // entries is safe (worst case: one extra row later). Keep it to the newest ~5k ids.
+    if (lastWritten.size > 5000) {
+      const excess = lastWritten.size - 5000;
+      let i = 0;
+      for (const k of lastWritten.keys()) { if (i++ >= excess) break; lastWritten.delete(k); }
+    }
     return batch.length;
   } catch (_) {
     // table might not exist yet (early boot) — drop silently, same as the old path

server/server.js

@@ -583,6 +583,7 @@ const otaBreaker = require('./lib/ota-breaker');
 otaBreaker.startSweep();   // #144: periodically evict idle breaker buckets so keyed state stays bounded
 require('./lib/reconnect-throttle').startSweep();   // #146: same, for the reconnect throttle's per-device buckets
 require('./lib/flap-limiter').startSweep();          // #146 Item B: evict idle flap-limiter buckets
+require('./lib/content-ack-limiter').startSweep();   // #146 Item E: evict idle content-ack buckets
 const apkCache = require('./lib/apk-cache');
 apkCache.start();                                    // #146 Item C: resolve APK path/size/mtime once + refresh on interval (no per-request fs)
 const { getBand } = require('./services/loop-lag');  // #146 Item C: critical-band download shed
@@ -725,11 +726,12 @@ app.post('/api/provision/pair', requireAuth, resolveTenancy, checkDeviceLimit, (
   res.json(updated);
 });
 
-// #146 Item C: OTA update-check log. One line per check today; Item E coalesces the
-// high-frequency lines. Never keys on IP for any decision (SNAT). Kept as a helper so
-// both the offer and no-offer paths log consistently.
+// #146 Item C/E: OTA update-check log — COALESCED (one summarized line per reason per
+// window) so a poll flood can't turn synchronous stdout writes into a loop hog. Never
+// keys on IP for any decision (SNAT).
+const logCoalescer = require('./lib/log-coalescer');
 function logOtaCheck(deviceId, client, latest, available, reason) {
-  console.log(`[ota] update check: device=${deviceId || 'none'} client=${client || 'unknown'} latest=${latest} update_available=${available} reason=${reason}`);
+  logCoalescer.record(`ota-check:${reason}:${available}`, `[ota] update check: latest=${latest} update_available=${available} reason=${reason}`);
 }
 
 // #146 Item C: GLOBAL download admission (lib/ota-download-guard) — concurrency + rate

server/services/loop-lag.js

@@ -17,6 +17,8 @@
 const { monitorEventLoopDelay } = require('perf_hooks');
 const { db } = require('../db/database');
 const config = require('../config');
+const { chunkedDelete } = require('../lib/chunked-prune');   // #146 Item E: chunked lag prune
+const logCoalescer = require('../lib/log-coalescer');        // #146 Item E: coalesced band lines
 
 const NS_PER_MS = 1e6;
 // A band releases only once p99 falls below this fraction of the band's entry
@@ -28,6 +30,7 @@ let histogram = null;
 let band = 'normal';
 let calmSamples = 0;
 let current = { mean_ms: 0, p50_ms: 0, p99_ms: 0, max_ms: 0, band: 'normal', sampled_at: 0 };
+const lagBuffer = [];   // #146 Item E: pending telemetry rows, batch-inserted on flush
 
 // Pure band-transition function (exported for deterministic unit tests). Given the
 // current band, the window p99 (ms), and the running calm-sample count, returns the
@@ -67,17 +70,20 @@ function sample() {
   [band, calmSamples] = nextBand(band, snap.p99_ms, calmSamples);
   current = { ...snap, band, sampled_at: Math.floor(Date.now() / 1000) };
 
-  try {
-    db.prepare(
-      'INSERT INTO event_loop_lag (sampled_at, mean_ms, p50_ms, p99_ms, max_ms, band) VALUES (?, ?, ?, ?, ?, ?)'
-    ).run(current.sampled_at, snap.mean_ms, snap.p50_ms, snap.p99_ms, snap.max_ms, band);
-  } catch (_) { /* table may not exist on a partially-migrated DB */ }
+  // #146 Item E: BUFFER the telemetry row (batch-inserted on the flush interval) instead
+  // of a synchronous INSERT per sample — under DB contention (a bloated table slowing
+  // writes) a per-sample INSERT is itself a per-tick loop cost. Bounded: drop the oldest
+  // if the buffer overflows (never let telemetry grow unbounded and cook the loop).
+  lagBuffer.push({ ...snap, sampled_at: current.sampled_at, band });
+  if (lagBuffer.length > config.lagBufferMax) lagBuffer.splice(0, lagBuffer.length - config.lagBufferMax);
 
-  // Observable: log whenever we're loaded or when the band changes (incl. back to
-  // normal). Healthy steady state stays quiet.
-  if (band !== 'normal' || prev !== 'normal') {
-    const tag = band !== prev ? ` (was ${prev})` : '';
-    console.log(`[loop-lag] band=${band}${tag} mean=${snap.mean_ms}ms p99=${snap.p99_ms}ms max=${snap.max_ms}ms`);
+  // Observable: a band CHANGE logs immediately; a repeated "still at band X" line is
+  // COALESCED (one summarized line per flush) so a sustained-critical storm can't turn
+  // logging into its own loop hog. Healthy steady state stays quiet.
+  if (band !== prev) {
+    console.log(`[loop-lag] band=${band} (was ${prev}) mean=${snap.mean_ms}ms p99=${snap.p99_ms}ms max=${snap.max_ms}ms`);
+  } else if (band !== 'normal') {
+    logCoalescer.record(`loop-lag:${band}`, `[loop-lag] band=${band} p99=${snap.p99_ms}ms max=${snap.max_ms}ms`);
   }
 
   // #143 global pressure valve — log ONLY the band edge (open/close), not per shed
@@ -89,11 +95,24 @@ function sample() {
   }
 }
 
-function pruneLag() {
+// #146 Item E: flush buffered telemetry rows in ONE batched transaction.
+const _insLag = db.prepare('INSERT INTO event_loop_lag (sampled_at, mean_ms, p50_ms, p99_ms, max_ms, band) VALUES (?, ?, ?, ?, ?, ?)');
+function flushLag() {
+  if (!lagBuffer.length) return;
+  const rows = lagBuffer.splice(0);
+  try {
+    db.transaction((rs) => { for (const r of rs) _insLag.run(r.sampled_at, r.mean_ms, r.p50_ms, r.p99_ms, r.max_ms, r.band); })(rows);
+  } catch (_) { /* table may not exist on a partially-migrated DB — drop the batch */ }
+}
+
+// #146 Item E: chunked prune (rides idx_event_loop_lag_sampled) so this table can never
+// repeat the status_log bloat-then-freeze. Async; callers fire-and-forget.
+const _delLag = db.prepare('DELETE FROM event_loop_lag WHERE rowid IN (SELECT rowid FROM event_loop_lag WHERE sampled_at < ? LIMIT ?)');
+async function pruneLag() {
   try {
     const cutoff = Math.floor(Date.now() / 1000) - Math.round(config.lagTelemetryRetentionDays * 86400);
-    const n = db.prepare('DELETE FROM event_loop_lag WHERE sampled_at < ?').run(cutoff).changes;
-    if (n > 0) console.log(`[loop-lag] pruned ${n} sample(s) older than ${config.lagTelemetryRetentionDays}d`);
+    const { deleted } = await chunkedDelete((lim) => _delLag.run(cutoff, lim).changes, { batch: config.statusLogPruneBatch });
+    if (deleted > 0) console.log(`[loop-lag] pruned ${deleted} sample(s) older than ${config.lagTelemetryRetentionDays}d`);
   } catch (_) { /* ignore */ }
 }
 
@@ -101,12 +120,13 @@ function startLoopLagMonitor() {
   if (histogram) return; // idempotent
   histogram = monitorEventLoopDelay({ resolution: config.lagResolutionMs });
   histogram.enable();
+  logCoalescer.start(config.logCoalesceFlushMs);           // #146 Item E: start the coalesced-log flusher
   const t1 = setInterval(sample, config.lagSampleIntervalMs);
-  pruneLag(); // sweep stale rows on boot
-  const t2 = setInterval(pruneLag, config.lagPruneIntervalMs);
+  const t3 = setInterval(flushLag, config.lagFlushMs);      // #146 Item E: batch-insert buffered telemetry
+  pruneLag().catch(() => {});                               // sweep stale rows on boot (chunked, async)
+  const t2 = setInterval(() => pruneLag().catch(() => {}), config.lagPruneIntervalMs);
   // Don't keep the process alive on these timers (matters for tests / clean exit).
-  if (t1.unref) t1.unref();
-  if (t2.unref) t2.unref();
+  for (const t of [t1, t2, t3]) if (t.unref) t.unref();
 }
 
 function getBand() { return band; }

server/test/log-selfprotect.test.js

@@ -0,0 +1,57 @@
+'use strict';
+
+// #146 hardening (Item E) — log/write self-protection. The coalescer collapses N
+// identical high-frequency lines into ONE summarized line, and its buffer is bounded.
+// The content-ack limiter's per-device Map is now swept (no unbounded growth).
+
+const os = require('node:os');
+const path = require('node:path');
+const crypto = require('node:crypto');
+process.env.DATA_DIR = path.join(os.tmpdir(), 'st-logself-' + crypto.randomBytes(4).toString('hex'));
+
+const { test } = require('node:test');
+const assert = require('node:assert/strict');
+const coalescer = require('../lib/log-coalescer');
+const contentAck = require('../lib/content-ack-limiter');
+
+function capture(fn) {
+  const lines = [];
+  const rl = console.log, rw = console.warn;
+  console.log = (...a) => lines.push(a.join(' '));
+  console.warn = (...a) => lines.push(a.join(' '));
+  try { fn(); } finally { console.log = rl; console.warn = rw; }
+  return lines;
+}
+
+test('N identical lines in a window emit ONE summarized line with the count', () => {
+  coalescer.reset();
+  for (let i = 0; i < 47; i++) coalescer.record('loop-lag:critical', '[loop-lag] band=critical p99=1502ms');
+  const out = capture(() => coalescer.flush());
+  assert.equal(out.length, 1, 'coalesced to a single line');
+  assert.match(out[0], /x47/, 'shows the count');
+  assert.match(out[0], /band=critical/, 'keeps the sample text');
+});
+
+test('a single occurrence logs verbatim (no count suffix)', () => {
+  coalescer.reset();
+  coalescer.record('k', 'one-off line');
+  const out = capture(() => coalescer.flush());
+  assert.deepEqual(out, ['one-off line']);
+});
+
+test('buffer stays BOUNDED under a flood of distinct keys (auto-flush at cap)', () => {
+  coalescer.reset();
+  capture(() => { for (let i = 0; i < 5000; i++) coalescer.record('key-' + i, 'line ' + i); });
+  assert.ok(coalescer._size() <= 500, `buffer bounded (<=500), was ${coalescer._size()}`);
+});
+
+test('content-ack limiter Map is swept of idle buckets (no unbounded growth)', () => {
+  contentAck.reset();
+  const now = 0;
+  contentAck.check('dev-a', 'c1', 'ready', 'normal', now);
+  contentAck.check('dev-b', 'c1', 'ready', 'normal', now);
+  assert.ok(contentAck._size() >= 2);
+  const swept = contentAck.sweep(10 * 300000);   // far past the idle window
+  assert.ok(swept >= 2, 'idle buckets evicted');
+  assert.equal(contentAck._size(), 0, 'map drained');
+});

server/test/loop-lag-integration.test.js

@@ -26,6 +26,7 @@ before(async () => {
     env: {
       ...process.env, DATA_DIR, SELF_HOSTED: 'true', PORT: String(PORT), NODE_ENV: 'test',
       LAG_SAMPLE_INTERVAL_MS: '200',          // sample fast
+      LAG_FLUSH_MS: '200',                    // #146 Item E: batch-insert fast so persistence is observable in-test
       LAG_TELEMETRY_RETENTION_DAYS: '0.00001', // ~0.86s retention
       LAG_PRUNE_INTERVAL_MS: '400',           // prune often
     },

server/ws/deviceSocket.js

@@ -12,6 +12,7 @@ const statusLogWriter = require('../lib/status-log-writer');
 const { protectSocket } = require('../lib/safe-socket');
 const flapLimiter = require('../lib/flap-limiter');
 const { resolveIdentity } = require('../lib/device-identity');
+const logCoalescer = require('../lib/log-coalescer');
 const loopLag = require('../services/loop-lag');
 
 // Debounce window for marking a device offline on socket disconnect. Brief
@@ -554,7 +555,7 @@ module.exports = function setupDeviceSocket(io) {
           emitToDeviceWorkspace(dashboardNs, device_id, 'dashboard:device-status', { device_id, status: 'online' });
           // Only log a genuine reconnect (new socket). Same-socket periodic refreshes stay
           // quiet so the log reflects real connection events, not the 45s refresh cadence.
-          if (!isPlaylistRefresh) console.log(`Device reconnected: ${device_id}`);
+          if (!isPlaylistRefresh) logCoalescer.record('device-reconnected', `Device reconnected: ${device_id}`);
           return;
         }