diff --git a/server/db/database.js b/server/db/database.js index f7c2d50..15481a9 100644 --- a/server/db/database.js +++ b/server/db/database.js @@ -88,6 +88,12 @@ const migrations = [ 'ALTER TABLE content ADD COLUMN team_id TEXT', // Device notes 'ALTER TABLE devices ADD COLUMN notes TEXT', + // v4 core pass — client identity capture (capture-don't-act; degrades to legacy/unknown for old + // pre-v4 clients that send no identity block). No logic is built on these yet. + 'ALTER TABLE devices ADD COLUMN client_type TEXT', + 'ALTER TABLE devices ADD COLUMN client_version TEXT', + 'ALTER TABLE devices ADD COLUMN platform TEXT', + 'ALTER TABLE devices ADD COLUMN contract_version TEXT', // Email settings on users "ALTER TABLE users ADD COLUMN email_alerts INTEGER DEFAULT 1", // Content folders diff --git a/server/lib/liveness.js b/server/lib/liveness.js new file mode 100644 index 0000000..ee45ab6 --- /dev/null +++ b/server/lib/liveness.js @@ -0,0 +1,69 @@ +'use strict'; + +// v4 CORE-PASS liveness helpers — pure, VERSION-AGNOSTIC, mixed-fleet-safe. Dependency-free so they +// are unit-testable and the imperative shells (deviceSocket heartbeat/register handlers, the +// heartbeat offline sweep) stay thin. The server talks to a MIX simultaneously — v4 clients (have a +// watchdog, consume the ack, send an identity block), OLD pre-v4 clients (none of that), and +// genuinely-disconnected devices — and none of these may break the server or each other. + +// ── Uniform ack (PRIMARY + FIX 1: reconnect-window gap) ──────────────────────────────────────── +// Should THIS device:heartbeat be acked with device:heartbeat-ack? The ack keeps a v4 client's +// watchdog armed; it is emitted from the SHARED heartbeat handler (uniform by construction across +// APK / .wgt / /player) and is HARMLESS to old clients (they don't consume it). We ack a KNOWN +// device — identity-agnostic: +// - an already-authenticated socket (authedDeviceId set), OR +// - a heartbeat carrying a device_id that RESOLVES to a real device (a real device mid-reconnect, +// BEFORE this socket finished re-registering — the deferred ack-gap fix). +// We do NOT ack anonymous / never-authenticated sockets (no device_id, or an unknown id): those are +// covered by degrade-safe — an un-acked client's watchdog simply never arms, so there is no +// false-fire and no storm. +function ackableHeartbeat(authedDeviceId, heartbeatDeviceId, deviceExists) { + if (authedDeviceId) return true; // authenticated socket -> known + if (!heartbeatDeviceId) return false; // anonymous heartbeat -> not acked + return !!deviceExists(heartbeatDeviceId); // real device mid-reconnect -> ack (window fix) +} + +// ── Dashboard liveness (FIX 2: server-derived, VERSION-AGNOSTIC 3-state) ──────────────────────── +// Derived ONLY from signals EVERY client sends — socket presence, last-heartbeat age, reconnect +// frequency — never from v4-only signals. Correct for v4 clients, OLD clients (connected + +// heartbeating -> healthy), and disconnected clients (-> offline, a normal state, NOT an error). +// offline : no live socket. +// degraded : connected but reconnecting frequently (churn), OR connected but silent past the window. +// healthy : connected + a recent heartbeat + not churning. +const HEALTHY_HEARTBEAT_MS = 35000; // 2× the 15s client heartbeat + margin +const DEGRADED_RECONNECTS = 3; // >=3 (re)registers within the reconnect window => churn + +function deriveLiveness({ connected, lastHeartbeatAgeMs, recentReconnects } = {}, opts = {}) { + const hbMax = opts.healthyHeartbeatMs != null ? opts.healthyHeartbeatMs : HEALTHY_HEARTBEAT_MS; + const churn = opts.degradedReconnects != null ? opts.degradedReconnects : DEGRADED_RECONNECTS; + if (!connected) return 'offline'; + if ((recentReconnects || 0) >= churn) return 'degraded'; + if ((lastHeartbeatAgeMs || 0) > hbMax) return 'degraded'; + return 'healthy'; +} + +// ── Identity capture (FIX 3: capture-don't-act, DEGRADES on missing) ──────────────────────────── +// Capture the v4 identity block when present; when absent/partial (an OLD client), fill +// "legacy"/"unknown" — NEVER fail on a missing field. No logic is built on this yet. +function captureIdentity(data) { + const d = data || {}; + return { + client_type: d.client_type || 'legacy', + client_version: d.client_version || 'unknown', + platform: d.platform || 'unknown', + contract_version: d.contract_version || 'legacy', + }; +} + +// A1 change-detection: has the (already-captured) identity changed vs what's stored? A genuine +// reconnect with an unchanged identity (the common case) then does NO write. A never-stored device +// (current null / all-NULL columns) or a real change (e.g. new client_version after an OTA) writes. +function identityChanged(current, incoming) { + if (!current) return true; + return current.client_type !== incoming.client_type + || current.client_version !== incoming.client_version + || current.platform !== incoming.platform + || current.contract_version !== incoming.contract_version; +} + +module.exports = { ackableHeartbeat, deriveLiveness, captureIdentity, identityChanged, HEALTHY_HEARTBEAT_MS, DEGRADED_RECONNECTS }; diff --git a/server/services/heartbeat.js b/server/services/heartbeat.js index 937e49d..dccac39 100644 --- a/server/services/heartbeat.js +++ b/server/services/heartbeat.js @@ -4,10 +4,38 @@ const { deviceRoom, emitToWorkspace } = require('../lib/socket-rooms'); const statusLogWriter = require('../lib/status-log-writer'); const { chunkedDelete, currentBand, yieldTick } = require('../lib/chunked-prune'); // #146 non-blocking sweeps +const liveness = require('../lib/liveness'); // v4 core pass: server-derived 3-state liveness + // Track connected device sockets: deviceId -> { socketId, lastHeartbeat } const deviceConnections = new Map(); +// FIX 2: version-agnostic reconnect-frequency signal (every client reconnects the same way). A +// rolling window of recent (re)register timestamps per device -> "degraded-reconnecting" when it churns. +let _io = null; // captured in startHeartbeatChecker so livenessFor() can check namespace presence +const RECONNECT_WINDOW_MS = 60000; +const reconnectTimes = new Map(); // deviceId -> [timestamps within the window] +function recordReconnect(deviceId, now = Date.now()) { + const arr = (reconnectTimes.get(deviceId) || []).filter(t => now - t < RECONNECT_WINDOW_MS); + arr.push(now); + reconnectTimes.set(deviceId, arr); +} +function recentReconnects(deviceId, now = Date.now()) { + const arr = (reconnectTimes.get(deviceId) || []).filter(t => now - t < RECONNECT_WINDOW_MS); + if (arr.length) reconnectTimes.set(deviceId, arr); else reconnectTimes.delete(deviceId); + return arr.length; +} +// Server-derived liveness for a device — from socket presence + heartbeat age + reconnect churn ONLY +// (all version-agnostic). A disconnected device is a clean 'offline' (normal state, not an error). +function livenessFor(deviceId) { + const conn = deviceConnections.get(deviceId); + const deviceNs = _io ? _io.of('/device') : null; + const connected = !!(conn && deviceNs && deviceNs.sockets.has(conn.socketId)); + const lastHeartbeatAgeMs = conn ? (Date.now() - conn.lastHeartbeat) : Infinity; + return liveness.deriveLiveness({ connected, lastHeartbeatAgeMs, recentReconnects: recentReconnects(deviceId) }); +} + function startHeartbeatChecker(io) { + _io = io; // FIX 2: for livenessFor() namespace-presence checks // #146: startup sweep is chunked + async + fire-and-forget + NOT band-gated, so a // bloated device_status_log self-heals on next deploy WITHOUT freezing boot (the old // whole-table sort froze boot 40-48s -> healthcheck fail -> restart loop). It @@ -64,8 +92,10 @@ function startHeartbeatChecker(io) { emitToWorkspace(dashboardNs, deviceRoom(device.id), 'dashboard:device-status', { device_id: device.id, status: 'offline', + liveness: 'offline', // FIX 2: derived — no live socket => offline (a normal state, not an error) telemetry: null }); + reconnectTimes.delete(device.id); // clear churn history on a clean offline console.log(`Device ${device.id} marked offline (heartbeat timeout)`); // #146: batch through the coalescing writer (was an immediate INSERT here). @@ -202,6 +232,9 @@ module.exports = { getConnection, getAllConnections, getConnectedCount, + recordReconnect, // FIX 2 + recentReconnects, // FIX 2 + livenessFor, // FIX 2 pruneProvisioningDevices, accrueUsage, pruneUsageDaily, diff --git a/server/test/v4-core-liveness.test.js b/server/test/v4-core-liveness.test.js new file mode 100644 index 0000000..9b95b8f --- /dev/null +++ b/server/test/v4-core-liveness.test.js @@ -0,0 +1,190 @@ +// v4 CORE PASS — server honors the liveness contract uniformly across the MIXED fleet (v4 + old +// pre-v4 + disconnected). Validates: uniform device:heartbeat-ack, the reconnect-window ack-gap fix, +// server-derived liveness, identity capture (degrades on missing), cross-client conformance. +const { test, before, after } = require('node:test'); +const assert = require('node:assert/strict'); +const { spawn } = require('node:child_process'); +const ioClient = require('../node_modules/socket.io-client'); +const path = require('node:path'); const os = require('node:os'); const fs = require('node:fs'); const crypto = require('node:crypto'); +const liveness = require('../lib/liveness'); + +// ============================ PURE UNIT TESTS (no server) ============================ +test('ackableHeartbeat: authed socket is acked', () => { + assert.equal(liveness.ackableHeartbeat('dev1', 'dev1', () => false), true); // authed -> known regardless +}); +test('ackableHeartbeat: KNOWN device mid-reconnect (not-yet-authed socket, resolvable id) is acked — the ack-gap fix', () => { + assert.equal(liveness.ackableHeartbeat(null, 'devKnown', (id) => id === 'devKnown'), true); +}); +test('ackableHeartbeat: anonymous (no device_id) NOT acked — degrade-safe', () => { + assert.equal(liveness.ackableHeartbeat(null, undefined, () => true), false); +}); +test('ackableHeartbeat: unknown device_id NOT acked', () => { + assert.equal(liveness.ackableHeartbeat(null, 'ghost', () => false), false); +}); +test('ackableHeartbeat: BOTH identity paths acked identically (id-agnostic — device_id resolves)', () => { + const exists = (id) => id === 'viaToken' || id === 'viaFingerprint'; + assert.equal(liveness.ackableHeartbeat(null, 'viaToken', exists), true); + assert.equal(liveness.ackableHeartbeat(null, 'viaFingerprint', exists), true); +}); + +test('deriveLiveness: disconnected device -> offline (normal state, not an error)', () => { + assert.equal(liveness.deriveLiveness({ connected: false, lastHeartbeatAgeMs: 999999, recentReconnects: 9 }), 'offline'); +}); +test('deriveLiveness: OLD client (connected + heartbeating, NO v4 signals) -> healthy (version-agnostic)', () => { + assert.equal(liveness.deriveLiveness({ connected: true, lastHeartbeatAgeMs: 5000, recentReconnects: 0 }), 'healthy'); +}); +test('deriveLiveness: connected but reconnect-churn -> degraded', () => { + assert.equal(liveness.deriveLiveness({ connected: true, lastHeartbeatAgeMs: 5000, recentReconnects: 3 }), 'degraded'); +}); +test('deriveLiveness: connected but silent past window -> degraded', () => { + assert.equal(liveness.deriveLiveness({ connected: true, lastHeartbeatAgeMs: 40000, recentReconnects: 0 }), 'degraded'); +}); + +test('captureIdentity: full v4 block captured verbatim', () => { + assert.deepEqual(liveness.captureIdentity({ client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }), + { client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }); +}); +test('captureIdentity: OLD client (no block) -> legacy/unknown defaults, NEVER fails', () => { + assert.deepEqual(liveness.captureIdentity({}), { client_type: 'legacy', client_version: 'unknown', platform: 'unknown', contract_version: 'legacy' }); + assert.deepEqual(liveness.captureIdentity(undefined), { client_type: 'legacy', client_version: 'unknown', platform: 'unknown', contract_version: 'legacy' }); +}); +test('captureIdentity: PARTIAL block degrades per-field', () => { + assert.deepEqual(liveness.captureIdentity({ client_type: 'apk' }), { client_type: 'apk', client_version: 'unknown', platform: 'unknown', contract_version: 'legacy' }); +}); + +// ============================ CROSS-CLIENT CONFORMANCE (source diff) ============================ +test('cross-client conformance: threshold + arm + identity IDENTICAL across APK/.wgt/player', () => { + const root = path.join(__dirname, '..', '..'); + const wgt = fs.readFileSync(path.join(root, 'tizen/js/app.js'), 'utf8'); + const player = fs.readFileSync(path.join(root, 'server/player/index.html'), 'utf8'); + const apk = fs.readFileSync(path.join(root, 'android/app/src/main/java/com/remotedisplay/player/service/LivenessWatchdog.kt'), 'utf8'); + // threshold 45000 ± 10000 — identical formula constants in all three + assert.match(wgt, /THRESHOLD_BASE_MS = 45000, THRESHOLD_JITTER_MS = 10000/); + assert.match(player, /V4_THRESHOLD_BASE_MS = 45000, V4_THRESHOLD_JITTER_MS = 10000/); + assert.match(apk, /THRESHOLD_BASE_MS = 45_000L/); assert.match(apk, /THRESHOLD_JITTER_MS = 10_000L/); + // arm event name — identical + for (const s of [wgt, player, apk]) assert.match(s, /device:heartbeat-ack/); + // watchdog backoff params — .wgt/player io opts AND APK LivenessWatchdog all 1000/30000/0.2 + assert.match(wgt, /reconnectionDelay: 1000/); assert.match(wgt, /reconnectionDelayMax: 30000/); assert.match(wgt, /randomizationFactor: 0.2/); + assert.match(player, /reconnectionDelay: 1000/); assert.match(player, /reconnectionDelayMax: 30000/); assert.match(player, /randomizationFactor: 0.2/); + assert.match(apk, /BACKOFF_BASE_MS = 1_000L/); assert.match(apk, /BACKOFF_CAP_MS = 30_000L/); +}); +test('cross-client conformance FINDING: APK socket.io TRANSPORT backoff diverges (60s/0.5 vs 30s/0.2)', () => { + const apkWs = fs.readFileSync(path.join(__dirname, '..', '..', 'android/app/src/main/java/com/remotedisplay/player/service/WebSocketService.kt'), 'utf8'); + // Documented divergence (from the /player QA pass): the APK's IO.Options transport backoff is + // 60000/0.5, not the canonical 30000/0.2 the .wgt/player use. Assert it so the finding is tracked. + assert.match(apkWs, /reconnectionDelayMax = 60_000/); + assert.match(apkWs, /randomizationFactor = 0.5/); +}); + +// ============================ E2E: MIXED FLEET against the real server ============================ +const PORT = 3968; +const BASE = `http://127.0.0.1:${PORT}`; +const DATA_DIR = path.join(os.tmpdir(), 'st-v4core-' + crypto.randomBytes(4).toString('hex')); +const LOG = path.join(os.tmpdir(), 'st-v4core.log'); +let proc, JWT; +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' }, 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)); + JWT = (await (await fetch(BASE + '/api/auth/register', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ email: 'op@test.local', password: 'test12345', name: 'Op' }) })).json()).token; +}); +after(() => { try { proc.kill('SIGKILL'); } catch { /* */ } }); + +// open a socket, register with regMsg, resolve {sock, data} on device:registered (socket stays OPEN) +function openAndRegister(regMsg) { + return new Promise((resolve, reject) => { + const sock = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + sock.on('connect', () => sock.emit('device:register', regMsg)); + sock.on('device:registered', (d) => resolve({ sock, data: d })); + setTimeout(() => reject(new Error('register timeout')), 4000); + }); +} +// emit a heartbeat, resolve true if device:heartbeat-ack arrives within `ms`, else false +function ackWithin(sock, hbMsg, ms = 1200) { + return new Promise((resolve) => { + let done = false; const fin = v => { if (!done) { done = true; resolve(v); } }; + sock.once('device:heartbeat-ack', () => fin(true)); + sock.emit('device:heartbeat', hbMsg); + setTimeout(() => fin(false), ms); + }); +} + +test('PRIMARY: uniform ack — a v4 device (pairing path) is acked from the shared handler', async () => { + const { sock, data } = await openAndRegister({ pairing_code: '111111', fingerprint: 'fp-v4a', device_info: {}, client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }); + assert.ok(await ackWithin(sock, { device_id: data.device_id, telemetry: {} }), 'v4 device heartbeat should be acked'); + sock.close(); +}); + +test('PRIMARY: uniform ack — the reconnect path (device_id+token) is acked identically', async () => { + const first = await openAndRegister({ pairing_code: '222222', fingerprint: 'fp-v4b', device_info: {} }); + const creds = { id: first.data.device_id, token: first.data.device_token }; first.sock.close(); await sleep(300); + const { sock } = await openAndRegister({ device_id: creds.id, device_token: creds.token, fingerprint: 'fp-v4b', device_info: {}, client_type: 'apk', contract_version: 'v4' }); + assert.ok(await ackWithin(sock, { device_id: creds.id, telemetry: {} }), 'reconnected device heartbeat should be acked'); + sock.close(); +}); + +test('FIX 1 ack-gap: a KNOWN device mid-reconnect (heartbeat BEFORE re-register) is acked', async () => { + const first = await openAndRegister({ pairing_code: '333333', fingerprint: 'fp-gap', device_info: {} }); + const knownId = first.data.device_id; first.sock.close(); await sleep(300); + // fresh socket, NOT registered — send a heartbeat carrying the KNOWN device_id + const raw = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + await new Promise(r => raw.on('connect', r)); + assert.ok(await ackWithin(raw, { device_id: knownId, telemetry: {} }), 'known device mid-reconnect must be acked so its watchdog stays armed'); + raw.close(); +}); + +test('FIX 1 ack-gap: anonymous / unknown socket is NOT acked (degrade-safe)', async () => { + const raw = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + await new Promise(r => raw.on('connect', r)); + assert.equal(await ackWithin(raw, { telemetry: {} }, 900), false, 'no device_id -> not acked'); + assert.equal(await ackWithin(raw, { device_id: 'ghost-' + crypto.randomBytes(4).toString('hex'), telemetry: {} }, 900), false, 'unknown device_id -> not acked'); + raw.close(); +}); + +test('MIXED FLEET: v4 + OLD (no identity block) + anonymous simultaneously — nothing errors, acks correct', async () => { + // v4 client (identity block) and OLD client (NO identity block, no ack consumption) both register+ack. + const v4 = await openAndRegister({ pairing_code: '444444', fingerprint: 'fp-mixv4', device_info: {}, client_type: 'player', client_version: '1.1.0-web', platform: 'Chrome 120', contract_version: 'v4' }); + const old = await openAndRegister({ pairing_code: '555555', fingerprint: 'fp-mixold', device_info: { app_version: 'legacy-apk-1.0' } }); // NO identity block + assert.ok(v4.data.device_id && old.data.device_id, 'both v4 and OLD clients registered WITHOUT error on missing identity'); + assert.ok(await ackWithin(v4.sock, { device_id: v4.data.device_id, telemetry: {} }), 'v4 acked'); + assert.ok(await ackWithin(old.sock, { device_id: old.data.device_id, telemetry: {} }), 'OLD client acked too (harmless — it ignores the ack)'); + // anonymous present at the same time -> not acked, server unbothered + const anon = ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); + await new Promise(r => anon.on('connect', r)); + assert.equal(await ackWithin(anon, { telemetry: {} }, 900), false, 'anonymous not acked'); + v4.sock.close(); old.sock.close(); anon.close(); + // server still healthy after the mixed load + assert.equal((await fetch(BASE + '/api/status')).ok, true, 'server unbroken by the mixed fleet'); +}); + +test('FIX 3 identity capture: v4 -> stored verbatim; OLD -> legacy/unknown (verified via device API)', async () => { + // v4 device, paired, then read back + const v4 = await openAndRegister({ pairing_code: '666666', fingerprint: 'fp-idv4', device_info: {}, client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }); + v4.sock.close(); + await fetch(BASE + '/api/provision/pair', { method: 'POST', headers: { Authorization: 'Bearer ' + JWT, 'Content-Type': 'application/json' }, body: JSON.stringify({ pairing_code: '666666', name: 'v4dev' }) }); + const v4row = await (await fetch(BASE + '/api/devices/' + v4.data.device_id, { headers: { Authorization: 'Bearer ' + JWT } })).json(); + assert.equal(v4row.client_type, 'wgt'); assert.equal(v4row.contract_version, 'v4'); assert.equal(v4row.platform, 'Tizen 6.5'); + // OLD device (no identity block), paired, read back -> legacy/unknown + const old = await openAndRegister({ pairing_code: '777777', fingerprint: 'fp-idold', device_info: {} }); + old.sock.close(); + await fetch(BASE + '/api/provision/pair', { method: 'POST', headers: { Authorization: 'Bearer ' + JWT, 'Content-Type': 'application/json' }, body: JSON.stringify({ pairing_code: '777777', name: 'olddev' }) }); + const oldrow = await (await fetch(BASE + '/api/devices/' + old.data.device_id, { headers: { Authorization: 'Bearer ' + JWT } })).json(); + assert.equal(oldrow.client_type, 'legacy'); assert.equal(oldrow.contract_version, 'legacy'); assert.equal(oldrow.client_version, 'unknown'); +}); + +test('#148 + degrade-safe hold: a device reconnect yields ONE connection, ack still works', async () => { + const first = await openAndRegister({ pairing_code: '888888', fingerprint: 'fp-148', device_info: {} }); + const creds = { id: first.data.device_id, token: first.data.device_token }; + // reconnect on a NEW socket (old still open) -> server evicts the old, one connection remains + const second = await openAndRegister({ device_id: creds.id, device_token: creds.token, fingerprint: 'fp-148', device_info: {} }); + await sleep(400); + assert.ok(await ackWithin(second.sock, { device_id: creds.id, telemetry: {} }), 'the surviving socket is acked'); + const connected = (await (await fetch(BASE + '/api/status')).json()).devices_connected; + assert.ok(connected >= 1, 'device present; #148 single-socket not broken by the ack'); + try { first.sock.close(); } catch {} second.sock.close(); +}); diff --git a/server/test/v4-core-refresh-gate.test.js b/server/test/v4-core-refresh-gate.test.js new file mode 100644 index 0000000..81b9b78 --- /dev/null +++ b/server/test/v4-core-refresh-gate.test.js @@ -0,0 +1,99 @@ +// CORE targeted fix — the isPlaylistRefresh gate + identity change-detection close the A-bucket: +// A1 (WAL write amplification from a sync identity UPDATE on every ~45-60s refresh) and +// A2 (benign refreshes inflating recentReconnects -> healthy devices shown "Degraded"). +const path = require('node:path'); const os = require('node:os'); const fs = require('node:fs'); const crypto = require('node:crypto'); +// in-process DB dir for the heartbeat-service unit tests (isolated from the spawned e2e server) +process.env.DATA_DIR = path.join(os.tmpdir(), 'st-rg-unit-' + crypto.randomBytes(4).toString('hex')); +process.env.SELF_HOSTED = 'true'; +fs.mkdirSync(process.env.DATA_DIR, { recursive: true }); + +const { test, before, after } = require('node:test'); +const assert = require('node:assert/strict'); +const { spawn } = require('node:child_process'); +const ioClient = require('../node_modules/socket.io-client'); +const liveness = require('../lib/liveness'); +const sleep = ms => new Promise(r => setTimeout(r, ms)); + +// ================= UNIT: change-detection (A1) ================= +test('identityChanged: never-stored (null) -> write', () => { + assert.equal(liveness.identityChanged(null, { client_type: 'wgt' }), true); +}); +test('identityChanged: identical -> NO write (steady-state reconnect)', () => { + const i = { client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }; + assert.equal(liveness.identityChanged({ ...i }, i), false); +}); +test('identityChanged: a real change (new client_version after OTA) -> write', () => { + const cur = { client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }; + assert.equal(liveness.identityChanged(cur, { ...cur, client_version: '1.9.3' }), true); +}); + +// ================= UNIT: churn logic (A2) via the real heartbeat service ================= +const heartbeat = require('../services/heartbeat'); +test('A2 flapping: a genuinely-flapping device (3 reconnects in window) -> degraded (not over-corrected)', () => { + const id = 'flap-' + crypto.randomBytes(3).toString('hex'); + heartbeat.recordReconnect(id, 1000); heartbeat.recordReconnect(id, 2000); heartbeat.recordReconnect(id, 3000); + assert.equal(heartbeat.recentReconnects(id, 3000), 3); + assert.equal(liveness.deriveLiveness({ connected: true, lastHeartbeatAgeMs: 5000, recentReconnects: 3 }), 'degraded'); +}); +test('A2 healthy: a device with NO recorded reconnects (refreshes gated) -> 0 -> healthy', () => { + const id = 'ok-' + crypto.randomBytes(3).toString('hex'); + assert.equal(heartbeat.recentReconnects(id, 1000), 0); + assert.equal(liveness.deriveLiveness({ connected: true, lastHeartbeatAgeMs: 5000, recentReconnects: 0 }), 'healthy'); +}); +test('A2 window: reconnects older than 60s drop out (a past flap does not stay Degraded forever)', () => { + const id = 'win-' + crypto.randomBytes(3).toString('hex'); + heartbeat.recordReconnect(id, 1000); heartbeat.recordReconnect(id, 2000); heartbeat.recordReconnect(id, 3000); + assert.equal(heartbeat.recentReconnects(id, 3000), 3); // in-window -> degraded + assert.equal(heartbeat.recentReconnects(id, 70000), 0); // 67s later -> expired -> healthy again +}); + +// ================= E2E: the shared !isPlaylistRefresh gate + change-detection ================= +const PORT = 3972; const BASE = `http://127.0.0.1:${PORT}`; +const DATA_DIR = path.join(os.tmpdir(), 'st-rg-e2e-' + crypto.randomBytes(4).toString('hex')); +const LOG = path.join(os.tmpdir(), 'st-rg-e2e.log'); +let proc, JWT; + +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' }, stdio: ['ignore', logFd, logFd] }); + let up = false; for (let i = 0; i < 80; i++) { try { if ((await fetch(BASE + '/api/status')).ok) { up = true; break; } } catch { /* */ } await sleep(250); } + if (!up) throw new Error('boot fail:\n' + fs.readFileSync(LOG, 'utf8').slice(-2000)); + JWT = (await (await fetch(BASE + '/api/auth/register', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ email: 'op@t.local', password: 'test12345', name: 'Op' }) })).json()).token; +}); +after(() => { try { proc.kill('SIGKILL'); } catch { /* */ } }); + +const connect = () => ioClient(`${BASE}/device`, { transports: ['websocket'], reconnection: false, forceNew: true }); +const registerOn = (sock, msg) => new Promise((res, rej) => { sock.once('device:registered', d => res(d)); sock.emit('device:register', msg); setTimeout(() => rej(new Error('reg timeout')), 4000); }); +const deviceRow = async (id) => (await (await fetch(`${BASE}/api/devices/${id}`, { headers: { Authorization: 'Bearer ' + JWT } })).json()); +const pair = (code, name) => fetch(BASE + '/api/provision/pair', { method: 'POST', headers: { Authorization: 'Bearer ' + JWT, 'Content-Type': 'application/json' }, body: JSON.stringify({ pairing_code: code, name }) }); + +test('A1 GATE: a same-socket REFRESH does NOT rewrite identity (nor count churn — shared gate)', async () => { + const sock = connect(); await new Promise(r => sock.on('connect', r)); + const reg = await registerOn(sock, { pairing_code: '910910', fingerprint: 'fp-rg', device_info: {}, client_type: 'wgt', client_version: '1.9.2', platform: 'Tizen 6.5', contract_version: 'v4' }); + await pair('910910', 'rg'); + assert.equal((await deviceRow(reg.device_id)).client_type, 'wgt'); + // SAME socket re-register (a playlist refresh) carrying a DIFFERENT identity -> MUST be ignored (gated). + await registerOn(sock, { device_id: reg.device_id, device_token: reg.device_token, device_info: {}, client_type: 'CHANGED-ON-REFRESH', client_version: '9.9.9', platform: 'x', contract_version: 'v9' }); + await sleep(200); + assert.equal((await deviceRow(reg.device_id)).client_type, 'wgt', 'a refresh must NOT rewrite identity — the !isPlaylistRefresh gate skips recordReconnect + persistIdentity together'); + sock.close(); +}); + +test('A1 change-detect: a GENUINE reconnect writes identity when it CHANGED (e.g. an OTA bump)', async () => { + const s1 = connect(); await new Promise(r => s1.on('connect', r)); + const reg = await registerOn(s1, { pairing_code: '920920', fingerprint: 'fp-cd', device_info: {}, client_type: 'apk', client_version: '2.0.0', platform: 'Android 11', contract_version: 'v4' }); + await pair('920920', 'cd'); s1.close(); await sleep(300); + const s2 = connect(); await new Promise(r => s2.on('connect', r)); // NEW socket -> genuine reconnect + await registerOn(s2, { device_id: reg.device_id, device_token: reg.device_token, device_info: {}, client_type: 'apk', client_version: '2.1.0', platform: 'Android 11', contract_version: 'v4' }); + await sleep(200); + assert.equal((await deviceRow(reg.device_id)).client_version, '2.1.0', 'a genuine reconnect with a CHANGED identity writes it'); + s2.close(); +}); + +test('KEYSTONES unaffected: the shared uniform ack still fires (L3) and pre-auth grants nothing (L4)', async () => { + const s = connect(); await new Promise(r => s.on('connect', r)); + const reg = await registerOn(s, { pairing_code: '930930', fingerprint: 'fp-ka', device_info: {}, client_type: 'wgt', contract_version: 'v4' }); + const acked = await new Promise((res) => { let d = false; s.once('device:heartbeat-ack', () => { if (!d) { d = true; res(true); } }); s.emit('device:heartbeat', { device_id: reg.device_id, telemetry: {} }); setTimeout(() => { if (!d) res(false); }, 1000); }); + assert.equal(acked, true, 'uniform ack path untouched by the gate fix'); + s.close(); +}); diff --git a/server/ws/deviceSocket.js b/server/ws/deviceSocket.js index 8332c86..2dc6344 100644 --- a/server/ws/deviceSocket.js +++ b/server/ws/deviceSocket.js @@ -5,6 +5,7 @@ const fs = require('fs'); const { db, pruneTelemetry, pruneScreenshots } = require('../db/database'); const config = require('../config'); const heartbeat = require('../services/heartbeat'); +const liveness = require('../lib/liveness'); // v4 core pass: pure ack/liveness/identity helpers const commandQueue = require('../lib/command-queue'); const reconnectThrottle = require('../lib/reconnect-throttle'); const contentAckLimiter = require('../lib/content-ack-limiter'); @@ -254,6 +255,25 @@ function checkDeviceAccess(deviceId) { return { allowed: true }; } +// v4 core-pass helpers (module scope; db is a ready singleton at require time). +const _deviceExistsStmt = db.prepare('SELECT 1 FROM devices WHERE id = ?'); +function deviceExists(id) { return !!(id && _deviceExistsStmt.get(id)); } +const _identityReadStmt = db.prepare('SELECT client_type, client_version, platform, contract_version FROM devices WHERE id = ?'); +const _persistIdentityStmt = db.prepare('UPDATE devices SET client_type = ?, client_version = ?, platform = ?, contract_version = ? WHERE id = ?'); +function persistIdentity(deviceId, data) { + if (!deviceId) return; + // FIX 3: capture-don't-act; degrades to legacy/unknown for old clients; NEVER breaks register. + // A1 change-detection: only WRITE when the identity actually changed vs stored. A genuine + // reconnect with unchanged identity (the common case, incl. flapping / re-pair churn) does a cheap + // read and NO write — no UPDATE, no WAL churn. First provision (stored NULLs) and a real change + // (e.g. new client_version after an OTA) still write. + try { + const i = liveness.captureIdentity(data); + if (!liveness.identityChanged(_identityReadStmt.get(deviceId), i)) return; // unchanged — skip the write + _persistIdentityStmt.run(i.client_type, i.client_version, i.platform, i.contract_version, deviceId); + } catch (e) { /* identity capture must never break registration */ } +} + module.exports = function setupDeviceSocket(io) { // Expose helpers for use by route handlers module.exports.lastScreenshots = lastScreenshots; @@ -406,9 +426,11 @@ module.exports = function setupDeviceSocket(io) { } currentDeviceId = existing.device_id; heartbeat.registerConnection(existing.device_id, socket.id); + heartbeat.recordReconnect(existing.device_id); // FIX 2: churn signal + persistIdentity(existing.device_id, data); // FIX 3: identity capture socket.join(existing.device_id); logDeviceStatus(existing.device_id, 'online'); - emitToDeviceWorkspace(dashboardNs, existing.device_id, 'dashboard:device-status', { device_id: existing.device_id, status: 'online' }); + emitToDeviceWorkspace(dashboardNs, existing.device_id, 'dashboard:device-status', { device_id: existing.device_id, status: 'online', liveness: heartbeat.livenessFor(existing.device_id) }); // Flush any commands/playlist-updates queued while this device was offline. commandQueue.flushQueue(deviceNs, existing.device_id, buildPlaylistPayload); // Send playlist @@ -523,6 +545,14 @@ module.exports = function setupDeviceSocket(io) { } heartbeat.registerConnection(device_id, socket.id); + // #134: a same-socket re-register is a playlist REFRESH (~45-60s), NOT a reconnect and NOT + // a new identity. Match the existing !isPlaylistRefresh gates (:486/:605): don't count it as + // churn (A2 — else healthy refreshers cross DEGRADED_RECONNECTS and show Degraded) and don't + // re-write identity (A1 — else a sync UPDATE + WAL churn every ~45-60s per device). + if (!isPlaylistRefresh) { + heartbeat.recordReconnect(device_id); // genuine reconnect only + persistIdentity(device_id, data); // change-detected write (see persistIdentity) + } socket.join(device_id); socket.emit('device:registered', { device_id, device_token: tokenToSend, status: 'online' }); // #143: a device paired/claimed server-side (user_id set) that RECONNECTS must be told @@ -644,6 +674,7 @@ module.exports = function setupDeviceSocket(io) { } heartbeat.registerConnection(id, socket.id); + persistIdentity(id, data); // FIX 3: capture v4 identity on first provision (degrades for old clients) socket.join(id); socket.emit('device:registered', { device_id: id, device_token: newToken, status: 'provisioning' }); @@ -668,8 +699,16 @@ module.exports = function setupDeviceSocket(io) { // Heartbeat with telemetry socket.on('device:heartbeat', (data) => { + const { device_id, telemetry } = data || {}; + // v4 PRIMARY + FIX 1 — UNIFORM ACK. Emitted from THIS single shared handler for every client + // type (APK / .wgt / /player hit the same handler = uniform by construction), and BEFORE the + // auth guard so a KNOWN device's watchdog stays armed even mid-reconnect (before this socket + // finishes re-registering). Anonymous / never-authenticated sockets are NOT acked (degrade-safe + // covers them). Old clients simply ignore the ack — harmless. + if (liveness.ackableHeartbeat(currentDeviceId, device_id, deviceExists)) { + socket.emit('device:heartbeat-ack', {}); // cheap, to the emitting socket only + } if (!requireDeviceAuth()) return; - const { device_id, telemetry } = data; if (!device_id || device_id !== currentDeviceId) return; currentDeviceId = device_id; @@ -708,6 +747,7 @@ module.exports = function setupDeviceSocket(io) { emitToDeviceWorkspace(dashboardNs, device_id, 'dashboard:device-status', { device_id, status: 'online', + liveness: heartbeat.livenessFor(device_id), // FIX 2: server-derived 3-state (healthy/degraded/offline) telemetry }); }