From e1ce36b2a8daa2ebdbd2a882b41649860f4e4ab6 Mon Sep 17 00:00:00 2001 From: ScreenTinker Date: Thu, 2 Jul 2026 19:12:46 -0500 Subject: [PATCH] =?UTF-8?q?fix(#148)=20patch2:=20per-device=20session-sett?= =?UTF-8?q?le=20debounce=20=E2=80=94=20absorb=20duplicate-socket=20storms?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Field-safe SERVER net. A device opening duplicate/rapid sockets (the APK duplicate-socket bug, separate track) currently churns through evictions during the reconnect-throttle's 30s post-restart WARM-UP (only the hard ceiling 20 applies then, so an 8-in-9s burst passes undamped and each new socket evicts the prior). This makes the server absorb it: a thrashing PAIRED device converges to ONE stable connection and stays online. - lib/session-settle.js (decision only; bounded, swept): shouldHold(deviceId, incumbentAlive) — true only when a socket was accepted for this device within SESSION_SETTLE_WINDOW_MS (config, default 2500ms) AND the incumbent is alive. Warm-up-independent. - deviceSocket register gate (just before evictPriorSocket): if a LIVE incumbent exists and we're inside the window, SOFT-REFUSE the new socket (device:throttled reason=session_settle + disconnect) and keep the incumbent; else accept + evict + (re)arm the window. - LIVENESS SAFEGUARD (load-bearing): only hold when the incumbent socket is actually in the /device namespace — a dead/half-open incumbent is replaced, NEVER stranding the device (max hold is the 2.5s window from the incumbent's accept, then any new socket is accepted). - Soft refusal, NEVER a quarantine (reuses patch1's paired-safe philosophy); single-session enforcement intact for a legitimate move; unpaired/abusive flapping still caught by the existing limiters. O(1), no loop impact. Tests (liveness first-class): live incumbent holds + DEAD incumbent replaced (not stranded); storm of 6 sockets converges to ONE, stays online, not quarantined (during warm-up); single- session move past the window replaces cleanly; unit decision + bounded sweep. The evicted-socket-rearm test shrinks its settle window so it still exercises the eviction path. Suite 336/336. --- server/config.js | 6 ++ server/lib/session-settle.js | 48 +++++++++++ server/server.js | 1 + server/test/148-eviction-storm.test.js | 101 +++++++++++++++++++++++ server/test/evicted-socket-rearm.test.js | 6 ++ server/test/session-settle.test.js | 41 +++++++++ server/ws/deviceSocket.js | 20 +++++ 7 files changed, 223 insertions(+) create mode 100644 server/lib/session-settle.js create mode 100644 server/test/148-eviction-storm.test.js create mode 100644 server/test/session-settle.test.js diff --git a/server/config.js b/server/config.js index 769e05c..8bf7344 100644 --- a/server/config.js +++ b/server/config.js @@ -173,6 +173,12 @@ module.exports = { // only the hard ceiling (no rate-band throttle) so a deploy can't throttle // healthy screens. Throttle state is in-memory and resets on restart. reconnectWarmupMs: parseInt(process.env.RECONNECT_WARMUP_MS) || 30000, + // #148 patch2: per-device session-settle debounce window. A device opening duplicate/rapid + // sockets within this window keeps its LIVE incumbent and the duplicate is soft-refused, so + // it converges on one connection and stays online (closes the reconnect-throttle warm-up + // gap). Warm-up-independent. ~2-3s: long enough to swallow a burst, short enough that a + // genuine move (after the incumbent is gone / the window passes) is accepted within seconds. + sessionSettleWindowMs: parseInt(process.env.SESSION_SETTLE_WINDOW_MS) || 2500, reconnectBandElevatedMult: parseFloat(process.env.RECONNECT_BAND_ELEVATED_MULT) || 2, reconnectBandCriticalMult: parseFloat(process.env.RECONNECT_BAND_CRITICAL_MULT) || 4, diff --git a/server/lib/session-settle.js b/server/lib/session-settle.js new file mode 100644 index 0000000..23d9a3e --- /dev/null +++ b/server/lib/session-settle.js @@ -0,0 +1,48 @@ +'use strict'; + +// #148 patch2 — per-device SESSION-SETTLE debounce (the field-safe net). When a device opens +// duplicate/rapid sockets in a burst, keep it on ONE live incumbent connection and soft-refuse +// the duplicates, so it converges and STAYS ONLINE — instead of churning through evictions. +// This closes the gap the reconnect-throttle's 30s post-restart warm-up leaves open (during +// warm-up only the hard ceiling applies, so an 8-in-9s duplicate burst passes undamped and +// each new socket evicts the prior). This debounce is warm-up-INDEPENDENT. +// +// DECISION ONLY: this module never touches sockets or the DB. The caller supplies whether the +// incumbent is actually alive (the LIVENESS SAFEGUARD) and does the refuse/disconnect. Bounded: +// one small timestamp per device_id, idle entries swept. + +const config = require('../config'); + +const state = new Map(); // device_id -> lastAcceptedMs + +// Should this NEW socket be soft-refused (keep the incumbent)? TRUE only when a socket was +// accepted for this device within the settle window AND the incumbent is genuinely alive. +// incumbentAlive is the load-bearing safeguard: if the incumbent is dead/half-open the caller +// passes false and we return false -> accept the new socket, never stranding the device. +function shouldHold(deviceId, incumbentAlive, now = Date.now()) { + if (!incumbentAlive) return false; + const last = state.get(deviceId) || 0; + return (now - last) < config.sessionSettleWindowMs; +} + +// Record that a socket was ACCEPTED (evicted+registered) for this device — (re)arms the window. +function accepted(deviceId, now = Date.now()) { state.set(deviceId, now); } + +// Bounded: drop entries idle well past the window. +function sweep(now = Date.now()) { + let n = 0; + for (const [k, t] of state) if (now - t > config.sessionSettleWindowMs * 4) { state.delete(k); n++; } + return n; +} +let sweepTimer = null; +function startSweep() { + if (sweepTimer) return sweepTimer; + sweepTimer = setInterval(() => sweep(), 60000); + if (sweepTimer.unref) sweepTimer.unref(); + return sweepTimer; +} + +function reset() { state.clear(); } +function _size() { return state.size; } + +module.exports = { shouldHold, accepted, sweep, startSweep, reset, _size }; diff --git a/server/server.js b/server/server.js index 29c25cb..5a43764 100644 --- a/server/server.js +++ b/server/server.js @@ -600,6 +600,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/session-settle').startSweep(); // #148 patch2: evict idle session-settle entries 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) diff --git a/server/test/148-eviction-storm.test.js b/server/test/148-eviction-storm.test.js new file mode 100644 index 0000000..1a3036a --- /dev/null +++ b/server/test/148-eviction-storm.test.js @@ -0,0 +1,101 @@ +'use strict'; + +// #148 patch2 — booted end-to-end: the session-settle debounce absorbs a device opening +// duplicate/rapid sockets. Covers the LIVENESS safeguard (critical), storm convergence during +// the warm-up window, and that single-session still works for a legitimate move. + +const { test, before, after } = require('node:test'); +const assert = require('node:assert/strict'); +const { spawn } = require('node:child_process'); +const ioClient = require('socket.io-client'); +const path = require('node:path'); const os = require('node:os'); const fs = require('node:fs'); const crypto = require('node:crypto'); + +const PORT = 3955; +const base = `http://127.0.0.1:${PORT}`; +const DATA_DIR = path.join(os.tmpdir(), 'st-storm-' + crypto.randomBytes(4).toString('hex')); +let proc; +const sleep = (ms) => new Promise(r => setTimeout(r, ms)); + +before(async () => { + const logFd = fs.openSync(path.join(os.tmpdir(), 'st-storm.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', SESSION_SETTLE_WINDOW_MS: '2500' }, + 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'); +}); +after(() => { try { proc.kill('SIGKILL'); } catch { /* */ } }); + +const connected = async () => (await (await fetch(base + '/api/status')).json()).devices_connected; + +function provision() { + return new Promise((resolve) => { + const s = ioClient(`${base}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + s.on('connect', () => s.emit('device:register', { pairing_code: String(crypto.randomInt(100000, 1000000)) })); + s.on('device:registered', (d) => resolve({ sock: s, id: d.device_id, token: d.device_token })); + setTimeout(() => resolve(null), 3000); + }); +} +// Reconnect an existing device on a fresh socket; resolve with whether it was ACCEPTED +// (device:registered) or soft-refused (device:throttled reason). +function reconnect(deviceId, token) { + return new Promise((resolve) => { + const s = ioClient(`${base}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + let done = false; const finish = (r) => { if (!done) { done = true; resolve({ sock: s, ...r }); } }; + s.on('connect', () => s.emit('device:register', { device_id: deviceId, device_token: token, device_info: {} })); + s.on('device:registered', () => finish({ accepted: true })); + s.on('device:throttled', (d) => finish({ accepted: false, reason: d.reason })); + s.on('device:auth-error', (d) => finish({ accepted: false, reason: 'auth:' + d.error })); + setTimeout(() => finish({ accepted: false, reason: 'timeout' }), 3000); + }); +} + +test('LIVENESS (critical): live incumbent holds; DEAD incumbent is replaced (never stranded)', async () => { + const p = await provision(); assert.ok(p, 'provisioned'); p.sock.close(); await sleep(150); + + const b = await reconnect(p.id, p.token); // socket B -> accepted, becomes the live incumbent + assert.equal(b.accepted, true, 'B accepted as incumbent'); + await sleep(150); + + const c = await reconnect(p.id, p.token); // socket C, rapid duplicate, incumbent B alive + assert.equal(c.accepted, false, 'C is soft-refused while B is alive'); + assert.equal(c.reason, 'session_settle', 'C refused specifically by the session-settle debounce'); + assert.equal(b.sock.connected, true, 'incumbent B is kept'); + + // LIVENESS SAFEGUARD: kill the incumbent, then a new socket MUST be accepted (not stranded). + try { b.sock.io.engine.transport.ws.terminate(); } catch { b.sock.close(); } + await sleep(600); // let the server drop B from the namespace + const d = await reconnect(p.id, p.token); // socket D, incumbent now dead + assert.equal(d.accepted, true, 'D accepted after the incumbent died — device NOT stranded offline'); + d.sock.close(); +}); + +test('STORM converges to ONE connection during warm-up, stays online, not quarantined', async () => { + const p = await provision(); assert.ok(p); p.sock.close(); await sleep(150); + // 6 sockets opened ~together for the same device_id — the storm. (Server is <30s old = + // inside the reconnect-throttle warm-up, the exact gap this closes.) + const results = await Promise.all(Array.from({ length: 6 }, () => reconnect(p.id, p.token))); + const accepted = results.filter(r => r.accepted); + const settled = results.filter(r => r.reason === 'session_settle'); + assert.equal(accepted.length, 1, 'exactly ONE socket accepted — converged, no evict<->reconnect loop'); + assert.ok(settled.length >= 4, `duplicates soft-refused via session-settle (got ${settled.length})`); + assert.ok(results.every(r => r.reason !== 'quarantined'), 'a paired device is NEVER quarantined by the debounce'); + await sleep(200); + assert.ok((await connected()) >= 1, 'device stays ONLINE on the one connection'); + results.forEach(r => { try { r.sock.close(); } catch { /* */ } }); +}); + +test('single-session intact: a legitimate move (after the window) cleanly replaces the socket', async () => { + const p = await provision(); assert.ok(p); p.sock.close(); await sleep(150); + const first = await reconnect(p.id, p.token); + assert.equal(first.accepted, true); + await sleep(2700); // past the 2500ms settle window + const moved = await reconnect(p.id, p.token); // a genuine move to a new socket + assert.equal(moved.accepted, true, 'a new connection past the window is accepted (single-session replace)'); + await sleep(300); + assert.equal(first.sock.connected, false, 'the old socket was evicted (single-session enforced)'); + moved.sock.close(); +}); diff --git a/server/test/evicted-socket-rearm.test.js b/server/test/evicted-socket-rearm.test.js index 6dcde96..caaeef7 100644 --- a/server/test/evicted-socket-rearm.test.js +++ b/server/test/evicted-socket-rearm.test.js @@ -25,6 +25,12 @@ 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')); +// #148 patch2: this test exercises the EVICTION path (orthogonal to the session-settle +// debounce). Shrink the settle window so the test's ~60ms gap between socket1 and socket2 is +// past it, i.e. socket2 is ACCEPTED and evicts socket1 (the scenario under test) rather than +// being soft-refused as a rapid duplicate. (The settle behaviour itself is covered by +// test/session-settle.test.js and test/148-eviction-storm.test.js.) +process.env.SESSION_SETTLE_WINDOW_MS = '30'; process.env.SELF_HOSTED = 'true'; process.env.NODE_ENV = 'test'; diff --git a/server/test/session-settle.test.js b/server/test/session-settle.test.js new file mode 100644 index 0000000..d758877 --- /dev/null +++ b/server/test/session-settle.test.js @@ -0,0 +1,41 @@ +'use strict'; + +// #148 patch2 — session-settle DECISION unit tests. The liveness safeguard is the load-bearing +// one: a dead incumbent must NEVER be held (else we recreate #148 by stranding the device). + +process.env.SESSION_SETTLE_WINDOW_MS = '2500'; +const { test, beforeEach } = require('node:test'); +const assert = require('node:assert/strict'); +const ss = require('../lib/session-settle'); + +beforeEach(() => ss.reset()); + +test('HOLD: live incumbent + a socket accepted within the window -> refuse the duplicate', () => { + ss.accepted('dev', 1000); + assert.equal(ss.shouldHold('dev', true, 1500), true); // 500ms into the 2500ms window, incumbent alive +}); + +test('LIVENESS SAFEGUARD (critical): a DEAD incumbent is NEVER held -> accept the new socket', () => { + ss.accepted('dev', 1000); + // Within the window, but the caller reports the incumbent is not alive -> must NOT hold, + // so the new socket is accepted and the corpse evicted (device never stranded offline). + assert.equal(ss.shouldHold('dev', false, 1500), false); +}); + +test('window elapsed -> accept (a genuine move to a new socket still works)', () => { + ss.accepted('dev', 1000); + assert.equal(ss.shouldHold('dev', true, 1000 + 2500), false); // exactly at the edge + assert.equal(ss.shouldHold('dev', true, 1000 + 5000), false); +}); + +test('first connection (no prior accept) -> accept', () => { + assert.equal(ss.shouldHold('fresh', true, 9999), false); +}); + +test('bounded: sweep drops entries idle past 4x the window', () => { + ss.accepted('a', 1000); + ss.accepted('b', 1000); + assert.equal(ss._size(), 2); + ss.sweep(1000 + 2500 * 4 + 1); + assert.equal(ss._size(), 0); +}); diff --git a/server/ws/deviceSocket.js b/server/ws/deviceSocket.js index 5097b83..f7733dc 100644 --- a/server/ws/deviceSocket.js +++ b/server/ws/deviceSocket.js @@ -11,6 +11,7 @@ const contentAckLimiter = require('../lib/content-ack-limiter'); const statusLogWriter = require('../lib/status-log-writer'); const { protectSocket } = require('../lib/safe-socket'); const flapLimiter = require('../lib/flap-limiter'); +const sessionSettle = require('../lib/session-settle'); // #148 patch2: eviction-storm debounce const { resolveIdentity } = require('../lib/device-identity'); const logCoalescer = require('../lib/log-coalescer'); const loopLag = require('../services/loop-lag'); @@ -476,6 +477,24 @@ module.exports = function setupDeviceSocket(io) { } } + // #148 patch2: SESSION-SETTLE debounce. A device opening duplicate/rapid sockets + // must converge on ONE live connection and stay online, not churn through evictions + // (the reconnect-throttle's 30s post-restart warm-up skips this — this does NOT). + // If a LIVE incumbent exists and we accepted a socket for this device within the + // settle window, soft-refuse THIS duplicate and keep the incumbent. + // LIVENESS SAFEGUARD (load-bearing): only hold when the incumbent socket is actually + // in the namespace — a dead/half-open incumbent is replaced below, NEVER stranding + // the device offline. Soft refusal (paired-safe), never a quarantine. + const priorConn = heartbeat.getConnection(device_id); + const incumbentAlive = !!(priorConn && priorConn.socketId !== socket.id && deviceNs.sockets.has(priorConn.socketId)); + if (sessionSettle.shouldHold(device_id, incumbentAlive)) { + logCoalescer.record(`settle:${device_id}`, `[settle] device ${device_id} keeping live incumbent ${priorConn.socketId}; soft-refusing duplicate ${socket.id}`); + evictedSockets.add(socket.id); // this refused socket's disconnect must NOT touch device state + socket.emit('device:throttled', { retry_after_ms: config.sessionSettleWindowMs, reason: 'session_settle' }); + process.nextTick(() => { try { socket.disconnect(true); } catch (_) { evictedSockets.delete(socket.id); } }); + return; + } + currentDeviceId = device_id; authenticated = true; // Cancel any pending offline timer - device is back in the grace window @@ -484,6 +503,7 @@ module.exports = function setupDeviceSocket(io) { pendingOfflines.delete(device_id); } evictPriorSocket(device_id, socket.id); + sessionSettle.accepted(device_id); // #148 patch2: (re)arm the settle window on an accepted connection db.prepare("UPDATE devices SET status = 'online', last_heartbeat = strftime('%s','now'), ip_address = ?, updated_at = strftime('%s','now') WHERE id = ?") .run(getClientIp(socket), device_id);