From 4bda49cf60a92d09f727f4e4b496265d8a2b8504 Mon Sep 17 00:00:00 2001 From: ScreenTinker Date: Tue, 30 Jun 2026 21:34:01 -0500 Subject: [PATCH] =?UTF-8?q?fix(#146)=20E:=20log/write=20self-protection=20?= =?UTF-8?q?=E2=80=94=20coalesced=20logs,=20batched=20telemetry,=20bounded?= =?UTF-8?q?=20maps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- server/config.js | 4 ++ server/lib/content-ack-limiter.js | 19 +++++++- server/lib/log-coalescer.js | 44 ++++++++++++++++++ server/lib/status-log-writer.js | 8 ++++ server/server.js | 10 +++-- server/services/loop-lag.js | 54 +++++++++++++++------- server/test/log-selfprotect.test.js | 57 ++++++++++++++++++++++++ server/test/loop-lag-integration.test.js | 1 + server/ws/deviceSocket.js | 3 +- 9 files changed, 177 insertions(+), 23 deletions(-) create mode 100644 server/lib/log-coalescer.js create mode 100644 server/test/log-selfprotect.test.js diff --git a/server/config.js b/server/config.js index 0ec940b..f90c7af 100644 --- a/server/config.js +++ b/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 — diff --git a/server/lib/content-ack-limiter.js b/server/lib/content-ack-limiter.js index 36cf8ff..3bc373f 100644 --- a/server/lib/content-ack-limiter.js +++ b/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 }; diff --git a/server/lib/log-coalescer.js b/server/lib/log-coalescer.js new file mode 100644 index 0000000..d0163ec --- /dev/null +++ b/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 }; diff --git a/server/lib/status-log-writer.js b/server/lib/status-log-writer.js index bb0ff62..8580bff 100644 --- a/server/lib/status-log-writer.js +++ b/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 diff --git a/server/server.js b/server/server.js index 2d44e91..6a2e794 100644 --- a/server/server.js +++ b/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 diff --git a/server/services/loop-lag.js b/server/services/loop-lag.js index f160ee0..1724ac0 100644 --- a/server/services/loop-lag.js +++ b/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; } diff --git a/server/test/log-selfprotect.test.js b/server/test/log-selfprotect.test.js new file mode 100644 index 0000000..08cc91e --- /dev/null +++ b/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'); +}); diff --git a/server/test/loop-lag-integration.test.js b/server/test/loop-lag-integration.test.js index 3694ed0..8dbb35a 100644 --- a/server/test/loop-lag-integration.test.js +++ b/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 }, diff --git a/server/ws/deviceSocket.js b/server/ws/deviceSocket.js index e4fac2e..7899b9b 100644 --- a/server/ws/deviceSocket.js +++ b/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; }