mirror of
https://github.com/screentinker/screentinker.git
synced 2026-08-14 14:23:14 -06:00
feat(server): v4 liveness CORE pass — uniform heartbeat-ack + ack-gap + dashboard liveness + identity
Server-side keystone: the server now honors the v4 liveness contract uniformly across the MIXED fleet (v4 + old pre-v4 + disconnected), all three clients depending on it. - UNIFORM heartbeat-ack: emitted from the single shared device:heartbeat handler (uniform by construction; no per-client/per-path branch), BEFORE the auth guard so a known device's watchdog stays armed. Harmless to old clients (they ignore it). - RECONNECT-WINDOW ack-gap fix (ackableHeartbeat): ack a KNOWN device (authed socket OR a device_id that resolves) even mid-reconnect; NOT anonymous/never-authenticated sockets (degrade-safe); identity-agnostic. No state mutation before requireDeviceAuth (auth surface unchanged; device_ids are uuidv4). - DASHBOARD LIVENESS (deriveLiveness): server-derived, VERSION-AGNOSTIC Healthy/Degraded/Offline from signals every client sends (socket presence, heartbeat age, reconnect frequency); no client status-push. - IDENTITY CAPTURE (capture-don't-act): client_type/client_version/platform/contract_version columns; degrades to legacy/unknown for old clients; NEVER breaks register. - A-BUCKET FIX (QA): recordReconnect + persistIdentity gated on !isPlaylistRefresh (a ~45-60s refresh is not a reconnect/new identity — matches #134), and the identity write is change-detected — closing the WAL write-amplification (A1) and the benign-refresh -> false-"Degraded" (A2) regressions. New lib/liveness.js (pure helpers, unit-tested). 30 new tests (uniform ack, ack-gap, mixed fleet, identity capture, cross-client conformance, refresh-gate reproduce-then-prove); 366/366 total. OTA artifact-availability is a separate concern (out of scope). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
ScreenTinker
parent
80d6242806
commit
4cf156d4a0
|
|
@ -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
|
||||
|
|
|
|||
69
server/lib/liveness.js
Normal file
69
server/lib/liveness.js
Normal file
|
|
@ -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 };
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
190
server/test/v4-core-liveness.test.js
Normal file
190
server/test/v4-core-liveness.test.js
Normal file
|
|
@ -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();
|
||||
});
|
||||
99
server/test/v4-core-refresh-gate.test.js
Normal file
99
server/test/v4-core-refresh-gate.test.js
Normal file
|
|
@ -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();
|
||||
});
|
||||
|
|
@ -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
|
||||
});
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue