From 90050bd9ecf8cef492a79aba3f052007caf63d3a Mon Sep 17 00:00:00 2001 From: KoalaDev <6156589+Shik3i@users.noreply.github.com> Date: Thu, 20 Aug 2026 20:30:54 +0200 Subject: [PATCH] feat(sync): coalesce offline media intent --- docs/ARCHITECTURE.md | 27 +- docs/PROTOCOL.md | 25 +- extension/background.js | 228 +++++++--- .../canonical-media-state-background.test.mjs | 6 +- .../offline-media-intent-background.test.mjs | 73 ++++ extension/offline-media-intent.js | 407 ++++++++++++++++++ extension/offline-media-intent.test.mjs | 265 ++++++++++++ scripts/test-server-ws.mjs | 70 +++ tests/e2e/extension.spec.mjs | 184 ++++++++ 9 files changed, 1228 insertions(+), 57 deletions(-) create mode 100644 extension/offline-media-intent-background.test.mjs create mode 100644 extension/offline-media-intent.js create mode 100644 extension/offline-media-intent.test.mjs diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2d64677..3c86ff8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -50,8 +50,31 @@ relay media event. Force Sync remains a two-phase ACK protocol. `PREPARE` is temporary choreography; the matching `EXECUTE` commits its validated target to canonical state. Per-sender -`seq`, peer heartbeats, and the existing reconnect event queue remain separate. -Offline media-command compaction is intentionally deferred. +`seq`, peer heartbeats, and the reconnect queue remain separate mechanisms. + +## 3.2 Offline Media Intent + +Canonical Media State and Offline Media Intent have different ownership: + +- Canonical Media State is the relay's last accepted shared playback truth. +- Offline Media Intent is one room-scoped client representation of local + `PLAY`/`PAUSE`/`SEEK` commands that have not reached the relay yet. + +Contiguous offline controls merge into a bounded logical queue entry. Every +retained coordination event, including Force Sync and Episode Lobby events, is +an ordering barrier. Stale offline `PING`, `PONG`, heartbeat `PEER_STATUS`, and +`EVENT_ACK` frames are not persisted because they are no longer meaningful after +reconnect. Force Sync ACK remains transactional and is not dropped or merged. + +Media intent waits for the reconnecting room's `ROOM_DATA`. An authorized intent +takes precedence over the older canonical snapshot, materializes into the +minimum ordered legacy `SEEK` plus `PLAY`/`PAUSE` frames needed by old peers, and +thereby advances relay canonical state normally. With no intent, canonical +recovery is unchanged. Role loss discards room-driving intent before recovery; +intentional Host Control solo mode and an active Episode Lobby remain +authoritative. MV3 session restoration migrates the previous raw queue format, +preserves barriers, repairs `localSeq` monotonically, and rejects another room's +intent. ## 4. Episode Auto-Sync Maintains continuous synchronized viewing when watching series: diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 6088b5a..8cf7e42 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -141,8 +141,29 @@ relaying the established event names, payloads, and order unchanged. This makes server-first rollout safe: old clients populate recovery state without needing to understand or acknowledge it, and new clients consume it only when the relay advertises the capability. No protocol-version or minimum-version bump is -required. Offline `play`/`pause`/`seek` compaction is planned separately and is -not part of Media State v1. +required. Offline `play`/`pause`/`seek` compaction remains the separate +client-owned layer described below rather than part of the relay capability. + +### Offline media intent + +Offline media intent is client-side queue state, not a relay protocol feature. +An updated extension coalesces contiguous unsent `play`, `pause`, and `seek` +commands for one room. Retained non-media events are ordering barriers. On a +successful rejoin, the extension first reads `room_data` so current Host Control +and Episode Lobby authority can be applied, then replays an authorized intent as +the minimum existing legacy media-event sequence. Actual wire frames, rather +than logical queue entries, consume the paced reconnect budget. + +Pending authorized local intent takes precedence over an older canonical +snapshot because it has not yet been accepted by the relay. Its legacy replay +then updates canonical state like any other accepted control. Without pending +intent, canonical recovery proceeds normally. Intent made stale by a room switch, +role loss, intentional solo mode, or an active Episode Lobby is discarded and +cannot suppress server recovery. + +This requires no event, capability, ACK, protocol-version, or minimum-version +change. Old relays receive ordinary `play`/`pause`/`seek`; old peers see only the +same existing relayed events. ## Ephemeral encrypted chat diff --git a/extension/background.js b/extension/background.js index cacf35a..24742d5 100644 --- a/extension/background.js +++ b/extension/background.js @@ -9,6 +9,19 @@ import { buildChatRelayPayload, encodeSocketEvent } from './chat-wire.js'; import { createChatEchoTracker, createChatSendLimiter, createLatestTaskQueue, normalizeRoomId, shouldShowChatNotification } from './chat-session.js'; import { createChatActivityStore } from './chat-activity.js'; import { canonicalMediaStateFromRoomData, createCanonicalMediaStateTracker } from './canonical-media-state.js'; +import { + drainQueuedBatch, + enqueueQueuedEvent, + isMediaQueueEvent, + isQueuedMediaIntent, + maxQueuedSequence, + mediaIntentNeedsSequenceReservation, + normalizePersistedEventQueue, + queuedMediaIntentCount, + queuedWireCount, + reconcileQueuedRoomIntent, + reserveLatestMediaIntentSequence +} from './offline-media-intent.js'; import { HOST_ACCESS_REQUIRED_STATUS, normalizeTabId, inspectTabHostAccess, isHostAccessError, addTabHostAccessRequest, removeTabHostAccessRequest } from './host-access.js'; import { MEDIA_FRAME_ACCESS_REQUIRED, @@ -234,20 +247,16 @@ let storageInitialized = false; let pendingLogs = []; let pendingHistory = []; let eventQueue = []; +let eventQueueVersion = 0; let flushTimer = null; // paces draining of eventQueue after (re)connect +let flushInProgress = false; let isNamespaceJoined = false; +let awaitingRoomData = false; +let pendingRoomDataRoomId = null; let lastActionState = { action: null, senderId: null, timestamp: 0, acks: [] }; let localSeq = 0; // Monotonically increasing command sequence for this peer const lastSeqBySender = {}; // senderId → last received seq (stale command guard) const canonicalMediaStateTracker = createCanonicalMediaStateTracker(); -const CANONICAL_MEDIA_EVENTS = new Set([ - EVENTS.PLAY, - EVENTS.PAUSE, - EVENTS.SEEK, - EVENTS.FORCE_SYNC_PREPARE, - EVENTS.FORCE_SYNC_EXECUTE -]); -let flushedMediaControlsBeforeRoomData = false; // --- Host Control Mode --- let controlMode = CONTROL_MODES.EVERYONE; // 'everyone' | 'host-only' @@ -296,6 +305,7 @@ function clearChatActivity() { async function clearFailedJoinCredentials() { webJoinCoordinator.invalidate(); connectIntent = false; + clearEventQueue(); clearCanonicalMediaRecovery(); chatSecretGuard = ''; invalidateChatSession(); @@ -442,7 +452,6 @@ function ensureState() { ); if (data.lastActionState) lastActionState = data.lastActionState; - if (data.eventQueue) eventQueue = [...eventQueue, ...data.eventQueue].slice(0, 50); if (data.isForceSyncInitiator !== undefined && isForceSyncInitiator === false) { isForceSyncInitiator = data.isForceSyncInitiator; } @@ -482,10 +491,19 @@ function ensureState() { } } - if (data.localSeq !== undefined && !isNaN(data.localSeq)) localSeq = data.localSeq; + if (Number.isSafeInteger(data.localSeq) && data.localSeq >= 0) localSeq = data.localSeq; + eventQueue = normalizePersistedEventQueue( + [...eventQueue, ...(Array.isArray(data.eventQueue) ? data.eventQueue : [])], + currentRoom?.roomId || null + ); + eventQueueVersion++; + // An MV3 restart can restore queue and sequence writes in either + // order. Never let the sender sequence fall behind persisted work. + localSeq = Math.max(localSeq, maxQueuedSequence(eventQueue)); if (data.lastSeqBySender && typeof data.lastSeqBySender === 'object') Object.assign(lastSeqBySender, data.lastSeqBySender); storageInitialized = true; + chrome.storage.session.set({ eventQueue, localSeq }).catch(() => {}); // Process any early logs/history that weren't captured in the spread if (pendingLogs.length > 0) { @@ -728,6 +746,8 @@ function forceDisconnect() { currentServerUrl = null; isConnecting = false; isNamespaceJoined = false; + awaitingRoomData = false; + pendingRoomDataRoomId = null; invalidateChatSession(); isForceSyncInitiator = false; expectedAcksCount = 0; @@ -735,7 +755,9 @@ function forceDisconnect() { lastContentHeartbeatAt = null; forceSyncAcks.clear(); if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; } + flushInProgress = false; eventQueue = []; + eventQueueVersion++; chrome.storage.session.set({ isForceSyncInitiator: false, forceSyncAcks: [], @@ -1082,6 +1104,7 @@ async function connect() { try { if (!peerId) peerId = await getPeerId(); settings = await getSettings(); + pendingRoomDataRoomId = settings.roomId || currentRoom?.roomId || null; } catch (e) { throw new Error(`[Storage Error] ${e.message}`); } @@ -1180,6 +1203,8 @@ async function connect() { addLog('Joined Namespace /', 'success'); const settings = await getSettings(); if (settings.roomId) { + awaitingRoomData = true; + pendingRoomDataRoomId = settings.roomId; const sharedTitles = getSharedTitleFields(settings); emit(EVENTS.JOIN_ROOM, { roomId: settings.roomId, @@ -1190,14 +1215,11 @@ async function connect() { clientCapabilities: CLIENT_CAPABILITIES, protocolVersion: PROTOCOL_VERSION }); + } else { + awaitingRoomData = false; + pendingRoomDataRoomId = null; + flushEventQueue(); } - // Preserve the existing immediate queue-drain behavior. Remember - // whether local media intent was flushed before ROOM_DATA so an - // older snapshot cannot snap the player backward on reconnect. - flushedMediaControlsBeforeRoomData = eventQueue.some(item => - item && CANONICAL_MEDIA_EVENTS.has(item.event) - ); - flushEventQueue(); } else if (msg.startsWith('42')) { try { const payload = JSON.parse(msg.substring(2)); @@ -1215,6 +1237,8 @@ async function connect() { socket.onclose = () => { isConnecting = false; isNamespaceJoined = false; + awaitingRoomData = false; + pendingRoomDataRoomId = null; invalidateChatSession(); stopPing(); if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; } @@ -1490,7 +1514,10 @@ function scheduleReconnect() { // Slow reconnect logic is now handled in the keepAlive alarm function emit(event, data) { - if (socket && socket.readyState === WebSocket.OPEN && isNamespaceJoined) { + const mustWaitForRoomData = awaitingRoomData + && event !== EVENTS.JOIN_ROOM + && event !== EVENTS.GET_ROOMS; + if (socket && socket.readyState === WebSocket.OPEN && isNamespaceJoined && !mustWaitForRoomData) { try { const msg = encodeSocketEvent(event, data, chatSecretGuard); socket.send(msg); @@ -1522,41 +1549,120 @@ function emitLive(event, data) { } function queueEvent(event, data) { - eventQueue.push({ event, data }); - if (eventQueue.length > 50) { - eventQueue.shift(); + const queueRoomId = currentRoom?.roomId || pendingRoomDataRoomId; + const queued = enqueueQueuedEvent(eventQueue, event, data, { roomId: queueRoomId }); + eventQueue = queued.queue; + eventQueueVersion++; + + if (isMediaQueueEvent(event)) { + const latest = eventQueue.at(-1); + if (mediaIntentNeedsSequenceReservation(latest)) { + const nextSequence = Math.max(localSeq, maxQueuedSequence(eventQueue)) + 1; + const reserved = reserveLatestMediaIntentSequence(eventQueue, queueRoomId, nextSequence); + eventQueue = reserved.queue; + if (reserved.reserved) { + localSeq = nextSequence; + chrome.storage.session.set({ localSeq }).catch(() => {}); + } + } + const sourceCount = isQueuedMediaIntent(eventQueue.at(-1)) + ? eventQueue.at(-1).intent.sourceEventCount + : 0; + if (sourceCount === 1) { + addLog('Offline media intent queued; replay deferred until ROOM_DATA', 'info'); + } else if ([10, 50, 100, 500, 1000].includes(sourceCount)) { + addLog(`Collapsed ${sourceCount} queued media commands into one intent`, 'info'); + } + } + if (queued.dropped > 0) { addLog('Event queue cap reached, dropping oldest event', 'warn'); } - chrome.storage.session.set({ eventQueue }); + chrome.storage.session.set({ eventQueue }).catch(() => {}); +} + +function clearEventQueue() { + if (flushTimer) { + clearTimeout(flushTimer); + flushTimer = null; + } + eventQueue = []; + eventQueueVersion++; + chrome.storage.session.set({ eventQueue: [] }).catch(() => {}); +} + +function applyQueuedRoomPolicy(roomId, policy, reason) { + const result = reconcileQueuedRoomIntent(eventQueue, { roomId, ...policy }); + eventQueue = result.queue; + eventQueueVersion++; + const { discarded } = result; + if (discarded > 0) { + chrome.storage.session.set({ eventQueue }).catch(() => {}); + addLog(`${reason}: discarded ${discarded} queued room action${discarded === 1 ? '' : 's'}`, 'warn'); + } + return result; } /** - * Drain the offline event queue in paced batches. A reconnect after a long - * outage can leave up to 50 queued events; dumping them in one tick would - * exceed the server's per-socket event budget and get us disconnected right - * after rejoining. We send FLUSH_BATCH_SIZE events, then wait - * FLUSH_BATCH_INTERVAL_MS before the next batch. Remaining events drain across - * subsequent batches; if the connection drops mid-drain, the rest stay queued. + * Drain logical queue entries only after ROOM_DATA confirms the current room, + * role and lobby. Pacing counts materialized wire frames, not logical entries. + * A two-frame media intent is kept whole at a batch boundary and retained in + * full if either send fails; replaying its first frame again is idempotent and + * preserves the final desired state on the next reconnect. */ -function flushEventQueue() { - if (flushTimer) return; // a drain is already in progress - const drainBatch = () => { - flushTimer = null; - if (!socket || socket.readyState !== WebSocket.OPEN || !isNamespaceJoined) { - return; // lost the connection — leave the rest queued for next connect - } - let sent = 0; - while (eventQueue.length > 0 && sent < FLUSH_BATCH_SIZE) { - const queuedMsg = eventQueue.shift(); - emit(queuedMsg.event, queuedMsg.data); - sent++; +async function flushEventQueue() { + if (flushTimer || flushInProgress || awaitingRoomData) return; + if (!socket || socket.readyState !== WebSocket.OPEN || !isNamespaceJoined) return; + flushInProgress = true; + try { + // Resolve privacy settings before taking queue ownership. If new work + // arrives during the drain, keep the original logical entries too; a + // later idempotent replay is safer than losing the concurrent intent. + const replaySettings = eventQueue.some(isQueuedMediaIntent) ? await getSettings() : null; + const drainSource = eventQueue; + const drainVersion = eventQueueVersion; + const result = await drainQueuedBatch(drainSource, { + roomId: currentRoom?.roomId || null, + maxWireEvents: FLUSH_BATCH_SIZE, + sendFrame: async (frame, entry) => { + let payload = frame.data && typeof frame.data === 'object' ? { ...frame.data } : {}; + if (isQueuedMediaIntent(entry)) { + payload = withTitlePrivacy(payload, replaySettings, ['mediaTitle']); + payload.peerId = peerId; + } + return emitLive(frame.event, payload); + } + }); + if (eventQueueVersion === drainVersion) { + eventQueue = result.queue; + eventQueueVersion++; + } else { + // enqueueQueuedEvent preserves object identity for untouched queue + // entries. Remove only entries the drain completed, leaving any + // concurrently appended or merged work in place. A partial intent + // failure returns the complete entry in result.queue, so it is not + // included in consumedEntries and remains retryable as a unit. + const consumedCount = drainSource.length - result.queue.length; + const consumedEntries = new Set(drainSource.slice(0, consumedCount)); + eventQueue = eventQueue.filter(entry => !consumedEntries.has(entry)); + eventQueueVersion++; + addLog('Queue changed during replay; reconciled completed and concurrent work', 'info'); } chrome.storage.session.set({ eventQueue }).catch(() => {}); - if (eventQueue.length > 0) { - flushTimer = setTimeout(drainBatch, FLUSH_BATCH_INTERVAL_MS); + if (result.droppedStaleIntents > 0) { + addLog(`Dropped ${result.droppedStaleIntents} stale media intent${result.droppedStaleIntents === 1 ? '' : 's'} for a previous room`, 'warn'); } - }; - drainBatch(); + if (result.sentWireEvents > 0) { + addLog(`Replayed ${result.sentWireEvents} queued wire event${result.sentWireEvents === 1 ? '' : 's'}`, 'info'); + } + if (eventQueue.length > 0 && socket?.readyState === WebSocket.OPEN && isNamespaceJoined) { + flushTimer = setTimeout(() => { + flushTimer = null; + flushEventQueue().catch(error => addLog(`Queue replay failed: ${error.message}`, 'warn')); + }, FLUSH_BATCH_INTERVAL_MS); + } + } finally { + flushInProgress = false; + } } function addToHistory(action, senderId) { @@ -1678,7 +1784,7 @@ async function tryApplyPendingCanonicalMediaState() { } } -async function handleCanonicalRoomData(data, hadQueuedMediaControls) { +async function handleCanonicalRoomData(data, hasPendingLocalIntent) { canonicalMediaStateTracker.adoptRoom(data?.roomId || null); const canonicalSnapshot = canonicalMediaStateFromRoomData(data); if (canonicalSnapshot.status === 'unsupported' || canonicalSnapshot.status === 'empty') { @@ -1703,9 +1809,9 @@ async function handleCanonicalRoomData(data, hadQueuedMediaControls) { persistCanonicalMediaRecovery(); addLog(`Canonical media state received: r${mediaState.revision} ${mediaState.playbackState} @ ${mediaState.currentTime.toFixed(2)}s`, 'info'); - if (hadQueuedMediaControls) { + if (hasPendingLocalIntent) { markCanonicalMediaStateHandled(data.roomId, mediaState.revision); - addLog(`Canonical media state r${mediaState.revision} skipped: queued local media controls are authoritative for this reconnect`, 'info'); + addLog(`Canonical media state r${mediaState.revision} skipped: authorized queued local intent takes precedence`, 'info'); return; } const result = await tryApplyPendingCanonicalMediaState(); @@ -1735,8 +1841,6 @@ async function handleServerEvent(event, data) { } switch (event) { case EVENTS.ROOM_DATA: { - const hadQueuedMediaControls = flushedMediaControlsBeforeRoomData; - flushedMediaControlsBeforeRoomData = false; if (currentRoom?.roomId !== data.roomId) { invalidateChatSession(); clearChatActivity(); @@ -1767,7 +1871,10 @@ async function handleServerEvent(event, data) { } // Recover server-tracked active Episode Lobby if present - if (data && data.activeLobby && !episodeLobby) { + if (!data?.activeLobby && episodeLobby) { + clearEpisodeLobbyState(); + addLog('Discarded stale local Episode Lobby after ROOM_DATA confirmed it ended', 'info'); + } else if (data && data.activeLobby && !episodeLobby) { episodeLobby = { expectedTitle: data.activeLobby.expectedTitle, initiatorPeerId: data.activeLobby.initiatorPeerId, @@ -1801,7 +1908,22 @@ async function handleServerEvent(event, data) { // Inform Website Bridge & Popup const joinStatusMsg = { type: 'JOIN_STATUS', success: true, message: 'Joined' }; await broadcastJoinStatus(joinStatusMsg); - await handleCanonicalRoomData(data, hadQueuedMediaControls); + awaitingRoomData = false; + pendingRoomDataRoomId = null; + + const lostRoomAuthority = controlMode === CONTROL_MODES.HOST_ONLY + && hostPeerId + && !amController(); + const queuePolicy = applyQueuedRoomPolicy(data.roomId, { + canControl: !lostRoomAuthority, + activeLobby: !!episodeLobby, + desynced: hcmDesynced + }, lostRoomAuthority + ? 'Host Control role changed while offline' + : (episodeLobby ? 'Active Episode Lobby takes precedence' : 'Reconnect queue policy')); + + await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent); + await flushEventQueue(); break; } case EVENTS.CONTROL_MODE: @@ -4013,6 +4135,9 @@ async function handleAsyncMessage(message, sender, sendResponse) { episodeLobby: episodeLobby, reconnectAttempts, reconnectSlowMode: reconnectFailed, + queuedLogicalEvents: eventQueue.length, + queuedMediaIntents: queuedMediaIntentCount(eventQueue, currentRoom?.roomId || pendingRoomDataRoomId), + queuedWireEvents: queuedWireCount(eventQueue), roomId: currentRoom ? currentRoom.roomId : null, serverUrl: currentServerUrl, version: chrome.runtime.getManifest().version, @@ -4216,6 +4341,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { hcmDesynced = !!message.desynced; if (storageInitialized) chrome.storage.session.set({ hcmDesynced }); if (hcmDesynced) { + applyQueuedRoomPolicy(currentRoom?.roomId, { desynced: true }, 'Intentional solo mode'); const pending = canonicalMediaStateTracker.getPending(currentRoom?.roomId); if (pending) { markCanonicalMediaStateHandled(pending.roomId, pending.mediaState.revision); diff --git a/extension/canonical-media-state-background.test.mjs b/extension/canonical-media-state-background.test.mjs index 97083a4..67024ac 100644 --- a/extension/canonical-media-state-background.test.mjs +++ b/extension/canonical-media-state-background.test.mjs @@ -64,8 +64,10 @@ describe('canonical ROOM_DATA recovery contract', () => { const roomData = functionBody(backgroundSource, 'handleCanonicalRoomData', 'handleServerEvent'); expect(apply).toContain('if (hcmDesynced)'); expect(apply).toContain('if (episodeLobby)'); - expect(roomData).toContain('if (hadQueuedMediaControls)'); - expect(backgroundSource).toContain('flushedMediaControlsBeforeRoomData = eventQueue.some'); + expect(roomData).toContain('if (hasPendingLocalIntent)'); + expect(backgroundSource).toContain('awaitingRoomData = true'); + expect(backgroundSource).toContain('await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)'); + expect(backgroundSource).toContain('await flushEventQueue()'); }); it('reuses existing seek abstractions, suppression and drift tolerance', () => { diff --git a/extension/offline-media-intent-background.test.mjs b/extension/offline-media-intent-background.test.mjs new file mode 100644 index 0000000..882e848 --- /dev/null +++ b/extension/offline-media-intent-background.test.mjs @@ -0,0 +1,73 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; + +const extensionDir = path.dirname(fileURLToPath(import.meta.url)); +const backgroundSource = fs.readFileSync(path.join(extensionDir, 'background.js'), 'utf8'); + +function functionBody(name, nextName) { + const start = backgroundSource.indexOf(`function ${name}(`); + const end = backgroundSource.indexOf(`function ${nextName}(`, start + 1); + return backgroundSource.slice(start, end === -1 ? backgroundSource.length : end); +} + +describe('offline media intent background integration', () => { + it('keeps online sends immediate and defers reconnect work only while ROOM_DATA is pending', () => { + const emit = functionBody('emit', 'emitLive'); + expect(emit).toContain('mustWaitForRoomData'); + expect(emit).toContain('socket.send(msg)'); + expect(emit).toContain('queueEvent(event, data)'); + expect(emit).not.toContain('setTimeout'); + expect(backgroundSource).toMatch(/awaitingRoomData = true;[\s\S]*emit\(EVENTS\.JOIN_ROOM/); + }); + + it('restores and migrates the persisted MV3 queue with a monotonic local sequence', () => { + expect(backgroundSource).toContain('normalizePersistedEventQueue('); + expect(backgroundSource).toContain('localSeq = Math.max(localSeq, maxQueuedSequence(eventQueue))'); + expect(backgroundSource).toContain('chrome.storage.session.set({ eventQueue, localSeq })'); + expect(backgroundSource).not.toContain('storage.sync.set({ eventQueue'); + }); + + it('reconciles Host Control, Episode Lobby and solo mode before canonical recovery and replay', () => { + const roomDataStart = backgroundSource.indexOf('case EVENTS.ROOM_DATA:'); + const roomDataEnd = backgroundSource.indexOf('case EVENTS.CONTROL_MODE:', roomDataStart); + const roomData = backgroundSource.slice(roomDataStart, roomDataEnd); + expect(roomData.indexOf('applyQueuedRoomPolicy(data.roomId')) + .toBeLessThan(roomData.indexOf('handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)')); + expect(roomData.indexOf('handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)')) + .toBeLessThan(roomData.indexOf('flushEventQueue()')); + expect(roomData).toContain('activeLobby: !!episodeLobby'); + expect(roomData).toContain('desynced: hcmDesynced'); + expect(roomData).toContain('if (!data?.activeLobby && episodeLobby)'); + }); + + it('clears queued room intent on failed join, leave and room switch paths', () => { + expect(functionBody('clearFailedJoinCredentials', 'invalidateChatSession')).toContain('clearEventQueue()'); + expect(functionBody('forceDisconnect', 'persistRoomIdleState')).toContain('eventQueue = []'); + expect(functionBody('leaveOldRoomIfSwitching', 'resetAudioProcessingInTab')).toContain('forceDisconnect()'); + const leaveHandler = backgroundSource.slice( + backgroundSource.indexOf("message.type === 'LEAVE_ROOM'"), + backgroundSource.indexOf("message.type === 'CLEAR_LOGS'") + ); + expect(leaveHandler).toContain('forceDisconnect()'); + }); + + it('paces actual frames through a failure-retaining logical drain', () => { + const flush = functionBody('flushEventQueue', 'addToHistory'); + expect(flush).toContain('drainQueuedBatch(drainSource'); + expect(flush).toContain('maxWireEvents: FLUSH_BATCH_SIZE'); + expect(flush).toContain('return emitLive(frame.event, payload)'); + expect(flush).toContain('if (eventQueueVersion === drainVersion)'); + expect(flush).toContain('const consumedEntries = new Set(drainSource.slice(0, consumedCount))'); + expect(flush).toContain('eventQueue = eventQueue.filter(entry => !consumedEntries.has(entry))'); + expect(flush).not.toMatch(/eventQueue\.shift\(\)[\s\S]*emit\(/); + }); + + it('exposes bounded queue diagnostics without media-title content', () => { + expect(backgroundSource).toContain('queuedLogicalEvents: eventQueue.length'); + expect(backgroundSource).toContain('queuedMediaIntents: queuedMediaIntentCount'); + expect(backgroundSource).toContain('queuedWireEvents: queuedWireCount(eventQueue)'); + expect(backgroundSource).not.toMatch(/Offline media intent[^`'\n]*mediaTitle/); + }); +}); diff --git a/extension/offline-media-intent.js b/extension/offline-media-intent.js new file mode 100644 index 0000000..40ed8be --- /dev/null +++ b/extension/offline-media-intent.js @@ -0,0 +1,407 @@ +import { EVENTS, MAX_MEDIA_TIME } from './shared/constants.js'; + +export const MEDIA_INTENT_KIND = 'media-intent'; +export const MAX_LOGICAL_QUEUE_SIZE = 50; + +const MEDIA_EVENTS = new Set([EVENTS.PLAY, EVENTS.PAUSE, EVENTS.SEEK]); +const STALE_OFFLINE_EVENTS = new Set([EVENTS.PING, EVENTS.PONG, EVENTS.PEER_STATUS, EVENTS.EVENT_ACK]); +const FORCE_SYNC_EVENTS = new Set([EVENTS.FORCE_SYNC_PREPARE, EVENTS.FORCE_SYNC_EXECUTE]); +const HOST_GATED_EVENTS = new Set([ + ...MEDIA_EVENTS, + ...FORCE_SYNC_EVENTS, + EVENTS.EPISODE_LOBBY, + EVENTS.EPISODE_LOBBY_CANCEL +]); + +function validSequence(value) { + return Number.isSafeInteger(value) && value >= 0 ? value : null; +} + +function validTimestamp(value) { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 ? value : null; +} + +function mediaTime(value) { + if (typeof value !== 'number' || !Number.isFinite(value)) return null; + return Math.max(0, Math.min(MAX_MEDIA_TIME, value)); +} + +function playbackStateFor(event, data) { + if (event === EVENTS.PLAY) return 'playing'; + if (event === EVENTS.PAUSE) return 'paused'; + return data?.playbackState === 'playing' || data?.playbackState === 'paused' + ? data.playbackState + : null; +} + +function mediaPositionFor(event, data) { + if (event === EVENTS.SEEK) { + return mediaTime(data?.targetTime) ?? mediaTime(data?.currentTime); + } + return mediaTime(data?.currentTime); +} + +function sanitizedMediaTitle(value) { + return typeof value === 'string' && value ? value.substring(0, 100) : null; +} + +function trimQueue(queue, maxEntries) { + const trimmed = queue.slice(); + let dropped = 0; + while (trimmed.length > maxEntries) { + trimmed.shift(); + dropped++; + } + return { queue: trimmed, dropped }; +} + +function createIntentEntry(event, data, roomId) { + const currentTime = mediaPositionFor(event, data); + const playbackState = playbackStateFor(event, data); + if (event === EVENTS.SEEK && currentTime === null) return null; + if (playbackState === null && currentTime === null) return null; + return { + kind: MEDIA_INTENT_KIND, + roomId, + intent: { + playbackState, + currentTime, + latestEvent: event, + previousSeq: null, + latestSeq: validSequence(data?.seq), + actionTimestamp: validTimestamp(data?.actionTimestamp), + mediaTitle: sanitizedMediaTitle(data?.mediaTitle), + sourceEventCount: 1 + } + }; +} + +function mergeIntentEntry(entry, event, data) { + const incomingSeq = validSequence(data?.seq); + const previousLatestSeq = validSequence(entry.intent.latestSeq); + if (incomingSeq !== null && previousLatestSeq !== null && incomingSeq <= previousLatestSeq) { + return null; + } + + const incomingPosition = mediaPositionFor(event, data); + if (event === EVENTS.SEEK && incomingPosition === null) return entry; + const incomingState = playbackStateFor(event, data); + const merged = { + ...entry, + intent: { + ...entry.intent, + playbackState: incomingState ?? entry.intent.playbackState, + currentTime: incomingPosition ?? entry.intent.currentTime, + latestEvent: event, + actionTimestamp: validTimestamp(data?.actionTimestamp) ?? entry.intent.actionTimestamp, + mediaTitle: sanitizedMediaTitle(data?.mediaTitle) ?? entry.intent.mediaTitle, + sourceEventCount: entry.intent.sourceEventCount + 1 + } + }; + if (incomingSeq !== null) { + merged.intent.previousSeq = previousLatestSeq ?? validSequence(entry.intent.previousSeq); + merged.intent.latestSeq = incomingSeq; + } + return merged; +} + +export function isMediaQueueEvent(event) { + return MEDIA_EVENTS.has(event); +} + +export function isQueuedMediaIntent(entry) { + return entry?.kind === MEDIA_INTENT_KIND + && typeof entry.roomId === 'string' + && entry.roomId + && entry.intent + && typeof entry.intent === 'object'; +} + +export function enqueueQueuedEvent(queue, event, data, { + roomId, + maxEntries = MAX_LOGICAL_QUEUE_SIZE +} = {}) { + const next = Array.isArray(queue) ? queue.slice() : []; + let collapsed = 0; + + if (STALE_OFFLINE_EVENTS.has(event)) { + return { queue: next, collapsed: 0, dropped: 0, droppedStale: 1 }; + } + if (!isMediaQueueEvent(event)) { + next.push({ event, data }); + } else if (typeof roomId === 'string' && roomId) { + const last = next.at(-1); + const hasMergeTarget = isQueuedMediaIntent(last) && last.roomId === roomId; + const merged = hasMergeTarget + ? mergeIntentEntry(last, event, data) + : null; + if (merged) { + next[next.length - 1] = merged; + collapsed = 1; + } else if (!hasMergeTarget) { + const entry = createIntentEntry(event, data, roomId); + if (entry) next.push(entry); + } else { + return { queue: next, collapsed: 0, dropped: 0, droppedStale: 1 }; + } + } + + const trimmed = trimQueue(next, maxEntries); + return { queue: trimmed.queue, collapsed, dropped: trimmed.dropped, droppedStale: 0 }; +} + +function normalizeIntentEntry(entry, roomId) { + if (!isQueuedMediaIntent(entry) || entry.roomId !== roomId) return null; + const intent = entry.intent; + const playbackState = intent.playbackState === 'playing' || intent.playbackState === 'paused' + ? intent.playbackState + : null; + const currentTime = mediaTime(intent.currentTime); + const latestEvent = isMediaQueueEvent(intent.latestEvent) ? intent.latestEvent : null; + if (!latestEvent || (playbackState === null && currentTime === null)) return null; + return { + kind: MEDIA_INTENT_KIND, + roomId, + intent: { + playbackState, + currentTime, + latestEvent, + previousSeq: validSequence(intent.previousSeq), + latestSeq: validSequence(intent.latestSeq), + actionTimestamp: validTimestamp(intent.actionTimestamp), + mediaTitle: sanitizedMediaTitle(intent.mediaTitle), + sourceEventCount: Number.isSafeInteger(intent.sourceEventCount) && intent.sourceEventCount > 0 + ? intent.sourceEventCount + : 1 + } + }; +} + +export function normalizePersistedEventQueue(value, roomId, maxEntries = MAX_LOGICAL_QUEUE_SIZE) { + if (!Array.isArray(value) || typeof roomId !== 'string' || !roomId) return []; + let normalized = []; + for (const entry of value) { + if (isQueuedMediaIntent(entry)) { + const intentEntry = normalizeIntentEntry(entry, roomId); + if (intentEntry) normalized.push(intentEntry); + continue; + } + if (!entry || typeof entry !== 'object' || typeof entry.event !== 'string') continue; + if (STALE_OFFLINE_EVENTS.has(entry.event)) continue; + if (isMediaQueueEvent(entry.event)) { + normalized = enqueueQueuedEvent(normalized, entry.event, entry.data, { roomId, maxEntries }).queue; + } else { + normalized.push({ event: entry.event, data: entry.data }); + } + normalized = trimQueue(normalized, maxEntries).queue; + } + return normalized; +} + +export function mediaIntentNeedsSequenceReservation(entry) { + if (!isQueuedMediaIntent(entry)) return false; + const { playbackState, currentTime, previousSeq, latestSeq } = entry.intent; + return playbackState !== null + && currentTime !== null + && validSequence(latestSeq) !== null + && validSequence(previousSeq) === null; +} + +export function reserveLatestMediaIntentSequence(queue, roomId, nextSequence) { + const next = Array.isArray(queue) ? queue.slice() : []; + const index = next.length - 1; + const entry = next[index]; + if (!isQueuedMediaIntent(entry) + || entry.roomId !== roomId + || !mediaIntentNeedsSequenceReservation(entry) + || validSequence(nextSequence) === null + || nextSequence <= entry.intent.latestSeq) { + return { queue: next, reserved: false }; + } + next[index] = { + ...entry, + intent: { + ...entry.intent, + previousSeq: entry.intent.latestSeq, + latestSeq: nextSequence + } + }; + return { queue: next, reserved: true }; +} + +function frameData(intent, seq) { + const data = {}; + if (seq !== null) data.seq = seq; + if (intent.actionTimestamp !== null) data.actionTimestamp = intent.actionTimestamp; + if (intent.mediaTitle !== null) data.mediaTitle = intent.mediaTitle; + return data; +} + +export function materializeMediaIntent(entry) { + if (!isQueuedMediaIntent(entry)) return []; + const intent = entry.intent; + const currentTime = mediaTime(intent.currentTime); + const playbackState = intent.playbackState === 'playing' || intent.playbackState === 'paused' + ? intent.playbackState + : null; + const latestSeq = validSequence(intent.latestSeq); + const previousSeq = validSequence(intent.previousSeq); + const stateEvent = playbackState === 'playing' ? EVENTS.PLAY : EVENTS.PAUSE; + + if (playbackState === null && currentTime !== null) { + return [{ + event: EVENTS.SEEK, + data: { ...frameData(intent, latestSeq), currentTime, targetTime: currentTime } + }]; + } + if (playbackState !== null && currentTime === null) { + return [{ event: stateEvent, data: frameData(intent, latestSeq) }]; + } + if (playbackState === null || currentTime === null) return []; + + // Both materialized frames carry the latest genuine action timestamp: it is + // correlation metadata for existing ACK/activity paths, not scheduled wall + // time. A previous-format single PLAY/PAUSE has only one reserved sequence. Keep + // its original one-frame behavior during migration rather than inventing a + // sequence that could overtake a later transactional barrier. + if (previousSeq === null || latestSeq === null || previousSeq >= latestSeq) { + if (intent.latestEvent === EVENTS.SEEK) { + return [{ + event: EVENTS.SEEK, + data: { ...frameData(intent, latestSeq), currentTime, targetTime: currentTime } + }]; + } + return [{ + event: stateEvent, + data: { ...frameData(intent, latestSeq), currentTime } + }]; + } + + const seekFrame = { + event: EVENTS.SEEK, + data: { ...frameData(intent, intent.latestEvent === EVENTS.SEEK ? latestSeq : previousSeq), currentTime, targetTime: currentTime } + }; + const stateFrame = { + event: stateEvent, + data: { ...frameData(intent, intent.latestEvent === EVENTS.SEEK ? previousSeq : latestSeq), currentTime } + }; + return intent.latestEvent === EVENTS.SEEK + ? [stateFrame, seekFrame] + : [seekFrame, stateFrame]; +} + +export function queuedEntryWireCount(entry) { + return isQueuedMediaIntent(entry) ? materializeMediaIntent(entry).length : 1; +} + +export function queuedWireCount(queue) { + return Array.isArray(queue) + ? queue.reduce((total, entry) => total + queuedEntryWireCount(entry), 0) + : 0; +} + +export function queuedMediaIntentCount(queue, roomId = null) { + return Array.isArray(queue) + ? queue.filter(entry => isQueuedMediaIntent(entry) && (!roomId || entry.roomId === roomId)).length + : 0; +} + +export function hasQueuedMediaIntent(queue, roomId) { + return queuedMediaIntentCount(queue, roomId) > 0; +} + +export function discardQueuedMediaIntents(queue, roomId) { + return Array.isArray(queue) + ? queue.filter(entry => !isQueuedMediaIntent(entry) || entry.roomId !== roomId) + : []; +} + +export function reconcileQueuedRoomIntent(queue, { + roomId, + canControl = true, + activeLobby = false, + desynced = false +} = {}) { + const source = Array.isArray(queue) ? queue : []; + const blockedEvents = !canControl + ? HOST_GATED_EVENTS + : (activeLobby || desynced ? FORCE_SYNC_EVENTS : null); + const reconciled = source.filter(entry => { + if (isQueuedMediaIntent(entry) && entry.roomId === roomId) { + return canControl && !activeLobby && !desynced; + } + return !blockedEvents?.has(entry?.event); + }); + const hasPendingLocalIntent = reconciled.some(entry => + (isQueuedMediaIntent(entry) && entry.roomId === roomId) + || (!isQueuedMediaIntent(entry) && FORCE_SYNC_EVENTS.has(entry?.event)) + ); + return { + queue: reconciled, + discarded: source.length - reconciled.length, + hasPendingLocalIntent + }; +} + +export function maxQueuedSequence(queue) { + let max = 0; + for (const entry of Array.isArray(queue) ? queue : []) { + if (isQueuedMediaIntent(entry)) { + max = Math.max(max, validSequence(entry.intent.previousSeq) ?? 0, validSequence(entry.intent.latestSeq) ?? 0); + } else { + max = Math.max(max, validSequence(entry?.data?.seq) ?? 0); + } + } + return max; +} + +export async function drainQueuedBatch(queue, { + roomId, + maxWireEvents, + sendFrame +}) { + const remaining = Array.isArray(queue) ? queue.slice() : []; + let sentWireEvents = 0; + let droppedStaleIntents = 0; + + while (remaining.length > 0) { + const entry = remaining[0]; + if (isQueuedMediaIntent(entry) && entry.roomId !== roomId) { + remaining.shift(); + droppedStaleIntents++; + continue; + } + const frames = isQueuedMediaIntent(entry) + ? materializeMediaIntent(entry) + : [{ event: entry.event, data: entry.data }]; + if (frames.length === 0) { + remaining.shift(); + continue; + } + if (sentWireEvents > 0 && sentWireEvents + frames.length > maxWireEvents) { + return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'batch_full' }; + } + if (frames.length > maxWireEvents) { + return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'entry_exceeds_batch' }; + } + + let sentEntryFrames = 0; + for (const frame of frames) { + if (!await sendFrame(frame, entry)) { + return { + queue: remaining, + sentWireEvents: sentWireEvents + sentEntryFrames, + droppedStaleIntents, + status: 'send_failed' + }; + } + sentEntryFrames++; + } + sentWireEvents += sentEntryFrames; + remaining.shift(); + if (sentWireEvents >= maxWireEvents) { + return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'batch_full' }; + } + } + return { queue: remaining, sentWireEvents, droppedStaleIntents, status: 'drained' }; +} diff --git a/extension/offline-media-intent.test.mjs b/extension/offline-media-intent.test.mjs new file mode 100644 index 0000000..d7f3ead --- /dev/null +++ b/extension/offline-media-intent.test.mjs @@ -0,0 +1,265 @@ +import { describe, expect, it } from 'vitest'; +import { EVENTS, MAX_MEDIA_TIME } from './shared/constants.js'; +import { canonicalMediaStateFromRoomData } from './canonical-media-state.js'; +import { + discardQueuedMediaIntents, + drainQueuedBatch, + enqueueQueuedEvent, + hasQueuedMediaIntent, + materializeMediaIntent, + maxQueuedSequence, + normalizePersistedEventQueue, + queuedEntryWireCount, + queuedMediaIntentCount, + queuedWireCount, + reconcileQueuedRoomIntent, + reserveLatestMediaIntentSequence +} from './offline-media-intent.js'; + +const roomId = 'room-a'; +const media = (event, data, queue = []) => enqueueQueuedEvent(queue, event, data, { roomId }).queue; + +function reserve(queue, sequence) { + return reserveLatestMediaIntentSequence(queue, roomId, sequence).queue; +} + +describe('offline media intent coalescing', () => { + it('normalizes an empty or roomless persisted queue to empty', () => { + expect(normalizePersistedEventQueue([], roomId)).toEqual([]); + expect(normalizePersistedEventQueue([{ event: EVENTS.PLAY, data: { seq: 1 } }], null)).toEqual([]); + }); + + it('creates a PLAY intent and reserves ordered legacy SEEK + PLAY frames', () => { + let queue = media(EVENTS.PLAY, { currentTime: 10, seq: 5, actionTimestamp: 100 }); + queue = reserve(queue, 6); + expect(queue).toHaveLength(1); + expect(materializeMediaIntent(queue[0])).toEqual([ + { event: EVENTS.SEEK, data: { seq: 5, actionTimestamp: 100, currentTime: 10, targetTime: 10 } }, + { event: EVENTS.PLAY, data: { seq: 6, actionTimestamp: 100, currentTime: 10 } } + ]); + }); + + it('collapses repeated PLAY and SEEK while preserving final state and monotonic sequences', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2); + queue = media(EVENTS.PLAY, { currentTime: 20, seq: 3 }, queue); + queue = media(EVENTS.SEEK, { targetTime: 30, seq: 4 }, queue); + expect(queue).toHaveLength(1); + expect(queue[0].intent).toMatchObject({ + playbackState: 'playing', + currentTime: 30, + latestEvent: EVENTS.SEEK, + previousSeq: 3, + latestSeq: 4, + sourceEventCount: 3 + }); + expect(materializeMediaIntent(queue[0]).map(frame => frame.data.seq)).toEqual([3, 4]); + }); + + it('drops a regressing queued sequence instead of emitting stale ordering', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 10 }), 11); + const result = enqueueQueuedEvent(queue, EVENTS.PAUSE, { currentTime: 20, seq: 9 }, { roomId }); + expect(result.droppedStale).toBe(1); + expect(result.queue).toEqual(queue); + expect(maxQueuedSequence(result.queue)).toBe(11); + }); + + it('merges PLAY -> SEEK -> PAUSE into paused at the final position', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 500, seq: 1 }), 2); + queue = media(EVENTS.SEEK, { targetTime: 600, seq: 3 }, queue); + queue = media(EVENTS.PAUSE, { currentTime: 605, seq: 4 }, queue); + expect(materializeMediaIntent(queue[0])).toEqual([ + { event: EVENTS.SEEK, data: { seq: 3, currentTime: 605, targetTime: 605 } }, + { event: EVENTS.PAUSE, data: { seq: 4, currentTime: 605 } } + ]); + }); + + it('merges PAUSE -> SEEK -> PLAY into playing at the final position', () => { + let queue = reserve(media(EVENTS.PAUSE, { currentTime: 100, seq: 10 }), 11); + queue = media(EVENTS.SEEK, { targetTime: 200, seq: 12 }, queue); + queue = media(EVENTS.PLAY, { currentTime: 205, seq: 13 }, queue); + expect(materializeMediaIntent(queue[0]).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PLAY]); + expect(queue[0].intent).toMatchObject({ playbackState: 'playing', currentTime: 205 }); + }); + + it('clamps finite positions and rejects a SEEK with no trustworthy position', () => { + expect(media(EVENTS.SEEK, { targetTime: NaN, seq: 1 })).toEqual([]); + expect(media(EVENTS.SEEK, { targetTime: -50, seq: 1 })[0].intent.currentTime).toBe(0); + expect(media(EVENTS.SEEK, { targetTime: MAX_MEDIA_TIME + 50, seq: 1 })[0].intent.currentTime).toBe(MAX_MEDIA_TIME); + }); + + it('keeps title metadata bounded and never retains arbitrary payload fields', () => { + const queue = media(EVENTS.PAUSE, { + currentTime: 10, + seq: 1, + mediaTitle: 'x'.repeat(200), + password: 'secret', + chatKey: 'secret' + }); + expect(queue[0].intent.mediaTitle).toHaveLength(100); + expect(JSON.stringify(queue)).not.toContain('password'); + expect(JSON.stringify(queue)).not.toContain('chatKey'); + }); + + it('keeps a thousand-event media burst at one logical entry', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 0, seq: 1 }), 2); + for (let index = 1; index <= 1000; index++) { + queue = media(EVENTS.SEEK, { targetTime: index, seq: index + 2 }, queue); + } + queue = media(EVENTS.PAUSE, { currentTime: 1000, seq: 1003 }, queue); + expect(queue).toHaveLength(1); + expect(queue[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 1000, sourceEventCount: 1002 }); + }); + + it('treats retained coordination events as barriers', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2); + queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 50, seq: 3 }, { roomId }).queue; + queue = media(EVENTS.SEEK, { targetTime: 100, seq: 4 }, queue); + queue = media(EVENTS.PAUSE, { currentTime: 120, seq: 5 }, queue); + expect(queue).toHaveLength(3); + expect(queue[0].kind).toBe('media-intent'); + expect(queue[1].event).toBe(EVENTS.FORCE_SYNC_PREPARE); + expect(queue[2].kind).toBe('media-intent'); + expect(queue[2].intent).toMatchObject({ playbackState: 'paused', currentTime: 120 }); + }); + + it('does not persist stale liveness and command ACK frames as ordering barriers', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2); + for (const event of [EVENTS.PING, EVENTS.PONG, EVENTS.PEER_STATUS, EVENTS.EVENT_ACK]) { + const result = enqueueQueuedEvent(queue, event, { seq: 99 }, { roomId }); + expect(result.droppedStale).toBe(1); + queue = result.queue; + } + queue = media(EVENTS.PAUSE, { currentTime: 20, seq: 3 }, queue); + expect(queue).toHaveLength(1); + expect(queue[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 20 }); + }); + + it('preserves the bounded logical queue cap', () => { + let queue = []; + for (let index = 0; index < 60; index++) { + queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { index }, { roomId }).queue; + } + expect(queue).toHaveLength(50); + expect(queue[0].data.index).toBe(10); + expect(queue.at(-1).data.index).toBe(59); + }); + + it('migrates old raw media entries without crossing transactional barriers', () => { + const restored = normalizePersistedEventQueue([ + { event: EVENTS.PLAY, data: { currentTime: 10, seq: 1 } }, + { event: EVENTS.SEEK, data: { targetTime: 20, seq: 2 } }, + { event: EVENTS.PAUSE, data: { currentTime: 25, seq: 3 } }, + { event: EVENTS.EPISODE_LOBBY, data: { expectedTitle: 'S01E02' } }, + { event: EVENTS.PLAY, data: { currentTime: 30, seq: 4 } } + ], roomId); + expect(restored).toHaveLength(3); + expect(restored[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 25 }); + expect(restored[1].event).toBe(EVENTS.EPISODE_LOBBY); + expect(restored[2].intent).toMatchObject({ playbackState: 'playing', currentTime: 30 }); + expect(maxQueuedSequence(restored)).toBe(4); + }); + + it('discards stale-room intent without affecting the new room', () => { + const queue = reserve(media(EVENTS.PAUSE, { currentTime: 500, seq: 1 }), 2); + expect(hasQueuedMediaIntent(queue, roomId)).toBe(true); + expect(discardQueuedMediaIntents(queue, roomId)).toEqual([]); + expect(hasQueuedMediaIntent(queue, 'room-b')).toBe(false); + }); + + it('drops queued shared intent after controller role loss so canonical recovery can proceed', () => { + let queue = reserve(media(EVENTS.PAUSE, { currentTime: 1200, seq: 1 }), 2); + queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 1300, seq: 3 }, { roomId }).queue; + const result = reconcileQueuedRoomIntent(queue, { roomId, canControl: false }); + expect(result.queue).toEqual([]); + expect(result.discarded).toBe(2); + expect(result.hasPendingLocalIntent).toBe(false); + }); + + it('keeps an active Episode Lobby authoritative over queued media and Force Sync', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 100, seq: 1 }), 2); + queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 500, seq: 3 }, { roomId }).queue; + queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { title: 'S01E02' }, { roomId }).queue; + const result = reconcileQueuedRoomIntent(queue, { roomId, activeLobby: true }); + expect(result.queue).toEqual([{ event: EVENTS.EPISODE_READY, data: { title: 'S01E02' } }]); + expect(result.hasPendingLocalIntent).toBe(false); + }); + + it('does not let intentional solo mode retain future room-driving intent', () => { + const queue = reserve(media(EVENTS.SEEK, { targetTime: 600, playbackState: 'paused', seq: 1 }), 2); + const result = reconcileQueuedRoomIntent(queue, { roomId, desynced: true }); + expect(result.queue).toEqual([]); + expect(result.hasPendingLocalIntent).toBe(false); + }); + + it('materializes legacy events normally when an old relay has no media-state capability', () => { + expect(canonicalMediaStateFromRoomData({ + roomId, + capabilities: ['host-control', 'chat-v1'] + })).toEqual({ status: 'unsupported', mediaState: null }); + const queue = reserve(media(EVENTS.PAUSE, { currentTime: 75, seq: 20 }), 21); + expect(materializeMediaIntent(queue[0]).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PAUSE]); + expect(materializeMediaIntent(queue[0]).every(frame => + frame.data.mediaState === undefined && frame.data.revision === undefined + )).toBe(true); + }); +}); + +describe('offline media intent drain', () => { + it('counts actual wire frames and never splits an intent at a batch boundary', async () => { + let queue = enqueueQueuedEvent([], EVENTS.EPISODE_READY, { seq: 1 }, { roomId }).queue; + queue = reserve(media(EVENTS.PAUSE, { currentTime: 50, seq: 2 }, queue), 3); + const sent = []; + const first = await drainQueuedBatch(queue, { + roomId, + maxWireEvents: 2, + sendFrame: async frame => { sent.push(frame); return true; } + }); + expect(first.sentWireEvents).toBe(1); + expect(first.queue).toHaveLength(1); + expect(sent.map(frame => frame.event)).toEqual([EVENTS.EPISODE_READY]); + + const second = await drainQueuedBatch(first.queue, { + roomId, + maxWireEvents: 2, + sendFrame: async frame => { sent.push(frame); return true; } + }); + expect(second.sentWireEvents).toBe(2); + expect(second.queue).toEqual([]); + expect(sent.slice(1).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PAUSE]); + }); + + it('retains the whole logical intent after a partial send failure', async () => { + const queue = reserve(media(EVENTS.PLAY, { currentTime: 90, seq: 5 }), 6); + let calls = 0; + const result = await drainQueuedBatch(queue, { + roomId, + maxWireEvents: 10, + sendFrame: async () => ++calls === 1 + }); + expect(result.status).toBe('send_failed'); + expect(result.sentWireEvents).toBe(1); + expect(result.queue).toEqual(queue); + expect(materializeMediaIntent(result.queue[0]).map(frame => frame.data.seq)).toEqual([5, 6]); + }); + + it('drops stale-room intent during drain and preserves unrelated events', async () => { + let queue = reserve(media(EVENTS.PAUSE, { currentTime: 40, seq: 1 }), 2); + queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { seq: 3 }, { roomId }).queue; + const sent = []; + const result = await drainQueuedBatch(queue, { + roomId: 'room-b', + maxWireEvents: 10, + sendFrame: async frame => { sent.push(frame); return true; } + }); + expect(result.droppedStaleIntents).toBe(1); + expect(sent).toEqual([{ event: EVENTS.EPISODE_READY, data: { seq: 3 } }]); + }); + + it('reports logical and actual-wire queue sizes separately', () => { + let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2); + queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { seq: 3 }, { roomId }).queue; + expect(queuedMediaIntentCount(queue, roomId)).toBe(1); + expect(queuedEntryWireCount(queue[0])).toBe(2); + expect(queuedWireCount(queue)).toBe(3); + }); +}); diff --git a/scripts/test-server-ws.mjs b/scripts/test-server-ws.mjs index a8ca1c3..0efebb7 100644 --- a/scripts/test-server-ws.mjs +++ b/scripts/test-server-ws.mjs @@ -5,6 +5,11 @@ import { createRequire } from 'node:module'; import { fileURLToPath } from 'node:url'; import path from 'node:path'; import { connectionCounts, clearRateLimitMaps } from '../server/rate-limiter.js'; +import { + enqueueQueuedEvent, + materializeMediaIntent, + reserveLatestMediaIntentSequence +} from '../extension/offline-media-intent.js'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const require = createRequire(path.join(__dirname, '..', 'server', 'package.json')); @@ -70,6 +75,71 @@ try { close(); resetConnectionRate(); + // --- Combined reconnect model: compacted new-client intent over legacy wire --- + // The queue is entirely client-side. Feed its materialized PLAY/PAUSE/SEEK + // frames through the real relay and an old-client-like receiver to prove the + // optimization needs no new event, capability or ACK. + const coalescedRid = 'coalesced-media-'+Date.now(); + const legacyReceiver = await c(), coalescingSender = await c(); + await j(legacyReceiver, coalescedRid, 'legacy-receiver'); + await j(coalescingSender, coalescedRid, 'coalesce-sender', null, ['chat-v1']); + legacyReceiver._m.length = coalescingSender._m.length = 0; + + s(legacyReceiver, 'play', { currentTime: 100, seq: 1, actionTimestamp: 1 }); + await w(coalescingSender, 'play'); + const canonicalBeforeReplay = { ...mod.rooms.get(coalescedRid).mediaState }; + + let compactedQueue = enqueueQueuedEvent([], 'play', { + currentTime: 500, + seq: 10, + actionTimestamp: 10 + }, { roomId: coalescedRid }).queue; + compactedQueue = reserveLatestMediaIntentSequence(compactedQueue, coalescedRid, 11).queue; + for (const [targetTime, seq] of [[540, 12], [600, 13]]) { + compactedQueue = enqueueQueuedEvent(compactedQueue, 'seek', { + currentTime: targetTime, + targetTime, + seq, + actionTimestamp: seq + }, { roomId: coalescedRid }).queue; + } + compactedQueue = enqueueQueuedEvent(compactedQueue, 'pause', { + currentTime: 605, + seq: 14, + actionTimestamp: 14 + }, { roomId: coalescedRid }).queue; + assert.equal(compactedQueue.length, 1, 'offline playback burst is one logical queue entry'); + const replayFrames = materializeMediaIntent(compactedQueue[0]); + assert.deepEqual(replayFrames.map(frame => frame.event), ['seek', 'pause'], + 'compacted intent uses only the minimum existing legacy events'); + + const legacyReplayPayloads = []; + for (const frame of replayFrames) { + s(coalescingSender, frame.event, frame.data); + legacyReplayPayloads.push([frame.event, await w(legacyReceiver, frame.event)]); + } + assert.equal(legacyReplayPayloads[0][1].targetTime, 605); + assert.equal(legacyReplayPayloads[1][1].currentTime, 605); + for (const [event, payload] of legacyReplayPayloads) { + assert.ok(event === 'seek' || event === 'pause'); + assert.equal(payload.mediaState, undefined); + assert.equal(payload.revision, undefined); + } + const canonicalAfterReplay = mod.rooms.get(coalescedRid).mediaState; + assert.equal(canonicalAfterReplay.revision, canonicalBeforeReplay.revision + replayFrames.length); + assert.equal(canonicalAfterReplay.playbackState, 'paused'); + assert.equal(canonicalAfterReplay.currentTime, 605); + assert.equal(canonicalAfterReplay.updatedBy, 'coalesce-sender'); + + const coalescedLateJoiner = await c(); + const coalescedLateRoom = await j(coalescedLateJoiner, coalescedRid, 'coalesced-late'); + assert.equal(coalescedLateRoom.mediaState.revision, canonicalAfterReplay.revision); + assert.equal(coalescedLateRoom.mediaState.playbackState, 'paused'); + assert.equal(coalescedLateRoom.mediaState.currentTime, 605); + assert.equal(coalescedLateRoom.mediaState.updatedBy, 'coalesce-sender'); + close(); + resetConnectionRate(); + // --- Capabilities: ROOM_DATA advertises server features for client detection --- const capClient = await c(); s(capClient, 'join_room', { roomId: 'cap-'+Date.now(), peerId: 'capp', protocolVersion: '1.0.0' }); diff --git a/tests/e2e/extension.spec.mjs b/tests/e2e/extension.spec.mjs index a2b75cd..7168c43 100644 --- a/tests/e2e/extension.spec.mjs +++ b/tests/e2e/extension.spec.mjs @@ -1,4 +1,12 @@ import { test, expect } from './helpers/extension-fixture.mjs'; +import { createRequire } from 'node:module'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION } from '../../shared/constants.js'; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const require = createRequire(path.join(testDir, '..', '..', 'server', 'package.json')); +const NodeWebSocket = require('ws'); /** * Drives the packed extension itself: real background service worker, real @@ -63,6 +71,61 @@ async function applyCanonicalMediaState(context, extensionId, tabId, mediaState) }, { tabId, mediaState })); } +async function connectLegacyRelayClient(port) { + const socket = new NodeWebSocket( + `ws://127.0.0.1:${port}/socket.io/?EIO=4&transport=websocket&version=3.1.3&token=${OFFICIAL_SERVER_TOKEN}` + ); + socket.messages = []; + socket.on('message', value => socket.messages.push(value.toString())); + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error('legacy relay connection timed out')), 5000); + socket.once('open', () => { + clearTimeout(timeout); + resolve(); + }); + }); + socket.send('40'); + await expect.poll(() => socket.messages.filter(message => message.startsWith('0') || message.startsWith('40')).length).toBeGreaterThanOrEqual(2); + socket.messages.length = 0; + return socket; +} + +function sendLegacyRelayEvent(socket, event, data = {}) { + socket.send(`42${JSON.stringify([event, data])}`); +} + +async function waitForLegacyRelayEvent(socket, event, timeoutMs = 10_000) { + await expect.poll(() => socket.messages.some(message => { + if (!message.startsWith('42')) return false; + try { return JSON.parse(message.substring(2))[0] === event; } catch { return false; } + }), { timeout: timeoutMs }).toBe(true); + const index = socket.messages.findIndex(message => { + if (!message.startsWith('42')) return false; + try { return JSON.parse(message.substring(2))[0] === event; } catch { return false; } + }); + return JSON.parse(socket.messages.splice(index, 1)[0].substring(2))[1]; +} + +async function joinLegacyRelayRoom(socket, roomId, peerId) { + sendLegacyRelayEvent(socket, 'join_room', { roomId, peerId, protocolVersion: PROTOCOL_VERSION }); + return waitForLegacyRelayEvent(socket, 'room_data'); +} + +async function terminateExtensionServiceWorker(context, extensionId, page) { + const session = await context.newCDPSession(page); + try { + const { targetInfos } = await session.send('Target.getTargets'); + const target = targetInfos.find(candidate => + candidate.type === 'service_worker' + && candidate.url.startsWith(`chrome-extension://${extensionId}/`) + ); + if (!target) throw new Error('extension service worker target not found'); + await session.send('Target.closeTarget', { targetId: target.targetId }); + } finally { + await session.detach(); + } +} + async function setAudioSettings(context, extensionId, settings) { return withExtensionPage(context, extensionId, page => page.evaluate( value => chrome.storage.local.set({ audioSettings: value }), @@ -1047,6 +1110,127 @@ test('keeps controlling a Drive-style player across an ordinary play and pause', await expect.poll(() => playerFrame.locator('video').evaluate(video => video.paused)).toBe(false); }); +test('coalesces persisted offline media intent before canonical reconnect recovery', async ({ context, extensionId, baseURL }) => { + test.setTimeout(60_000); + const relay = await import('../../server/index.js'); + let legacy = null; + let lateJoiner = null; + try { + await relay.startServer(0, '127.0.0.1'); + const port = relay.httpServer.address().port; + const roomId = `e2e-coalesced-${Date.now()}`; + legacy = await connectLegacyRelayClient(port); + await joinLegacyRelayRoom(legacy, roomId, 'legacy-e2e'); + + const url = `${baseURL}/pages/simple-player.html`; + const page = await context.newPage(); + await page.goto(url); + await page.waitForFunction(() => window.__fixtureReady === true); + const { tabId } = await selectTargetTab(context, extensionId, url); + + await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async settings => { + await chrome.storage.local.set(settings); + return chrome.runtime.sendMessage({ type: 'CONNECT' }); + }, { + serverUrl: `ws://127.0.0.1:${port}`, + useCustomServer: true, + roomId, + password: '', + username: 'current-e2e' + })); + await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' })) + .toMatchObject({ status: 'connected', roomId, queuedLogicalEvents: 0 }); + + // Establish accepted server truth that would be stale for this client + // after its later offline actions. + sendLegacyRelayEvent(legacy, 'play', { currentTime: 1, seq: 1, actionTimestamp: 1 }); + sendLegacyRelayEvent(legacy, 'seek', { currentTime: 1, targetTime: 1, seq: 2, actionTimestamp: 2 }); + await expect.poll(() => relay.rooms.get(roomId)?.mediaState) + .toMatchObject({ revision: 2, playbackState: 'playing', currentTime: 1, updatedBy: 'legacy-e2e' }); + await expect.poll(() => page.locator('#player').evaluate(video => video.currentTime)).toBeGreaterThan(0.8); + legacy.messages.length = 0; + + const connectedStatus = await getExtensionState(context, extensionId, { type: 'GET_STATUS' }); + const extensionSocketId = Array.from(relay.rooms.get(roomId).peerData.entries()) + .find(([, data]) => data.peerId === connectedStatus.peerId)?.[0]; + expect(extensionSocketId).toBeTruthy(); + // Point future reconnect attempts at an unused local port, then sever + // only the extension socket. The relay and legacy peer stay live. + await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate( + serverUrl => chrome.storage.local.set({ serverUrl }), + 'ws://127.0.0.1:1' + )); + relay.io.sockets.sockets.get(extensionSocketId).disconnect(true); + await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }).then(status => status.status)) + .not.toBe('connected'); + + expect(await sendServerCommand(context, extensionId, tabId, 'play', { currentTime: 2 })).toMatchObject({ status: 'ok' }); + expect(await sendServerCommand(context, extensionId, tabId, 'seek', { targetTime: 4 })).toMatchObject({ status: 'ok' }); + expect(await sendServerCommand(context, extensionId, tabId, 'seek', { targetTime: 6 })).toMatchObject({ status: 'ok' }); + expect(await sendServerCommand(context, extensionId, tabId, 'pause', { currentTime: 6 })).toMatchObject({ status: 'ok' }); + await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' })) + .toMatchObject({ queuedLogicalEvents: 1, queuedMediaIntents: 1, queuedWireEvents: 2 }); + await expect.poll(() => page.locator('#player').evaluate(video => ({ paused: video.paused, currentTime: video.currentTime }))) + .toMatchObject({ paused: true }); + + // Terminate the actual MV3 worker. The next runtime message starts a new + // worker, which must migrate/restore the logical queue and local sequence. + await terminateExtensionServiceWorker(context, extensionId, page); + await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' }).then(status => status.queuedMediaIntents)) + .toBe(1); + const restoredQueue = await getExtensionState(context, extensionId, { type: 'GET_STATUS' }); + expect(restoredQueue.queuedLogicalEvents).toBeGreaterThanOrEqual(1); + expect(restoredQueue.queuedWireEvents).toBeGreaterThanOrEqual(2); + + await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async serverUrl => { + await chrome.storage.local.set({ serverUrl }); + chrome.alarms.create('keepAlive', { when: Date.now() + 50 }); + }, `ws://127.0.0.1:${port}`)); + + await page.locator('#player').evaluate(video => { + window.__koalaReconnectSeeks = []; + video.addEventListener('seeked', () => window.__koalaReconnectSeeks.push(video.currentTime)); + }); + const replaySeek = await waitForLegacyRelayEvent(legacy, 'seek', 20_000); + const replayPause = await waitForLegacyRelayEvent(legacy, 'pause', 20_000); + expect(replaySeek).toMatchObject({ currentTime: 6, targetTime: 6 }); + expect(replayPause).toMatchObject({ currentTime: 6 }); + expect(replaySeek.seq).toBeLessThan(replayPause.seq); + await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' })) + .toMatchObject({ status: 'connected', roomId, queuedLogicalEvents: 0, queuedMediaIntents: 0 }); + await expect.poll(() => relay.rooms.get(roomId)?.mediaState) + .toMatchObject({ playbackState: 'paused', currentTime: 6 }); + expect(relay.rooms.get(roomId).mediaState.revision).toBeGreaterThan(2); + + // The stale r2 snapshot must never have sought the local player back to 1 + // before its authorized pending intent replayed. + const reconnectSeeks = await page.evaluate(() => window.__koalaReconnectSeeks || []); + expect(reconnectSeeks.some(value => value < 3)).toBe(false); + await expect.poll(() => page.locator('#player').evaluate(video => ({ paused: video.paused, currentTime: video.currentTime }))) + .toMatchObject({ paused: true }); + + lateJoiner = await connectLegacyRelayClient(port); + const lateRoom = await joinLegacyRelayRoom(lateJoiner, roomId, 'late-e2e'); + expect(lateRoom.mediaState).toMatchObject({ + revision: relay.rooms.get(roomId).mediaState.revision, + playbackState: 'paused', + currentTime: 6 + }); + + await page.waitForTimeout(700); + const leakedMediaEvents = legacy.messages.filter(message => { + if (!message.startsWith('42')) return false; + try { return ['play', 'pause', 'seek'].includes(JSON.parse(message.substring(2))[0]); } catch { return false; } + }); + expect(leakedMediaEvents).toEqual([]); + } finally { + await context.setOffline(false).catch(() => {}); + try { legacy?.close(); } catch { /* already closed */ } + try { lateJoiner?.close(); } catch { /* already closed */ } + await relay.stopServerForTests(); + } +}); + function FRAMED_VIDEO_PAUSED() { return document.querySelector('iframe').contentDocument.querySelector('video').paused; }