diff --git a/server/config.js b/server/config.js index feb2a4e..2c41c8e 100644 --- a/server/config.js +++ b/server/config.js @@ -127,6 +127,10 @@ module.exports = { reconnectMaxBackoffMs: parseInt(process.env.RECONNECT_MAX_BACKOFF_MS) || 60000, reconnectMaxLevel: parseInt(process.env.RECONNECT_MAX_LEVEL) || 10, reconnectReleaseMs: parseInt(process.env.RECONNECT_RELEASE_MS) || 30000, + // #146 evict idle reconnect-throttle buckets so per-device state can't grow + // unbounded over churned device_ids (the sweep ota-breaker has but the #142 + // throttle lacked). A device quiet this long has its bucket dropped. + reconnectIdleResetMs: parseInt(process.env.RECONNECT_IDLE_RESET_MS) || 60 * 60 * 1000, // Cold start: for this long after process start, lag is high while the whole // fleet reconnects at once. Treat leniently — force the 'normal' band and apply // only the hard ceiling (no rate-band throttle) so a deploy can't throttle @@ -143,6 +147,18 @@ module.exports = { // is LOWER than the old hardcoded 7 days (the reporter's bloat happened under 7d); // 2-3 days is plenty for the dashboard's 24h uptime view + diagnostics. statusLogRetentionDays: parseFloat(process.env.STATUS_LOG_RETENTION_DAYS) || 3, + // #146 HARD per-device row-count ceiling on device_status_log, enforced by the + // global sweep alongside the age delete above. Age-based retention can't bound a + // write storm (rows are all younger than the window), so a reconnect storm grew + // the table to 1.1M. This cap keeps only the newest N transitions per device, so + // the table is bounded by (devices * N) REGARDLESS of churn — and the very first + // sweep trims the existing backlog (table healthy now, not in retentionDays). + statusLogMaxRowsPerDevice: parseInt(process.env.STATUS_LOG_MAX_ROWS_PER_DEVICE) || 500, + // #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 — + // breaking the storm -> table-growth -> slow-writes -> more-lag feedback loop. + statusLogFlushMs: parseInt(process.env.STATUS_LOG_FLUSH_MS) || 1000, // #142 content-ack dedup window (deviceSocket.js). A device (esp. older apps) // can spam "content : ready" for the same item; suppress identical diff --git a/server/db/database.js b/server/db/database.js index ecc90ca..05699b6 100644 --- a/server/db/database.js +++ b/server/db/database.js @@ -764,8 +764,28 @@ function pruneStatusLog() { try { const maxAgeSec = Math.round(config.statusLogRetentionDays * 86400); const n = db.prepare("DELETE FROM device_status_log WHERE timestamp < strftime('%s','now') - ?").run(maxAgeSec).changes; - if (n > 0) console.log(`[status-log] pruned ${n} row(s) older than ${config.statusLogRetentionDays}d`); - return n; + // #146 HARD per-device row-count cap. Age alone can't bound a write storm: a + // reconnect storm writes faster than the age window expires (rows all younger + // than retentionDays), which is how prod reached 1.1M rows. Keep only the newest + // N transitions per device so the table is bounded by (devices * N) regardless + // of churn — and because this runs on startup + the heartbeat interval, the FIRST + // sweep trims the existing backlog immediately (healthy now, not in retentionDays). + const cap = config.statusLogMaxRowsPerDevice; + let capped = 0; + if (cap > 0) { + capped = db.prepare(` + DELETE FROM device_status_log + WHERE id NOT IN ( + SELECT id FROM ( + SELECT id, ROW_NUMBER() OVER (PARTITION BY device_id ORDER BY timestamp DESC, id DESC) AS rn + FROM device_status_log + ) WHERE rn <= ? + ) + `).run(cap).changes; + } + const total = n + capped; + if (total > 0) console.log(`[status-log] pruned ${n} row(s) older than ${config.statusLogRetentionDays}d + ${capped} over the ${cap}/device cap`); + return total; } catch (_) { return 0; } } diff --git a/server/lib/ota-breaker.js b/server/lib/ota-breaker.js index 967fe78..4b2db7a 100644 --- a/server/lib/ota-breaker.js +++ b/server/lib/ota-breaker.js @@ -87,7 +87,10 @@ function decide(clientVersion, latestVersion, deviceId = null, now = Date.now()) if (b.hits.length > THRESHOLD) { // looping faster than a healthy device ever would const cd = COOLDOWNS_MS[Math.min(b.level, COOLDOWNS_MS.length - 1)]; b.blockedUntil = now + cd; - b.level++; + // #146 cosmetic: cap the level counter so the log doesn't read "level 32". The + // backoff is already capped (Math.min above); the counter just shouldn't run away + // past the point where it stops affecting the cooldown. + b.level = Math.min(b.level + 1, COOLDOWNS_MS.length); b.hits = []; // require a fresh burst to re-trip after cooldown return { update_available: false, reason: 'rate-backoff', retry_after_seconds: Math.ceil(cd / 1000), log: `[ota] breaker tripped key=${key} (>${THRESHOLD} checks/${Math.round(WINDOW_MS / 1000)}s, looping) -> backoff ${Math.round(cd / 1000)}s [level ${b.level}]` }; diff --git a/server/lib/reconnect-throttle.js b/server/lib/reconnect-throttle.js index 88a5c91..684b40c 100644 --- a/server/lib/reconnect-throttle.js +++ b/server/lib/reconnect-throttle.js @@ -24,7 +24,7 @@ const config = require('../config'); const loopLag = require('../services/loop-lag'); -// deviceId -> { hits: number[], level: number, blockedUntil: ms, lastThrottleAt: ms } +// deviceId -> { hits: number[], level: number, blockedUntil: ms, lastThrottleAt: ms, lastSeen: ms } const state = new Map(); let startedAt = Date.now(); @@ -53,7 +53,12 @@ function check(deviceId, now = Date.now(), bandOverride = null) { const band = bandOverride !== null ? bandOverride : (warmup ? 'normal' : loopLag.getBand()); let s = state.get(deviceId); - if (!s) { s = { hits: [], level: 0, blockedUntil: 0, lastThrottleAt: 0 }; state.set(deviceId, s); } + if (!s) { s = { hits: [], level: 0, blockedUntil: 0, lastThrottleAt: 0, lastSeen: now }; state.set(deviceId, s); } + // #146: a device quiet longer than the idle window starts fresh — so a long-gone + // device_id never carries stale escalation, and (with sweep below) its bucket is + // reclaimed. Mirrors ota-breaker's reset-on-access + sweep pair. + if (now - s.lastSeen > config.reconnectIdleResetMs) { s.hits = []; s.level = 0; s.blockedUntil = 0; s.lastThrottleAt = 0; } + s.lastSeen = now; // Already inside an enforced backoff window: reject and escalate (tighten fast). if (now < s.blockedUntil) { @@ -89,10 +94,28 @@ function allow(s, now, band) { return { allow: true, band, level: s.level }; } +// #146: actively EVICT idle per-device buckets so keyed state can't grow unbounded +// over churned device_ids (reinstalls mint new ids; reset-on-access alone never +// deletes). The OTA breaker grew this exact sweep in #144; the #142 throttle missed +// it. Bounded, off the hot path, runs on its own interval. +function sweep(now = Date.now()) { + let n = 0; + for (const [k, s] of state) if (now - s.lastSeen > config.reconnectIdleResetMs) { state.delete(k); n++; } + if (n > 0) console.log(`[throttle] swept ${n} idle reconnect bucket(s) (idle > ${Math.round(config.reconnectIdleResetMs / 60000)}m); ${state.size} remain`); + return n; +} +let sweepTimer = null; +function startSweep() { + if (sweepTimer) return sweepTimer; + sweepTimer = setInterval(() => sweep(), config.reconnectIdleResetMs); + if (sweepTimer.unref) sweepTimer.unref(); // don't keep the process alive on this timer + return sweepTimer; +} + // Test-only: clear state and optionally rewind the warm-up origin. function __resetForTest(opts = {}) { state.clear(); if (opts.startedAt !== undefined) startedAt = opts.startedAt; } -module.exports = { check, __resetForTest }; +module.exports = { check, sweep, startSweep, _size: () => state.size, __resetForTest }; diff --git a/server/lib/status-log-writer.js b/server/lib/status-log-writer.js new file mode 100644 index 0000000..bb0ff62 --- /dev/null +++ b/server/lib/status-log-writer.js @@ -0,0 +1,79 @@ +// #146 — batched, coalescing writer for device_status_log. +// +// Before: every device status transition (online/offline/offline_timeout) did its +// own synchronous INSERT on the hot path (deviceSocket.logDeviceStatus + the +// heartbeat checker). Under a reconnect storm that is one row PER flap, which both +// (a) bloats the table — it reached 1.1M rows in prod — and (b) makes each write +// slower as the table grows, lagging status processing further. A textbook feedback +// loop, the connection-layer twin of the OTA loop #144 contained. +// +// After: transitions are buffered in memory and flushed on an interval. The buffer +// keeps only the LATEST (net) status per device, so a device that flaps +// online->offline->online within a flush window collapses to at most one row — and +// if it ends where it started, zero rows. devices.status (the dashboard's source of +// truth) is still updated immediately by the callers; only the AUDIT log is batched, +// so coalescing storm noise loses nothing the uptime view needs. +// +// State is in-memory and resets on restart (like the throttle / breaker buckets). + +const { db } = require('../db/database'); +const config = require('../config'); + +const pending = new Map(); // deviceId -> latest desired status (net state) +const lastWritten = new Map(); // deviceId -> last status actually inserted +let timer = null; + +const insertStmt = () => db.prepare('INSERT INTO device_status_log (device_id, status) VALUES (?, ?)'); +// Per-device age prune — the #146 fix for the old hardcoded 7-day window in +// deviceSocket.js (now a single source of truth: config.statusLogRetentionDays). +const pruneDeviceStmt = () => + db.prepare("DELETE FROM device_status_log WHERE device_id = ? AND timestamp < strftime('%s','now') - ?"); + +// Record a transition. Cheap and allocation-light: just remembers the latest state. +function record(deviceId, status) { + if (!deviceId || !status) return; + pending.set(deviceId, status); +} + +// Write all buffered transitions whose net state differs from what's on disk. +// Returns the number of rows actually inserted (for tests/observability). +function flush() { + if (pending.size === 0) return 0; + const batch = []; + for (const [deviceId, status] of pending) { + if (lastWritten.get(deviceId) !== status) batch.push([deviceId, status]); + } + pending.clear(); + if (batch.length === 0) return 0; + + try { + const ins = insertStmt(); + const prune = pruneDeviceStmt(); + const ageSec = Math.round(config.statusLogRetentionDays * 86400); + const writeAll = db.transaction((rows) => { + for (const [deviceId, status] of rows) { + ins.run(deviceId, status); + lastWritten.set(deviceId, status); + prune.run(deviceId, ageSec); + } + }); + writeAll(batch); + return batch.length; + } catch (_) { + // table might not exist yet (early boot) — drop silently, same as the old path + return 0; + } +} + +function start() { + if (timer) return timer; + timer = setInterval(flush, config.statusLogFlushMs); + if (timer.unref) timer.unref(); // don't keep the process alive on the flush timer + return timer; +} + +// Test-only: force a synchronous flush and clear coalescing memory. +function flushNow() { return flush(); } +function __reset() { pending.clear(); lastWritten.clear(); } + +module.exports = { record, flush, flushNow, start, __reset }; diff --git a/server/server.js b/server/server.js index 2e71aa0..1bbca82 100644 --- a/server/server.js +++ b/server/server.js @@ -22,6 +22,7 @@ function logFatalAndExit(kind, err) { const e = err instanceof Error ? err : new Error('Non-error thrown: ' + require('util').inspect(err)); process.stderr.write(`\n[FATAL ${kind}] ${new Date().toISOString()}\n${e.stack || e.message}\n`); } catch (_) { /* the death handler must never throw */ } + try { require('./lib/status-log-writer').flush(); } catch (_) { /* #146 best-effort: drain buffered audit rows before close */ } try { require('./db/database').db.close(); } catch (_) { /* best-effort WAL flush */ } process.exit(1); } @@ -580,6 +581,7 @@ app.use('/api/status', require('./routes/status')); // APK version check endpoint (public, used by devices to check for updates) 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 app.get('/api/update/check', (req, res) => { const currentVersion = req.query.version; const deviceId = req.query.device_id || null; // #144: optional; beta4+ clients send it for per-device keying diff --git a/server/services/heartbeat.js b/server/services/heartbeat.js index 441b79c..54ae960 100644 --- a/server/services/heartbeat.js +++ b/server/services/heartbeat.js @@ -1,6 +1,7 @@ const { db, pruneStatusLog } = require('../db/database'); const config = require('../config'); const { deviceRoom, emitToWorkspace } = require('../lib/socket-rooms'); +const statusLogWriter = require('../lib/status-log-writer'); // Track connected device sockets: deviceId -> { socketId, lastHeartbeat } const deviceConnections = new Map(); @@ -10,6 +11,11 @@ function startHeartbeatChecker(io) { // table immediately after a deploy), then again on each interval below. pruneStatusLog(); + // #146: start the batched device_status_log flush loop. + statusLogWriter.start(); + + const deviceNs = io.of('/device'); + setInterval(() => { const now = Date.now(); const dashboardNs = io.of('/dashboard'); @@ -19,6 +25,17 @@ function startHeartbeatChecker(io) { for (const device of onlineDevices) { const conn = deviceConnections.get(device.id); + + // #146: a device with a live, still-connected socket is UP, even if its last + // heartbeat event is stuck behind a lagged event loop. Marking it offline on a + // stale in-memory lastHeartbeat was the second false-offline cause (the screen + // is online and playing, the CMS says offline). The socket still being in the + // /device namespace is the authoritative liveness signal — trust it over the + // (possibly queued) heartbeat clock. If the socket is genuinely gone, conn is + // either absent or points at a socket no longer in the namespace, and we fall + // through to the timeout below. + if (conn && deviceNs.sockets.has(conn.socketId)) continue; + const lastBeat = conn ? conn.lastHeartbeat : (device.last_heartbeat ? device.last_heartbeat * 1000 : 0); if (now - lastBeat > config.heartbeatTimeout) { @@ -34,9 +51,8 @@ function startHeartbeatChecker(io) { }); console.log(`Device ${device.id} marked offline (heartbeat timeout)`); - try { - db.prepare('INSERT INTO device_status_log (device_id, status) VALUES (?, ?)').run(device.id, 'offline_timeout'); - } catch (_) {} + // #146: batch through the coalescing writer (was an immediate INSERT here). + statusLogWriter.record(device.id, 'offline_timeout'); } } diff --git a/server/test/evicted-socket-rearm.test.js b/server/test/evicted-socket-rearm.test.js new file mode 100644 index 0000000..6dcde96 --- /dev/null +++ b/server/test/evicted-socket-rearm.test.js @@ -0,0 +1,108 @@ +'use strict'; + +// #146 cause-1 — the evicted-socket offline-timer RE-ARM race, test-PROVEN. +// +// This is the subtler/primary false-offline cause. In the register handler, +// evictPriorSocket() runs BEFORE registerConnection() puts the NEW socket in the +// connection map. So when the evicted OLD socket's 'disconnect' fires, the map still +// points at the old socket, the stale-disconnect guard passes, and (pre-fix) it ARMS +// a fresh 5s offline timer — for a device that just reconnected. Under loop-lag that +// timer fires before the new socket's registerConnection lands and marks a live, +// just-reconnected screen offline. The fix tags the evicted socket id so its +// disconnect handler bails instead of arming a timer. +// +// In-process (not a spawned server) so we can inspect deviceSocket's internal +// pendingOfflines/evictedSockets via the __ test hooks. Neutralize the +// `if (evictedSockets.delete(socket.id)) return;` line in deviceSocket.js and this +// test goes RED (a pending offline timer survives the reconnect) — same teeth +// standard as the cause-2 mutation check. + +const { test, before, after } = require('node:test'); +const assert = require('node:assert/strict'); +const path = require('node:path'); +const os = require('node:os'); +const crypto = require('node:crypto'); + +// Isolate the DB BEFORE requiring config/database (they read env at load time). +process.env.DATA_DIR = path.join(os.tmpdir(), 'st-evict-' + crypto.randomBytes(4).toString('hex')); +process.env.SELF_HOSTED = 'true'; +process.env.NODE_ENV = 'test'; + +const http = require('node:http'); +const { Server } = require('socket.io'); +const ioClient = require('socket.io-client'); +const setupDeviceSocket = require('../ws/deviceSocket'); + +let httpServer, io, base; +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +before(async () => { + httpServer = http.createServer(); + io = new Server(httpServer); + setupDeviceSocket(io); + await new Promise((r) => httpServer.listen(0, r)); + base = `http://127.0.0.1:${httpServer.address().port}`; +}); + +after(() => { + try { setupDeviceSocket.__resetTimers(); } catch { /* */ } + try { io.close(); } catch { /* */ } + try { httpServer.close(); } catch { /* */ } +}); + +const connect = () => ioClient(`${base}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + +function provision() { + const code = String(crypto.randomInt(100000, 1000000)); + return new Promise((resolve) => { + const s = connect(); + s.on('connect', () => s.emit('device:register', { pairing_code: code })); + s.on('device:registered', (d) => resolve({ sock: s, id: d.device_id, token: d.device_token })); + setTimeout(() => resolve(null), 3000); + }); +} + +// Register an existing device on a fresh socket (the genuine-reconnect path that +// triggers evictPriorSocket). Resolves true on device:registered. +function registerOn(sock, dev) { + return new Promise((resolve) => { + sock.on('device:registered', () => resolve(true)); + sock.on('connect', () => sock.emit('device:register', + { device_id: dev.id, device_token: dev.token, device_info: { app_version: 'test' } })); + setTimeout(() => resolve(false), 3000); + }); +} + +test('cause-1: a reconnect that evicts the prior socket leaves NO surviving offline timer', async () => { + const dev = await provision(); + assert.ok(dev, 'provisioned'); + dev.sock.close(); // drop the provisioning socket + await sleep(120); + + // socket1 becomes the established live connection. Its register also clears any + // pending-offline left by the provisioning socket's disconnect -> clean baseline. + const s1 = connect(); + assert.ok(await registerOn(s1, dev), 'socket1 registered'); + await sleep(60); + assert.equal(setupDeviceSocket.__pendingOfflineCount(), 0, + 'baseline: no offline timer pending after the first live registration'); + + // socket2 reconnects for the SAME device -> evictPriorSocket disconnects socket1. + // socket1's disconnect handler runs while the map still points at socket1 (the new + // socket isn't registered yet) — the exact window the cause-1 fix must cover. + const s2 = connect(); + assert.ok(await registerOn(s2, dev), 'socket2 registered (evicting socket1)'); + await sleep(300); // let socket1's eviction-disconnect process + + // THE PROOF: the evicted socket1 must NOT have armed an offline timer for a device + // that is, right now, live on socket2. Pre-fix this is true (timer armed) -> red. + assert.equal(setupDeviceSocket.__hasPendingOffline(dev.id), false, + 'no offline timer may survive the reconnect (cause-1 re-arm race)'); + + // Lifecycle (a): the eviction flag self-drains on the evicted socket's disconnect, + // so the set is bounded by in-flight evictions and cannot leak. + assert.equal(setupDeviceSocket.__evictedSize(), 0, + 'evictedSockets self-drained once the evicted socket disconnected'); + + s1.close(); s2.close(); +}); diff --git a/server/test/reconnect-storm-load.test.js b/server/test/reconnect-storm-load.test.js new file mode 100644 index 0000000..45fbb26 --- /dev/null +++ b/server/test/reconnect-storm-load.test.js @@ -0,0 +1,194 @@ +'use strict'; + +// #146 — reconnect/heartbeat STORM load harness (the connection-hot-path proof). +// +// Boots the real server and drives real websocket churn through the actual register/ +// disconnect/heartbeat path. Asserts the five guarantees beta5 must hold under load: +// (a) the reconnect breaker engages on a flapper and backs it off +// (b) a reconnecting device's offline status clears cleanly — and STAYS cleared +// (the false-offline self-reset: a live socket must never be re-marked offline) +// (c) a normal (non-flapping) reconnect is never throttled and is online immediately +// (d) status-log writes are batched/coalesced — N flaps != N row inserts +// (e) loop-lag stays bounded while the fleet churns +// +// Timings are compressed via env so the suite runs in seconds: +// HEARTBEAT_TIMEOUT=1500 + HEARTBEAT_INTERVAL=500 -> the checker decides liveness +// within ~2s, so (b)/(c) don't wait on the 45s prod timeout. +// STATUS_LOG_FLUSH_MS=300 -> batching is observable fast. +// RECONNECT_* tightened so a single-device storm trips without thousands of connects +// while a 12-device fleet herd stays under the ceiling (same shape as the #142 test). + +const { test, before, after } = require('node:test'); +const assert = require('node:assert/strict'); +const { spawn } = require('node:child_process'); +const path = require('node:path'); +const os = require('node:os'); +const fs = require('node:fs'); +const crypto = require('node:crypto'); +const ioClient = require('socket.io-client'); +const Database = require('better-sqlite3'); + +const PORT = 3997; // must be unique across the suite (files run concurrently under `node --test`) +const BASE = `http://127.0.0.1:${PORT}`; +const DATA_DIR = path.join(os.tmpdir(), 'st-storm-' + crypto.randomBytes(4).toString('hex')); +const LOG = path.join(os.tmpdir(), 'st-storm-' + crypto.randomBytes(4).toString('hex') + '.log'); +let proc, rdb; + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +before(async () => { + const logFd = fs.openSync(LOG, 'w'); + proc = spawn('node', ['server.js'], { + cwd: path.join(__dirname, '..'), + env: { + ...process.env, DATA_DIR, SELF_HOSTED: 'true', PORT: String(PORT), NODE_ENV: 'test', + HEARTBEAT_TIMEOUT: '1500', HEARTBEAT_INTERVAL: '500', + STATUS_LOG_FLUSH_MS: '300', + RECONNECT_HARD_CEILING: '8', RECONNECT_WINDOW_MS: '5000', RECONNECT_BASE_MAX: '3', + }, + stdio: ['ignore', logFd, logFd], + }); + let up = false; + for (let i = 0; i < 80; i++) { + try { const r = await fetch(BASE + '/api/status'); if (r.ok) { up = true; break; } } catch { /* */ } + await sleep(250); + } + if (!up) throw new Error('server did not boot:\n' + fs.readFileSync(LOG, 'utf8').slice(-2000)); + // Second connection to the same WAL db; SELECT-only, autocommit so each read sees + // the server's latest commit. Never writes. + rdb = new Database(path.join(DATA_DIR, 'db', 'remote_display.db')); + rdb.pragma('busy_timeout = 3000'); +}); + +after(() => { + try { rdb && rdb.close(); } catch { /* */ } + try { proc.kill('SIGKILL'); } catch { /* */ } +}); + +const statusOf = (id) => rdb.prepare('SELECT status FROM devices WHERE id = ?').get(id)?.status; +const logCount = (id) => rdb.prepare('SELECT COUNT(*) c FROM device_status_log WHERE device_id = ?').get(id).c; + +// Provision a brand-new device via a unique pairing code -> {id, token}. +function provision() { + const code = String(crypto.randomInt(100000, 1000000)); + return new Promise((resolve) => { + const sock = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + sock.on('connect', () => sock.emit('device:register', { pairing_code: code })); + sock.on('device:registered', (d) => { try { sock.close(); } catch { /* */ } resolve({ id: d.device_id, token: d.device_token }); }); + setTimeout(() => { try { sock.close(); } catch { /* */ } resolve(null); }, 4000); + }); +} + +// One genuine reconnect on a fresh socket that CLOSES right after -> {registered, throttled, retryAfterMs}. +function reconnect(dev) { + return new Promise((resolve) => { + const sock = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + let done = false; + const finish = (r) => { if (done) return; done = true; try { sock.close(); } catch { /* */ } resolve(r); }; + sock.on('connect', () => sock.emit('device:register', { device_id: dev.id, device_token: dev.token, device_info: { app_version: 'test' } })); + sock.on('device:registered', () => finish({ registered: true, throttled: false })); + sock.on('device:throttled', (m) => finish({ registered: false, throttled: true, retryAfterMs: m?.retry_after_ms })); + setTimeout(() => finish({ registered: false, throttled: false }), 1500); + }); +} + +// Genuine reconnect that KEEPS the socket open (and sends NO app-level heartbeats). +function reconnectHold(dev) { + return new Promise((resolve) => { + const sock = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + sock.on('connect', () => sock.emit('device:register', { device_id: dev.id, device_token: dev.token, device_info: { app_version: 'test' } })); + sock.on('device:registered', () => resolve(sock)); + setTimeout(() => resolve(sock), 1500); + }); +} + +// (a) breaker engages on a flapper + backs it off --------------------------------- +test('(a) a flapping device trips the reconnect breaker with a backoff', async () => { + const dev = await provision(); + assert.ok(dev, 'provisioned'); + let throttled = 0, registered = 0, sawBackoff = false; + for (let i = 0; i < 12; i++) { // 12 genuine reconnects in the 5s window > ceiling 8 + const r = await reconnect(dev); + if (r.registered) registered++; + if (r.throttled) { throttled++; if (r.retryAfterMs > 0) sawBackoff = true; } + } + assert.ok(throttled >= 1, `flapper must be throttled (got ${throttled})`); + assert.ok(sawBackoff, 'throttle must carry a positive backoff (retry_after_ms)'); + assert.ok(registered < 12, `not every flap should register (got ${registered}/12)`); +}); + +// (b) reconnecting device's offline clears AND stays clear ------------------------- +// This is the false-offline self-reset guarantee. We drive the device offline via the +// checker, reconnect, hold the socket open WITHOUT sending heartbeats, and verify it +// stays online across >2x the heartbeat timeout. Pre-fix, the checker re-marks a +// live-but-silent socket offline within ~1.5s (stuck-offline flapping); post-fix the +// live socket short-circuits the checker so it stays cleanly online. +test('(b) a reconnected device clears offline and is NOT re-marked offline while live', async () => { + const dev = await provision(); + assert.ok(dev); + + // Drive it offline: open then immediately close a socket, let the checker mark it. + const s0 = await reconnectHold(dev); + s0.close(); + let offline = false; + for (let i = 0; i < 12; i++) { await sleep(300); if (statusOf(dev.id) === 'offline') { offline = true; break; } } + assert.ok(offline, 'device should be marked offline after its socket drops'); + + // Reconnect and HOLD the socket open, sending no heartbeats. + const s1 = await reconnectHold(dev); + assert.equal(statusOf(dev.id), 'online', 'reconnect clears offline immediately'); + + // Stay live for >2x heartbeat timeout (1500ms). Must remain online the whole time. + for (let i = 0; i < 8; i++) { + await sleep(500); + assert.equal(statusOf(dev.id), 'online', `must stay online while the socket is live (tick ${i})`); + } + s1.close(); +}); + +// (c) a normal reconnect is never throttled and is online immediately -------------- +test('(c) a single normal reconnect is not throttled and clears status at once', async () => { + const dev = await provision(); + assert.ok(dev); + // make it offline first so we can see the clear + const s0 = await reconnectHold(dev); s0.close(); + let offline = false; + for (let i = 0; i < 12; i++) { await sleep(300); if (statusOf(dev.id) === 'offline') { offline = true; break; } } + assert.ok(offline, 'device offline before the clean reconnect'); + + const r = await reconnect(dev); + assert.ok(r.registered, 'normal reconnect registers'); + assert.ok(!r.throttled, 'normal reconnect is NOT throttled'); + // status went online on reconnect (reconnect() closes its socket, but the UPDATE + // to devices.status already happened during register). + assert.equal(statusOf(dev.id), 'online', 'a normal reconnect clears offline immediately'); +}); + +// (d) status-log writes are batched/coalesced ------------------------------------- +test('(d) a flap storm does NOT write one status-log row per transition', async () => { + const dev = await provision(); + assert.ok(dev); + const before = logCount(dev.id); + const FLAPS = 20; + for (let i = 0; i < FLAPS; i++) { const s = await reconnectHold(dev); s.close(); } + await sleep(1000); // let the 300ms flusher settle + const written = logCount(dev.id) - before; + assert.ok(written < FLAPS, `batched: ${written} rows for ${FLAPS} flaps must be < ${FLAPS}`); + assert.ok(written <= 6, `coalesced to net state: expected a handful of rows, got ${written}`); +}); + +// (e) loop-lag stays bounded under churn ------------------------------------------ +test('(e) loop-lag stays bounded while the fleet churns', async () => { + const fleet = []; + for (let i = 0; i < 10; i++) { const d = await provision(); if (d) fleet.push(d); } + assert.ok(fleet.length >= 8, 'fleet provisioned'); + // Two rounds of whole-fleet reconnects concurrently — a churn burst. + for (let round = 0; round < 2; round++) await Promise.all(fleet.map(reconnect)); + const r = await fetch(BASE + '/api/status'); + const body = await r.json(); + const p99 = body.loop_lag?.p99_ms; + assert.ok(typeof p99 === 'number', 'status exposes loop_lag.p99_ms'); + // Prod's runaway simmer bounced 300-1145ms with a 4345ms spike; under the bounded + // churn here it must stay well under that ceiling. + assert.ok(p99 < 1000, `loop-lag p99 must stay bounded under churn (was ${p99}ms)`); +}); diff --git a/server/ws/deviceSocket.js b/server/ws/deviceSocket.js index de9d251..230ea11 100644 --- a/server/ws/deviceSocket.js +++ b/server/ws/deviceSocket.js @@ -8,6 +8,7 @@ const heartbeat = require('../services/heartbeat'); const commandQueue = require('../lib/command-queue'); const reconnectThrottle = require('../lib/reconnect-throttle'); const contentAckLimiter = require('../lib/content-ack-limiter'); +const statusLogWriter = require('../lib/status-log-writer'); const loopLag = require('../services/loop-lag'); // Debounce window for marking a device offline on socket disconnect. Brief @@ -22,6 +23,15 @@ const loopLag = require('../services/loop-lag'); const pendingOfflines = new Map(); const OFFLINE_DEBOUNCE_MS = 5000; +// #146: socket ids we force-disconnected via evictPriorSocket because a NEWER socket +// took over the device. evictPriorSocket runs at register time BEFORE the new socket +// is put in the connection map (registerConnection is later in the same handler), so +// the evicted socket's disconnect handler would see the still-old map entry, pass the +// stale-disconnect guard, and ARM a fresh offline timer — re-marking the device that +// just reconnected offline (the self-reset race). Tagging the id here lets that +// disconnect handler bail out instead of arming a timer. Drained on consumption. +const evictedSockets = new Set(); + // Proof-of-play write throttle. A player stuck in a tight loop (e.g. a playlist // with 0-second item durations) fires device:play-event 'play_start' several // times per second; unthrottled this once bloated play_logs to ~900k rows @@ -74,12 +84,14 @@ function getClientIp(socket) { return socket.handshake.address; } +// #146: route status transitions through the batched, coalescing writer instead of +// an immediate INSERT-per-transition. A flapping device no longer writes a row per +// flap (the table-bloat feedback loop); the per-device age prune now lives in the +// writer and uses config.statusLogRetentionDays (was a hardcoded 7 days here — one +// source of truth). devices.status is still updated immediately by callers; only +// this audit log is deferred to the next flush. function logDeviceStatus(deviceId, status) { - try { - db.prepare('INSERT INTO device_status_log (device_id, status) VALUES (?, ?)').run(deviceId, status); - // Prune entries older than 7 days - db.prepare("DELETE FROM device_status_log WHERE device_id = ? AND timestamp < strftime('%s','now') - 604800").run(deviceId); - } catch (e) { /* table might not exist yet */ } + statusLogWriter.record(deviceId, status); } @@ -254,7 +266,10 @@ module.exports = function setupDeviceSocket(io) { const oldSocket = deviceNs.sockets.get(prior.socketId); if (oldSocket) { console.log(`Evicting prior socket ${prior.socketId} for device ${deviceId}`); - try { oldSocket.disconnect(true); } catch (_) {} + // Mark BEFORE disconnect: disconnect(true) fires the old socket's 'disconnect' + // handler synchronously, so the flag must already be set when it runs. + evictedSockets.add(prior.socketId); + try { oldSocket.disconnect(true); } catch (_) { evictedSockets.delete(prior.socketId); } } } @@ -772,6 +787,14 @@ module.exports = function setupDeviceSocket(io) { }); socket.on('disconnect', () => { + // #146: this socket was force-evicted by a newer registration for the same + // device. The new socket owns the device now (or is mid-register), so this + // disconnect must NOT arm an offline timer — doing so was the self-reset race + // that re-marked just-reconnected devices offline. The map-based stale guard + // below can't catch it because eviction runs before the new socket is in the + // map. Drain the flag and bail. (delete() returns true iff it was present.) + if (evictedSockets.delete(socket.id)) return; + if (!currentDeviceId) return; // Stale-disconnect guard: a newer socket already took over this device_id @@ -853,3 +876,16 @@ module.exports = function setupDeviceSocket(io) { return deviceNs; }; + +// #146 test hooks — read-only views of the internal offline-timer / eviction state, +// so the cause-1 re-arm race (evicted socket arming an offline timer for a +// just-reconnected device) is test-PROVEN, not just correct-by-construction. Prefixed +// `__` and never used by production code. +module.exports.__hasPendingOffline = (deviceId) => pendingOfflines.has(deviceId); +module.exports.__pendingOfflineCount = () => pendingOfflines.size; +module.exports.__evictedSize = () => evictedSockets.size; +module.exports.__resetTimers = () => { + for (const t of pendingOfflines.values()) clearTimeout(t); + pendingOfflines.clear(); + evictedSockets.clear(); +};