diff --git a/public/live.js b/public/live.js index 6a164edd..87c812f6 100644 --- a/public/live.js +++ b/public/live.js @@ -43,6 +43,7 @@ timelineScope: 3600000, // 1h default ms timelineTimestamps: [], // historical timestamps from DB for sparkline timelineFetchedScope: 0, // last fetched scope to avoid redundant fetches + replayGen: 0, // generation counter — incremented on each replay/rewind to discard stale async results }; // ROLE_COLORS loaded from shared roles.js (includes 'unknown') @@ -116,6 +117,7 @@ function vcrResumeLive() { stopReplay(); + VCR.replayGen++; // invalidate any in-flight async chunk processing VCR.playhead = -1; VCR.speed = 1; VCR.missedCount = 0; @@ -142,6 +144,8 @@ function vcrReplayFromTs(targetTs) { const fetchFrom = new Date(targetTs).toISOString(); stopReplay(); + VCR.replayGen++; + var gen = VCR.replayGen; vcrSetMode('REPLAY'); // Reload map nodes to match the replay time @@ -153,7 +157,10 @@ .then(r => r.json()) .then(data => { const pkts = data.packets || []; - const replayEntries = expandToBufferEntries(pkts); + return expandToBufferEntriesAsync(pkts); + }) + .then(function(replayEntries) { + if (gen !== VCR.replayGen) return; // stale async result — user changed mode if (replayEntries.length === 0) { vcrSetMode('PAUSED'); return; @@ -202,6 +209,8 @@ function vcrRewind(ms) { stopReplay(); + VCR.replayGen++; + var gen = VCR.replayGen; // Fetch packets from DB for the time window const now = Date.now(); const from = new Date(now - ms).toISOString(); @@ -212,8 +221,11 @@ // Prepend to buffer (avoid duplicates by ID) const existingIds = new Set(VCR.buffer.map(b => b.pkt.id).filter(Boolean)); const filtered = pkts.filter(p => !existingIds.has(p.id)); - const newEntries = expandToBufferEntries(filtered); - VCR.buffer = [...newEntries, ...VCR.buffer]; + return expandToBufferEntriesAsync(filtered); + }) + .then(function(newEntries) { + if (gen !== VCR.replayGen) return; // stale async result + VCR.buffer = [].concat(newEntries, VCR.buffer); VCR.playhead = 0; VCR.speed = 1; vcrSetMode('REPLAY'); @@ -274,15 +286,18 @@ // Get timestamp of last packet in buffer to fetch the next page const last = VCR.buffer[VCR.buffer.length - 1]; if (!last) return Promise.resolve(false); + var gen = VCR.replayGen; const since = new Date(last.ts + 1).toISOString(); // +1ms to avoid dupe return fetch(`/api/packets?limit=10000&grouped=false&expand=observations&since=${encodeURIComponent(since)}&order=asc`) .then(r => r.json()) .then(data => { const pkts = data.packets || []; if (pkts.length === 0) return false; - const newEntries = expandToBufferEntries(pkts); - VCR.buffer = VCR.buffer.concat(newEntries); - return true; + return expandToBufferEntriesAsync(pkts).then(function(newEntries) { + if (gen !== VCR.replayGen) return false; // stale + VCR.buffer = VCR.buffer.concat(newEntries); + return true; + }); }) .catch(() => false); } @@ -449,11 +464,53 @@ } // Expand a DB packet (with optional observations[]) into VCR buffer entries + /** + * Process packets into buffer entries in chunks to avoid blocking the main thread. + * Returns a Promise that resolves with the entries array. + * Each chunk processes CHUNK_SIZE packets, then yields to the event loop via setTimeout(0). + */ + var VCR_CHUNK_SIZE = 200; + function expandToBufferEntriesAsync(pkts) { + return new Promise(function(resolve) { + var entries = []; + var i = 0; + function processChunk() { + var end = Math.min(i + VCR_CHUNK_SIZE, pkts.length); + for (; i < end; i++) { + var p = pkts[i]; + if (p.observations && p.observations.length > 0) { + for (var j = 0; j < p.observations.length; j++) { + var obs = p.observations[j]; + entries.push({ + ts: new Date(obs.timestamp || p.timestamp || p.created_at).getTime(), + pkt: dbPacketToLive(Object.assign({}, p, obs, { hash: p.hash, raw_hex: p.raw_hex, decoded_json: p.decoded_json })) + }); + } + } else { + entries.push({ + ts: new Date(p.timestamp || p.created_at).getTime(), + pkt: dbPacketToLive(p) + }); + } + } + if (i < pkts.length) { + setTimeout(processChunk, 0); + } else { + resolve(entries); + } + } + processChunk(); + }); + } + + // Synchronous version kept for small datasets and backward compat (tests) function expandToBufferEntries(pkts) { - const entries = []; - for (const p of pkts) { + var entries = []; + for (var k = 0; k < pkts.length; k++) { + var p = pkts[k]; if (p.observations && p.observations.length > 0) { - for (const obs of p.observations) { + for (var j = 0; j < p.observations.length; j++) { + var obs = p.observations[j]; entries.push({ ts: new Date(obs.timestamp || p.timestamp || p.created_at).getTime(), pkt: dbPacketToLive(Object.assign({}, p, obs, { hash: p.hash, raw_hex: p.raw_hex, decoded_json: p.decoded_json })) @@ -1597,6 +1654,7 @@ window._vcrFormatTime = vcrFormatTime; window._liveDbPacketToLive = dbPacketToLive; window._liveExpandToBufferEntries = expandToBufferEntries; + window._liveExpandToBufferEntriesAsync = expandToBufferEntriesAsync; window._liveSEG_MAP = SEG_MAP; window._liveBufferPacket = bufferPacket; window._liveVCR = function() { return VCR; }; @@ -2584,7 +2642,7 @@ packetCount = 0; activeAnims = 0; nodeActivity = {}; pktTimestamps = []; feedDedup.clear(); - VCR.buffer = []; VCR.playhead = -1; VCR.mode = 'LIVE'; VCR.missedCount = 0; VCR.speed = 1; + VCR.buffer = []; VCR.playhead = -1; VCR.mode = 'LIVE'; VCR.missedCount = 0; VCR.speed = 1; VCR.replayGen = 0; } let _themeRefreshHandler = null; diff --git a/test-live.js b/test-live.js index b5676303..98d9325e 100644 --- a/test-live.js +++ b/test-live.js @@ -272,6 +272,48 @@ console.log('\n=== live.js: expandToBufferEntries ==='); }); } +// ===== expandToBufferEntriesAsync (chunked, non-blocking) ===== +console.log('\n=== live.js: expandToBufferEntriesAsync ==='); +{ + // Build a sandbox with packet-helpers loaded so expandToBufferEntries can call dbPacketToLive + const ctx = makeSandbox(); + addLiveGlobals(ctx); + loadInCtx(ctx, 'public/roles.js'); + loadInCtx(ctx, 'public/packet-helpers.js'); + try { loadInCtx(ctx, 'public/live.js'); } catch (e) { + for (const k of Object.keys(ctx.window)) ctx[k] = ctx.window[k]; + } + const expandSync = ctx.window._liveExpandToBufferEntries; + const expandAsync = ctx.window._liveExpandToBufferEntriesAsync; + assert.ok(expandAsync, '_liveExpandToBufferEntriesAsync must be exposed'); + + const pkts = []; + for (let i = 0; i < 500; i++) { + pkts.push({ + id: i, hash: 'h' + i, timestamp: new Date(1700000000000 + i * 1000).toISOString(), + decoded_json: '{"type":"GRP_TXT"}', path_json: '[]', + observations: [ + { timestamp: new Date(1700000000000 + i * 1000 + 100).toISOString(), snr: 5, observer_name: 'O1' }, + { timestamp: new Date(1700000000000 + i * 1000 + 200).toISOString(), snr: 8, observer_name: 'O2' }, + ], + }); + } + + test('sync expand handles 500 packets (1000 entries) correctly', () => { + const result = expandSync(pkts); + assert.strictEqual(result.length, 1000, '500 packets * 2 observations = 1000 entries'); + assert.strictEqual(result[0].pkt.hash, 'h0'); + assert.strictEqual(result[999].pkt.hash, 'h499'); + }); + + test('VCR_CHUNK_SIZE is defined and async function yields via setTimeout', () => { + const src = fs.readFileSync(__dirname + '/public/live.js', 'utf8'); + assert.ok(src.includes('VCR_CHUNK_SIZE'), 'VCR_CHUNK_SIZE constant must exist'); + assert.ok(src.includes('expandToBufferEntriesAsync'), 'async version must exist'); + assert.ok(src.includes('setTimeout(processChunk, 0)'), 'must yield via setTimeout between chunks'); + }); +} + // ===== SEG_MAP (7-segment display) ===== console.log('\n=== live.js: SEG_MAP ==='); {