From a477356d2505ee61554aeb48508126f56607a3dd Mon Sep 17 00:00:00 2001 From: KoalaDev <6156589+Shik3i@users.noreply.github.com> Date: Thu, 20 Aug 2026 22:39:38 +0200 Subject: [PATCH] fix(sync): harden media state reconnect races --- docs/ARCHITECTURE.md | 10 +- docs/PROTOCOL.md | 16 +- extension/background.js | 380 ++++++++++++------ .../canonical-media-state-background.test.mjs | 6 +- extension/canonical-media-state.js | 56 ++- extension/canonical-media-state.test.mjs | 27 ++ .../offline-media-intent-background.test.mjs | 21 +- extension/offline-media-intent.js | 125 +++++- extension/offline-media-intent.test.mjs | 91 ++++- scripts/test-server-ws.mjs | 90 ++++- server/index.js | 63 ++- tests/e2e/extension.spec.mjs | 5 +- 12 files changed, 697 insertions(+), 193 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3c86ff8..d81f198 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -45,11 +45,13 @@ the room/revision, respects Host Control solo mode and Episode Lobby, then sends internal `APPLY_CANONICAL_MEDIA_STATE` message to the existing content/video path. That path reuses frame election, Netflix/Disney page-API seeks, native play/pause, the 2-second drift tolerance, and programmatic-event suppression. The apply is -one-shot recovery: it creates no action history, notification, command ACK, or -relay media event. +one-shot recovery: a pending playing snapshot advances from its local receipt +time while waiting for a target, and the apply creates no action history, +notification, command ACK, or 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 +Force Sync remains a two-phase ACK protocol. A valid `PREPARE` is temporary +room-wide choreography; the next authorized `EXECUTE` commits the latest target +visible to peers to canonical state. Per-sender `seq`, peer heartbeats, and the reconnect queue remain separate mechanisms. ## 3.2 Offline Media Intent diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 8cf7e42..5b5ccad 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -119,8 +119,10 @@ Only accepted, sanitized room controls update canonical state: - `pause` uses its valid `currentTime`, or freezes an existing effective position. - `seek` prefers `targetTime` (with `currentTime` compatibility) and preserves the established playback state. -- `force_sync_prepare` records only temporary coordination state. Its matching - `force_sync_execute` commits the prepared target as playing. +- a valid `force_sync_prepare` records only temporary coordination state. The + next authorized `force_sync_execute` commits the latest room-wide prepared + target as playing. The latest valid prepare is also the only post-demotion + execute exemption in Host Control mode. `peer_status` heartbeats are observations and never rewrite canonical intent. Per-sender `seq` still orders commands from one sender; canonical `revision` @@ -130,7 +132,8 @@ On join/reconnect, a capable extension applies a valid snapshot once through an extension-internal recovery message. Existing seek/page-API and native-event suppression prevent `play`, `pause`, or `seek` echoes. Pending recovery is scoped to the room/revision in `chrome.storage.session`, waits for the selected media -target lifecycle, and is cleared on leave/switch. Intentional host-only guest +target lifecycle, and projects a still-playing snapshot from its local receipt +time before a delayed apply. It is cleared on leave/switch. Intentional host-only guest desync and an active Episode Lobby take precedence over snapshot recovery. Compatibility is additive: new clients use old behavior with a relay that omits @@ -329,10 +332,11 @@ them with the same sanitized relay envelope as other room events, including ### `force_sync_execute` The current extension sends sequence/action metadata but no target; the relay uses -the validated target retained from the matching `force_sync_prepare`. In +the latest validated room target retained from `force_sync_prepare`. In `host-only` mode, only controllers may send it. -The relay also allows a matching initiator's execute event after that initiator -started the prepare step, even if their controller state changed before execute. +The relay also allows that latest valid initiator's execute event after their +controller state changed before execute. Invalid prepares are dropped and grant +no exemption. ## Episode Lobby diff --git a/extension/background.js b/extension/background.js index 24742d5..cd952e4 100644 --- a/extension/background.js +++ b/extension/background.js @@ -80,6 +80,7 @@ chrome.runtime.onStartup.addListener(() => { // --- State Management --- let socket = null; +let connectionGeneration = 0; let isConnecting = false; let peerId = null; // initialized via getPeerId() let currentRoom = null; @@ -249,7 +250,7 @@ let pendingHistory = []; let eventQueue = []; let eventQueueVersion = 0; let flushTimer = null; // paces draining of eventQueue after (re)connect -let flushInProgress = false; +let flushInProgress = null; let isNamespaceJoined = false; let awaitingRoomData = false; let pendingRoomDataRoomId = null; @@ -382,9 +383,11 @@ function ensureState() { if (!restorationTask) { restorationTask = new Promise(resolve => { let resolved = false; + let restorationTimedOut = false; const done = () => { if (!resolved) { resolved = true; resolve(); } }; const storageTimeout = setTimeout(() => { + restorationTimedOut = true; addLog('Storage restoration timed out, continuing with defaults', 'warn'); storageInitialized = true; done(); @@ -399,6 +402,9 @@ function ensureState() { 'episodeLobby', 'localSeq', 'lastSeqBySender', 'expectedAcksCount', 'roomIdleSince', 'lastContentHeartbeatAt', 'hcmDesynced', 'chatActivityTimeline', 'canonicalMediaRecovery' ], (data) => { + // A late callback must not resurrect a room, queue or canonical + // snapshot after the worker already continued with defaults. + if (restorationTimedOut) return; clearTimeout(storageTimeout); if (data.expectedAcksCount !== undefined) expectedAcksCount = data.expectedAcksCount; if (data.currentTabId !== undefined) currentTabId = normalizeTabId(data.currentTabId); @@ -720,7 +726,8 @@ function resolveServerUrl(settings) { return (settings.serverUrl && settings.useCustomServer) ? settings.serverUrl : OFFICIAL_SERVER_URL; } -function forceDisconnect() { +function forceDisconnect({ preserveEventQueue = false } = {}) { + connectionGeneration++; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; @@ -755,15 +762,17 @@ function forceDisconnect() { lastContentHeartbeatAt = null; forceSyncAcks.clear(); if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; } - flushInProgress = false; - eventQueue = []; - eventQueueVersion++; + flushInProgress = null; + if (!preserveEventQueue) { + eventQueue = []; + eventQueueVersion++; + } chrome.storage.session.set({ isForceSyncInitiator: false, forceSyncAcks: [], forceSyncDeadline: null, expectedAcksCount: 0, - eventQueue: [], + eventQueue, episodeLobby: null, roomIdleSince: null, lastContentHeartbeatAt: null @@ -1096,6 +1105,8 @@ async function leaveRoomAfterIdleGrace(reason) { async function connect() { if (isConnecting) return; isConnecting = true; + const startingGeneration = connectionGeneration; + let attemptGeneration = startingGeneration; let finalUrl = ''; try { @@ -1104,6 +1115,7 @@ async function connect() { try { if (!peerId) peerId = await getPeerId(); settings = await getSettings(); + if (startingGeneration !== connectionGeneration) return; pendingRoomDataRoomId = settings.roomId || currentRoom?.roomId || null; } catch (e) { throw new Error(`[Storage Error] ${e.message}`); @@ -1170,114 +1182,125 @@ async function connect() { url.searchParams.set('version', chrome.runtime.getManifest().version); url.searchParams.set('token', OFFICIAL_SERVER_TOKEN); - socket = new WebSocket(url.toString()); + const generation = ++connectionGeneration; + attemptGeneration = generation; + const connectionSocket = new WebSocket(url.toString()); + socket = connectionSocket; + + // --- Phase 5: Event Listeners --- + connectionSocket.onopen = () => { + if (generation !== connectionGeneration || socket !== connectionSocket) return; + reconnectAttempts = 0; + reconnectStartTime = null; + reconnectFailed = false; + addLog('WebSocket Connection Opened', 'success'); + chrome.storage.session.set({ reconnectFailed: false, reconnectAttempts: 0, reconnectStartTime: null }).catch(() => {}); + isNamespaceJoined = false; + connectionSocket.send('40'); + }; + + connectionSocket.onmessage = async (event) => { + if (generation !== connectionGeneration || socket !== connectionSocket) return; + await ensureState(); + if (generation !== connectionGeneration || socket !== connectionSocket) return; + const msg = event.data; + if (msg === '2') { + connectionSocket.send('3'); + return; + } + if (msg.startsWith('0')) { + addLog(`Socket.IO Handshake: ${msg}`, 'info'); + } else if (msg.startsWith('40')) { + isConnecting = false; + isNamespaceJoined = true; + broadcastConnectionStatus('connected'); + startPing(); + addLog('Joined Namespace /', 'success'); + const joinedSettings = await getSettings(); + if (generation !== connectionGeneration || socket !== connectionSocket) return; + if (joinedSettings.roomId) { + awaitingRoomData = true; + pendingRoomDataRoomId = joinedSettings.roomId; + const sharedTitles = getSharedTitleFields(joinedSettings); + emit(EVENTS.JOIN_ROOM, { + roomId: joinedSettings.roomId, + password: joinedSettings.password, + peerId, + username: joinedSettings.username, + tabTitle: sharedTitles.tabTitle, + clientCapabilities: CLIENT_CAPABILITIES, + protocolVersion: PROTOCOL_VERSION + }); + } else { + awaitingRoomData = false; + pendingRoomDataRoomId = null; + flushEventQueue(); + } + } else if (msg.startsWith('42')) { + try { + const payload = JSON.parse(msg.substring(2)); + try { + await handleServerEvent(payload[0], payload[1], generation); + } catch (handlerErr) { + addLog(`Handler error for ${payload[0]}: ${handlerErr.message}`, 'error'); + } + } catch (_e) { + addLog(`Failed to parse message: ${msg}`, 'error'); + } + } + }; + + connectionSocket.onclose = () => { + if (generation !== connectionGeneration || socket !== connectionSocket) return; + // Invalidate any async message handler that began before the + // close event and is still suspended at an await boundary. + connectionGeneration++; + isConnecting = false; + isNamespaceJoined = false; + awaitingRoomData = false; + pendingRoomDataRoomId = null; + invalidateChatSession(); + stopPing(); + if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; } + + if (!connectIntent && !currentRoom) { + isForceSyncInitiator = false; + forceSyncAcks.clear(); + if (forceSyncTimeout) clearTimeout(forceSyncTimeout); + chrome.storage.session.set({ + isForceSyncInitiator: false, + forceSyncAcks: [], + forceSyncDeadline: null + }).catch(() => {}); + } + + if (currentRoom && !connectIntent) { + currentRoom.peers = []; + if (storageInitialized) chrome.storage.session.set({ currentRoom }).catch(() => {}); + chrome.runtime.sendMessage({ type: 'PEER_UPDATE', peers: [] }).catch(() => {}); + } + broadcastConnectionStatus('disconnected'); + socket = null; + if (currentRoom || connectIntent) { + addLog('Disconnected. Scheduling reconnect...', 'warn'); + scheduleReconnect(); + } else { + addLog('Disconnected. No active session — staying disconnected.', 'info'); + } + }; + + connectionSocket.onerror = () => { + if (generation !== connectionGeneration || socket !== connectionSocket) return; + broadcastConnectionStatus('disconnected'); + const logType = reconnectAttempts > 1 ? 'error' : 'warn'; + addLog('WebSocket Error: Connection failed', logType); + }; } catch (e) { throw new Error(`[Connection Error] ${e.message}`); } - // --- Phase 5: Event Listeners --- - socket.onopen = () => { - reconnectAttempts = 0; - reconnectStartTime = null; - reconnectFailed = false; - addLog('WebSocket Connection Opened', 'success'); - chrome.storage.session.set({ reconnectFailed: false, reconnectAttempts: 0, reconnectStartTime: null }).catch(() => {}); - isNamespaceJoined = false; - socket.send('40'); - }; - - socket.onmessage = async (event) => { - await ensureState(); - const msg = event.data; - if (msg === '2') { - socket.send('3'); - return; - } - if (msg.startsWith('0')) { - addLog(`Socket.IO Handshake: ${msg}`, 'info'); - } else if (msg.startsWith('40')) { - isConnecting = false; - isNamespaceJoined = true; - broadcastConnectionStatus('connected'); - startPing(); - 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, - password: settings.password, - peerId, - username: settings.username, - tabTitle: sharedTitles.tabTitle, - clientCapabilities: CLIENT_CAPABILITIES, - protocolVersion: PROTOCOL_VERSION - }); - } else { - awaitingRoomData = false; - pendingRoomDataRoomId = null; - flushEventQueue(); - } - } else if (msg.startsWith('42')) { - try { - const payload = JSON.parse(msg.substring(2)); - try { - await handleServerEvent(payload[0], payload[1]); - } catch (handlerErr) { - addLog(`Handler error for ${payload[0]}: ${handlerErr.message}`, 'error'); - } - } catch (_e) { - addLog(`Failed to parse message: ${msg}`, 'error'); - } - } - }; - - socket.onclose = () => { - isConnecting = false; - isNamespaceJoined = false; - awaitingRoomData = false; - pendingRoomDataRoomId = null; - invalidateChatSession(); - stopPing(); - if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; } - - if (!connectIntent && !currentRoom) { - isForceSyncInitiator = false; - forceSyncAcks.clear(); - if (forceSyncTimeout) clearTimeout(forceSyncTimeout); - chrome.storage.session.set({ - isForceSyncInitiator: false, - forceSyncAcks: [], - forceSyncDeadline: null - }).catch(() => {}); - } - - - if (currentRoom && !connectIntent) { - currentRoom.peers = []; - if (storageInitialized) chrome.storage.session.set({ currentRoom }).catch(() => {}); - chrome.runtime.sendMessage({ type: 'PEER_UPDATE', peers: [] }).catch(() => {}); - } - broadcastConnectionStatus('disconnected'); - if (currentRoom || connectIntent) { - addLog('Disconnected. Scheduling reconnect...', 'warn'); - socket = null; - scheduleReconnect(); - } else { - addLog('Disconnected. No active session — staying disconnected.', 'info'); - socket = null; - } - }; - - socket.onerror = () => { - broadcastConnectionStatus('disconnected'); - const logType = reconnectAttempts > 1 ? 'error' : 'warn'; - addLog('WebSocket Error: Connection failed', logType); - }; - } catch (e) { + if (attemptGeneration !== connectionGeneration) return; isConnecting = false; const logType = reconnectAttempts > 1 ? 'error' : 'warn'; const errMsg = (e && e.message) ? e.message : String(e || 'Unknown connection error'); @@ -1517,7 +1540,10 @@ function emit(event, data) { const mustWaitForRoomData = awaitingRoomData && event !== EVENTS.JOIN_ROOM && event !== EVENTS.GET_ROOMS; - if (socket && socket.readyState === WebSocket.OPEN && isNamespaceJoined && !mustWaitForRoomData) { + if (socket && socket.readyState === WebSocket.OPEN + && isNamespaceJoined + && !mustWaitForRoomData + && !flushInProgress) { try { const msg = encodeSocketEvent(event, data, chatSecretGuard); socket.send(msg); @@ -1606,24 +1632,49 @@ function applyQueuedRoomPolicy(roomId, policy, reason) { * 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. + * full if either send fails. Replaying its first legacy frame can repeat a + * canonical revision/activity side effect, but the ordered full retry remains + * final-state recoverable without a new delivery-ACK protocol. */ -async function flushEventQueue() { +async function flushEventQueue(replaySettingsOverride = undefined) { if (flushTimer || flushInProgress || awaitingRoomData) return; if (!socket || socket.readyState !== WebSocket.OPEN || !isNamespaceJoined) return; - flushInProgress = true; + const flushToken = {}; + const flushConnectionGeneration = connectionGeneration; + const flushSocket = socket; + const flushRoomId = currentRoom?.roomId || null; + flushInProgress = flushToken; 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; + // arrives during the drain, keep it behind the original logical entries; + // final-state retry is safer than losing concurrent intent. + const replaySettings = replaySettingsOverride !== undefined + ? replaySettingsOverride + : (eventQueue.some(isQueuedMediaIntent) ? await getSettings() : null); + if (flushConnectionGeneration !== connectionGeneration + || socket !== flushSocket + || currentRoom?.roomId !== flushRoomId + || awaitingRoomData) { + return; + } + applyQueuedRoomPolicy(flushRoomId, { + canControl: !(controlMode === CONTROL_MODES.HOST_ONLY && hostPeerId && !amController()), + activeLobby: !!episodeLobby, + desynced: hcmDesynced, + authoritativeLobby: !!currentRoom?.activeLobby + }, 'Queue replay authority changed'); const drainSource = eventQueue; const drainVersion = eventQueueVersion; const result = await drainQueuedBatch(drainSource, { roomId: currentRoom?.roomId || null, maxWireEvents: FLUSH_BATCH_SIZE, sendFrame: async (frame, entry) => { + if (flushConnectionGeneration !== connectionGeneration + || socket !== flushSocket + || currentRoom?.roomId !== flushRoomId + || awaitingRoomData) { + return false; + } let payload = frame.data && typeof frame.data === 'object' ? { ...frame.data } : {}; if (isQueuedMediaIntent(entry)) { payload = withTitlePrivacy(payload, replaySettings, ['mediaTitle']); @@ -1654,14 +1705,19 @@ async function flushEventQueue() { 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) { + if (eventQueue.length > 0 + && flushConnectionGeneration === connectionGeneration + && socket === flushSocket + && socket?.readyState === WebSocket.OPEN + && isNamespaceJoined + && !awaitingRoomData) { flushTimer = setTimeout(() => { flushTimer = null; flushEventQueue().catch(error => addLog(`Queue replay failed: ${error.message}`, 'warn')); }, FLUSH_BATCH_INTERVAL_MS); } } finally { - flushInProgress = false; + if (flushInProgress === flushToken) flushInProgress = null; } } @@ -1739,7 +1795,7 @@ function markCanonicalMediaStateHandled(roomId, revision) { async function tryApplyPendingCanonicalMediaState() { const roomId = currentRoom?.roomId; - const pending = canonicalMediaStateTracker.getPending(roomId); + const pending = canonicalMediaStateTracker.getPendingProjected(roomId); if (!pending || !roomId) return { status: 'none' }; const { mediaState } = pending; @@ -1756,6 +1812,9 @@ async function tryApplyPendingCanonicalMediaState() { const tabId = normalizeTabId(currentTabId); if (tabId === null || currentTargetHasVideo !== true) return { status: 'pending_no_target' }; + const targetGeneration = targetActivationGeneration; + const targetFrameId = normalizeFrameId(currentTargetFrameId); + const targetDocumentId = currentTargetDocumentId; try { const response = await sendMessageToContentTab(tabId, { @@ -1763,6 +1822,11 @@ async function tryApplyPendingCanonicalMediaState() { mediaState }); if (currentRoom?.roomId !== roomId) return { status: 'stale_room' }; + if (!isCurrentTargetIdentity(tabId, targetGeneration) + || normalizeFrameId(currentTargetFrameId) !== targetFrameId + || currentTargetDocumentId !== targetDocumentId) { + return { status: 'stale_target' }; + } const latestPending = canonicalMediaStateTracker.getPending(roomId); if (latestPending?.mediaState.revision !== mediaState.revision) return { status: 'superseded' }; @@ -1821,7 +1885,8 @@ async function handleCanonicalRoomData(data, hasPendingLocalIntent) { } // --- Event Handlers --- -async function handleServerEvent(event, data) { +async function handleServerEvent(event, data, expectedConnectionGeneration = connectionGeneration) { + if (expectedConnectionGeneration !== connectionGeneration) return; if (!data) { addLog(`Ignored server event ${event} due to empty payload`, 'warn'); return; @@ -1841,6 +1906,10 @@ async function handleServerEvent(event, data) { } switch (event) { case EVENTS.ROOM_DATA: { + if (pendingRoomDataRoomId && data.roomId !== pendingRoomDataRoomId) { + addLog(`Ignored stale ROOM_DATA for ${data.roomId}`, 'warn'); + return; + } if (currentRoom?.roomId !== data.roomId) { invalidateChatSession(); clearChatActivity(); @@ -1870,23 +1939,40 @@ async function handleServerEvent(event, data) { currentRoom.peers = []; } - // Recover server-tracked active Episode Lobby if present - if (!data?.activeLobby && episodeLobby) { + // ROOM_DATA is authoritative for an already-active server lobby, + // but a locally-created offline lobby has not reached the relay + // yet and must remain owned by its initiator until queued replay. + const hasQueuedLocalLobby = eventQueue.some(entry => + entry?.event === EVENTS.EPISODE_LOBBY + && (!entry.roomId || entry.roomId === data.roomId) + ); + if (!data?.activeLobby && episodeLobby && !hasQueuedLocalLobby) { clearEpisodeLobbyState(); addLog('Discarded stale local Episode Lobby after ROOM_DATA confirmed it ended', 'info'); - } else if (data && data.activeLobby && !episodeLobby) { + } else if (data?.activeLobby) { + const sameLobby = episodeLobby + && episodeLobby.expectedTitle === data.activeLobby.expectedTitle + && episodeLobby.initiatorPeerId === data.activeLobby.initiatorPeerId; + if (!sameLobby && episodeLobbyTimeout) { + clearTimeout(episodeLobbyTimeout); + episodeLobbyTimeout = null; + } episodeLobby = { expectedTitle: data.activeLobby.expectedTitle, initiatorPeerId: data.activeLobby.initiatorPeerId, readyPeers: data.activeLobby.readyPeers, - createdAt: Date.now() + createdAt: sameLobby && Number.isFinite(episodeLobby.createdAt) + ? episodeLobby.createdAt + : Date.now() }; persistEpisodeLobby(); broadcastLobbyUpdate(); - addLog(`Recovered active episode lobby from server: "${episodeLobby.expectedTitle}"`, 'info'); + if (!sameLobby) { + addLog(`Recovered active episode lobby from server: "${episodeLobby.expectedTitle}"`, 'info'); + } // Notify content script to start polling - if (currentTabId) { + if (!sameLobby && currentTabId) { const tabId = parseInt(currentTabId); if (!isNaN(tabId)) { sendMessageToCurrentContent({ @@ -1908,6 +1994,15 @@ async function handleServerEvent(event, data) { // Inform Website Bridge & Popup const joinStatusMsg = { type: 'JOIN_STATUS', success: true, message: 'Joined' }; await broadcastJoinStatus(joinStatusMsg); + if (expectedConnectionGeneration !== connectionGeneration) return; + // Resolve replay privacy before declaring ROOM_DATA complete. Local + // commands remain queued during this await, and any intervening + // role/lobby event is reflected by the policy below before the + // canonical precedence decision is made. + const replaySettings = eventQueue.some(isQueuedMediaIntent) + ? await getSettings() + : null; + if (expectedConnectionGeneration !== connectionGeneration) return; awaitingRoomData = false; pendingRoomDataRoomId = null; @@ -1917,13 +2012,14 @@ async function handleServerEvent(event, data) { const queuePolicy = applyQueuedRoomPolicy(data.roomId, { canControl: !lostRoomAuthority, activeLobby: !!episodeLobby, - desynced: hcmDesynced + desynced: hcmDesynced, + authoritativeLobby: !!data.activeLobby }, lostRoomAuthority ? 'Host Control role changed while offline' : (episodeLobby ? 'Active Episode Lobby takes precedence' : 'Reconnect queue policy')); await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent); - await flushEventQueue(); + await flushEventQueue(replaySettings); break; } case EVENTS.CONTROL_MODE: @@ -2026,6 +2122,10 @@ async function handleServerEvent(event, data) { case EVENTS.PAUSE: case EVENTS.SEEK: case EVENTS.FORCE_SYNC_PREPARE: + if (event === EVENTS.FORCE_SYNC_PREPARE && episodeLobby) { + if (currentRoom) currentRoom.activeLobby = null; + clearEpisodeLobbyState(); + } if (data.senderId && typeof data.seq === 'number') { const lastSeq = lastSeqBySender[data.senderId]; if (lastSeq !== undefined && data.seq <= lastSeq) { @@ -2087,6 +2187,10 @@ async function handleServerEvent(event, data) { } break; case EVENTS.FORCE_SYNC_EXECUTE: + if (episodeLobby) { + if (currentRoom) currentRoom.activeLobby = null; + clearEpisodeLobbyState(); + } if (data?.senderId && typeof data.seq === 'number') { const lastSeq = lastSeqBySender[data.senderId]; if (lastSeq !== undefined && data.seq <= lastSeq) break; @@ -2228,6 +2332,14 @@ async function handleServerEvent(event, data) { break; case EVENTS.EPISODE_LOBBY: if (data.senderId && data.expectedTitle) { + if (currentRoom) { + currentRoom.activeLobby = { + expectedTitle: data.expectedTitle, + initiatorPeerId: data.senderId, + readyPeers: [data.senderId] + }; + if (storageInitialized) chrome.storage.session.set({ currentRoom }); + } addLog(`Episode lobby from ${data.senderId}: "${data.expectedTitle}"`, 'info'); // If we already have a lobby for this same title, treat as dedup if (episodeLobby && sameEpisode(episodeLobby.expectedTitle, data.expectedTitle)) { @@ -2269,9 +2381,17 @@ async function handleServerEvent(event, data) { addLog(`Episode ready from ${data.senderId} (${episodeLobby.readyPeers.length})`, 'info'); checkEpisodeLobbyCompletion(); } + if (currentRoom?.activeLobby) { + currentRoom.activeLobby.readyPeers = [...episodeLobby.readyPeers]; + if (storageInitialized) chrome.storage.session.set({ currentRoom }); + } } break; case EVENTS.EPISODE_LOBBY_CANCEL: + if (currentRoom) { + currentRoom.activeLobby = null; + if (storageInitialized) chrome.storage.session.set({ currentRoom }); + } if (episodeLobby) { const title = episodeLobby.expectedTitle; clearEpisodeLobbyState(); @@ -4078,7 +4198,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { reconnectStartTime = null; reconnectAttempts = 0; chrome.storage.session.set({ reconnectFailed: false, reconnectAttempts: 0, reconnectStartTime: null }); - forceDisconnect(); + forceDisconnect({ preserveEventQueue: true }); connect(); sendResponse({ status: 'ok' }); } else if (message.type === 'GET_STATUS') { diff --git a/extension/canonical-media-state-background.test.mjs b/extension/canonical-media-state-background.test.mjs index 67024ac..0066af8 100644 --- a/extension/canonical-media-state-background.test.mjs +++ b/extension/canonical-media-state-background.test.mjs @@ -57,6 +57,10 @@ describe('canonical ROOM_DATA recovery contract', () => { expect(backgroundSource).toContain('tryApplyPendingCanonicalMediaState().catch(() => {})'); expect(backgroundSource).toMatch(/currentTargetHasVideo\) \{\s*await tryApplyPendingCanonicalMediaState\(\)/); expect(backgroundSource.match(/clearCanonicalMediaRecovery\(\)/g)?.length).toBeGreaterThanOrEqual(4); + const apply = functionBody(backgroundSource, 'tryApplyPendingCanonicalMediaState', 'handleCanonicalRoomData'); + expect(apply).toContain('getPendingProjected(roomId)'); + expect(apply).toContain('targetActivationGeneration'); + expect(apply).toContain("return { status: 'stale_target' }"); }); it('protects intentional desync, active Episode Lobby and queued reconnect intent', () => { @@ -67,7 +71,7 @@ describe('canonical ROOM_DATA recovery contract', () => { expect(roomData).toContain('if (hasPendingLocalIntent)'); expect(backgroundSource).toContain('awaitingRoomData = true'); expect(backgroundSource).toContain('await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)'); - expect(backgroundSource).toContain('await flushEventQueue()'); + expect(backgroundSource).toContain('await flushEventQueue(replaySettings)'); }); it('reuses existing seek abstractions, suppression and drift tolerance', () => { diff --git a/extension/canonical-media-state.js b/extension/canonical-media-state.js index 6993585..2e2f6d1 100644 --- a/extension/canonical-media-state.js +++ b/extension/canonical-media-state.js @@ -35,6 +35,22 @@ export function canonicalMediaStateFromRoomData(roomData) { : { status: 'invalid', mediaState: null }; } +export function projectCanonicalMediaState(mediaState, receivedAt, now = Date.now()) { + const validated = validateCanonicalMediaState(mediaState); + if (!validated) return null; + if (validated.playbackState !== 'playing') return validated; + const received = typeof receivedAt === 'number' && Number.isFinite(receivedAt) + ? receivedAt + : now; + return { + ...validated, + currentTime: Math.min( + MAX_MEDIA_TIME, + validated.currentTime + Math.max(0, now - received) / 1000 + ) + }; +} + function normalizeRoomId(roomId) { return typeof roomId === 'string' && roomId ? roomId : null; } @@ -60,11 +76,16 @@ export function createCanonicalMediaStateTracker() { beginRecovery(nextRoomId) { adoptRoom(nextRoomId); + // A relay restart or an empty-room recreation starts a new in-memory + // revision epoch. ROOM_DATA belongs to this fresh connection, so a + // lower revision is current truth rather than a stale packet from + // the previous epoch. + knownRevision = 0; appliedRevision = 0; pending = null; }, - receive(nextRoomId, value) { + receive(nextRoomId, value, receivedAt = Date.now()) { adoptRoom(nextRoomId); const mediaState = validateCanonicalMediaState(value); if (!roomId || !mediaState) return { status: 'invalid' }; @@ -74,7 +95,13 @@ export function createCanonicalMediaStateTracker() { return { status: 'duplicate' }; } knownRevision = Math.max(knownRevision, mediaState.revision); - pending = { roomId, mediaState }; + pending = { + roomId, + mediaState, + receivedAt: typeof receivedAt === 'number' && Number.isFinite(receivedAt) + ? receivedAt + : Date.now() + }; return { status: 'pending', mediaState }; }, @@ -84,6 +111,16 @@ export function createCanonicalMediaStateTracker() { : null; }, + getPendingProjected(nextRoomId = roomId, now = Date.now()) { + if (!pending || pending.roomId !== normalizeRoomId(nextRoomId)) return null; + const mediaState = projectCanonicalMediaState( + pending.mediaState, + pending.receivedAt, + now + ); + return mediaState ? { roomId: pending.roomId, mediaState } : null; + }, + markHandled(nextRoomId, revision) { if (normalizeRoomId(nextRoomId) !== roomId || !Number.isSafeInteger(revision) @@ -114,7 +151,14 @@ export function createCanonicalMediaStateTracker() { && restoredPending && restoredPending.revision >= appliedRevision && restoredPending.revision >= knownRevision) { - pending = { roomId, mediaState: restoredPending }; + pending = { + roomId, + mediaState: restoredPending, + receivedAt: typeof value.pending.receivedAt === 'number' + && Number.isFinite(value.pending.receivedAt) + ? value.pending.receivedAt + : Date.now() + }; knownRevision = restoredPending.revision; } return true; @@ -125,7 +169,11 @@ export function createCanonicalMediaStateTracker() { roomId, knownRevision, appliedRevision, - pending: pending ? { roomId: pending.roomId, mediaState: { ...pending.mediaState } } : null + pending: pending ? { + roomId: pending.roomId, + mediaState: { ...pending.mediaState }, + receivedAt: pending.receivedAt + } : null }; } }; diff --git a/extension/canonical-media-state.test.mjs b/extension/canonical-media-state.test.mjs index de556a6..a1fe540 100644 --- a/extension/canonical-media-state.test.mjs +++ b/extension/canonical-media-state.test.mjs @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'; import { canonicalMediaStateFromRoomData, createCanonicalMediaStateTracker, + projectCanonicalMediaState, validateCanonicalMediaState } from './canonical-media-state.js'; @@ -81,6 +82,15 @@ describe('ROOM_DATA capability compatibility', () => { }); describe('canonical media state tracker', () => { + it('projects a deferred playing snapshot from local receipt time and clamps it', () => { + expect(projectCanonicalMediaState(state(1, 100, 'playing'), 1_000, 31_000)) + .toMatchObject({ currentTime: 130, playbackState: 'playing' }); + expect(projectCanonicalMediaState(state(1, 100, 'paused'), 1_000, 31_000)) + .toMatchObject({ currentTime: 100, playbackState: 'paused' }); + expect(projectCanonicalMediaState(state(1, 86_390, 'playing'), 1_000, 31_000).currentTime) + .toBe(86_400); + }); + it('accepts a valid snapshot and applies each revision once', () => { const tracker = createCanonicalMediaStateTracker(); expect(tracker.receive('room-a', state(1)).status).toBe('pending'); @@ -116,6 +126,23 @@ describe('canonical media state tracker', () => { expect(tracker.getPending().mediaState.currentTime).toBe(100); }); + it('accepts a lower revision from a new relay or room epoch after reconnect', () => { + const tracker = createCanonicalMediaStateTracker(); + tracker.receive('room-a', state(12), 1_000); + tracker.markHandled('room-a', 12); + tracker.beginRecovery('room-a'); + expect(tracker.receive('room-a', state(1), 2_000).status).toBe('pending'); + expect(tracker.getPending('room-a').mediaState.revision).toBe(1); + }); + + it('persists receipt time so MV3 recovery projects only playing snapshots', () => { + const first = createCanonicalMediaStateTracker(); + first.receive('room-a', state(4, 50, 'playing'), 10_000); + const restored = createCanonicalMediaStateTracker(); + expect(restored.restore(first.snapshot(), 'room-a')).toBe(true); + expect(restored.getPendingProjected('room-a', 15_000).mediaState.currentTime).toBe(55); + }); + it('restores only room-scoped session state', () => { const first = createCanonicalMediaStateTracker(); first.receive('room-a', state(9)); diff --git a/extension/offline-media-intent-background.test.mjs b/extension/offline-media-intent-background.test.mjs index 882e848..71c5abf 100644 --- a/extension/offline-media-intent-background.test.mjs +++ b/extension/offline-media-intent-background.test.mjs @@ -16,6 +16,7 @@ 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('&& !flushInProgress'); expect(emit).toContain('socket.send(msg)'); expect(emit).toContain('queueEvent(event, data)'); expect(emit).not.toContain('setTimeout'); @@ -27,6 +28,7 @@ describe('offline media intent background integration', () => { 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'); + expect(backgroundSource).toContain('if (restorationTimedOut) return'); }); it('reconciles Host Control, Episode Lobby and solo mode before canonical recovery and replay', () => { @@ -36,10 +38,11 @@ describe('offline media intent background integration', () => { expect(roomData.indexOf('applyQueuedRoomPolicy(data.roomId')) .toBeLessThan(roomData.indexOf('handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)')); expect(roomData.indexOf('handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)')) - .toBeLessThan(roomData.indexOf('flushEventQueue()')); + .toBeLessThan(roomData.indexOf('flushEventQueue(replaySettings)')); expect(roomData).toContain('activeLobby: !!episodeLobby'); expect(roomData).toContain('desynced: hcmDesynced'); - expect(roomData).toContain('if (!data?.activeLobby && episodeLobby)'); + expect(roomData).toContain('if (!data?.activeLobby && episodeLobby && !hasQueuedLocalLobby)'); + expect(roomData).toContain('authoritativeLobby: !!data.activeLobby'); }); it('clears queued room intent on failed join, leave and room switch paths', () => { @@ -51,6 +54,11 @@ describe('offline media intent background integration', () => { backgroundSource.indexOf("message.type === 'CLEAR_LOGS'") ); expect(leaveHandler).toContain('forceDisconnect()'); + const retryHandler = backgroundSource.slice( + backgroundSource.indexOf("message.type === 'RETRY_CONNECT'"), + backgroundSource.indexOf("message.type === 'GET_STATUS'") + ); + expect(retryHandler).toContain('forceDisconnect({ preserveEventQueue: true })'); }); it('paces actual frames through a failure-retaining logical drain', () => { @@ -61,9 +69,18 @@ describe('offline media intent background integration', () => { 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).toContain('flushConnectionGeneration !== connectionGeneration'); expect(flush).not.toMatch(/eventQueue\.shift\(\)[\s\S]*emit\(/); }); + it('generation-scopes socket callbacks and stale ROOM_DATA work', () => { + expect(backgroundSource).toContain('let connectionGeneration = 0'); + expect(backgroundSource).toContain('socket !== connectionSocket'); + expect(backgroundSource).toContain('handleServerEvent(payload[0], payload[1], generation)'); + expect(backgroundSource).toContain('expectedConnectionGeneration !== connectionGeneration'); + expect(backgroundSource).toContain('data.roomId !== pendingRoomDataRoomId'); + }); + it('exposes bounded queue diagnostics without media-title content', () => { expect(backgroundSource).toContain('queuedLogicalEvents: eventQueue.length'); expect(backgroundSource).toContain('queuedMediaIntents: queuedMediaIntentCount'); diff --git a/extension/offline-media-intent.js b/extension/offline-media-intent.js index 40ed8be..0828a36 100644 --- a/extension/offline-media-intent.js +++ b/extension/offline-media-intent.js @@ -4,6 +4,7 @@ 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 KNOWN_EVENTS = new Set(Object.values(EVENTS)); 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([ @@ -49,6 +50,22 @@ function trimQueue(queue, maxEntries) { const trimmed = queue.slice(); let dropped = 0; while (trimmed.length > maxEntries) { + if (trimmed[0]?.event === EVENTS.FORCE_SYNC_PREPARE) { + const executeIndex = trimmed.findIndex(entry => entry?.event === EVENTS.FORCE_SYNC_EXECUTE); + if (executeIndex >= 0) { + trimmed.splice(0, executeIndex + 1); + dropped += executeIndex + 1; + continue; + } + // Preserve an incomplete oldest Force Sync transaction. Evict the + // next-oldest work until EXECUTE arrives, at which point the whole + // transaction can be evicted atomically if pressure continues. + if (trimmed.length > 1) { + trimmed.splice(1, 1); + dropped++; + continue; + } + } trimmed.shift(); dropped++; } @@ -127,8 +144,16 @@ export function enqueueQueuedEvent(queue, event, data, { if (STALE_OFFLINE_EVENTS.has(event)) { return { queue: next, collapsed: 0, dropped: 0, droppedStale: 1 }; } + if (!KNOWN_EVENTS.has(event)) { + return { queue: next, collapsed: 0, dropped: 0, droppedStale: 1 }; + } if (!isMediaQueueEvent(event)) { - next.push({ event, data }); + next.push({ + kind: 'event', + roomId: typeof roomId === 'string' && roomId ? roomId : null, + event, + data + }); } else if (typeof roomId === 'string' && roomId) { const last = next.at(-1); const hasMergeTarget = isQueuedMediaIntent(last) && last.roomId === roomId; @@ -177,23 +202,71 @@ function normalizeIntentEntry(entry, roomId) { }; } +function repairIntentSequences(entry, minimumSequence) { + const repaired = { + ...entry, + intent: { ...entry.intent } + }; + const hasState = repaired.intent.playbackState !== null; + const hasPosition = repaired.intent.currentTime !== null; + const previousSeq = validSequence(repaired.intent.previousSeq); + const latestSeq = validSequence(repaired.intent.latestSeq); + if (hasState && hasPosition) { + if (latestSeq === null) { + repaired.intent.previousSeq = null; + repaired.intent.latestSeq = minimumSequence + 1; + } else if (latestSeq <= minimumSequence) { + if (previousSeq === null) { + repaired.intent.previousSeq = null; + repaired.intent.latestSeq = minimumSequence + 1; + } else { + repaired.intent.previousSeq = minimumSequence + 1; + repaired.intent.latestSeq = minimumSequence + 2; + } + } else if (previousSeq !== null + && (previousSeq >= latestSeq || previousSeq <= minimumSequence)) { + repaired.intent.previousSeq = minimumSequence + 1; + repaired.intent.latestSeq = Math.max(latestSeq, minimumSequence + 2); + } + } else if (latestSeq === null || latestSeq <= minimumSequence) { + repaired.intent.previousSeq = null; + repaired.intent.latestSeq = minimumSequence + 1; + } + return repaired; +} + export function normalizePersistedEventQueue(value, roomId, maxEntries = MAX_LOGICAL_QUEUE_SIZE) { if (!Array.isArray(value) || typeof roomId !== 'string' || !roomId) return []; let normalized = []; + let maximumSequence = 0; for (const entry of value) { if (isQueuedMediaIntent(entry)) { - const intentEntry = normalizeIntentEntry(entry, roomId); - if (intentEntry) normalized.push(intentEntry); + let intentEntry = normalizeIntentEntry(entry, roomId); + if (intentEntry) { + intentEntry = repairIntentSequences(intentEntry, maximumSequence); + normalized.push(intentEntry); + maximumSequence = Math.max(maximumSequence, maxQueuedSequence([intentEntry])); + } continue; } if (!entry || typeof entry !== 'object' || typeof entry.event !== 'string') continue; + if (!KNOWN_EVENTS.has(entry.event)) continue; + if (typeof entry.roomId === 'string' && entry.roomId && entry.roomId !== roomId) continue; if (STALE_OFFLINE_EVENTS.has(entry.event)) continue; + const data = entry.data && typeof entry.data === 'object' ? { ...entry.data } : entry.data; + const queuedSequence = validSequence(data?.seq); + if (data && typeof data === 'object' + && ((isMediaQueueEvent(entry.event) && queuedSequence === null) + || (queuedSequence !== null && queuedSequence <= maximumSequence))) { + data.seq = maximumSequence + 1; + } if (isMediaQueueEvent(entry.event)) { - normalized = enqueueQueuedEvent(normalized, entry.event, entry.data, { roomId, maxEntries }).queue; + normalized = enqueueQueuedEvent(normalized, entry.event, data, { roomId, maxEntries }).queue; } else { - normalized.push({ event: entry.event, data: entry.data }); + normalized.push({ kind: 'event', roomId, event: entry.event, data }); } normalized = trimQueue(normalized, maxEntries).queue; + maximumSequence = Math.max(maximumSequence, maxQueuedSequence(normalized)); } return normalized; } @@ -229,10 +302,10 @@ export function reserveLatestMediaIntentSequence(queue, roomId, nextSequence) { return { queue: next, reserved: true }; } -function frameData(intent, seq) { +function frameData(intent, seq, includeActionTimestamp = true) { const data = {}; if (seq !== null) data.seq = seq; - if (intent.actionTimestamp !== null) data.actionTimestamp = intent.actionTimestamp; + if (includeActionTimestamp && intent.actionTimestamp !== null) data.actionTimestamp = intent.actionTimestamp; if (intent.mediaTitle !== null) data.mediaTitle = intent.mediaTitle; return data; } @@ -259,11 +332,11 @@ export function materializeMediaIntent(entry) { } 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 + // 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. + // sequence that could overtake a later transactional barrier. In a current + // two-frame intent, only the logical final frame carries actionTimestamp so + // its helper frame cannot falsely acknowledge the final user action. if (previousSeq === null || latestSeq === null || previousSeq >= latestSeq) { if (intent.latestEvent === EVENTS.SEEK) { return [{ @@ -279,11 +352,26 @@ export function materializeMediaIntent(entry) { const seekFrame = { event: EVENTS.SEEK, - data: { ...frameData(intent, intent.latestEvent === EVENTS.SEEK ? latestSeq : previousSeq), currentTime, targetTime: currentTime } + data: { + ...frameData( + intent, + intent.latestEvent === EVENTS.SEEK ? latestSeq : previousSeq, + intent.latestEvent === EVENTS.SEEK + ), + currentTime, + targetTime: currentTime + } }; const stateFrame = { event: stateEvent, - data: { ...frameData(intent, intent.latestEvent === EVENTS.SEEK ? previousSeq : latestSeq), currentTime } + data: { + ...frameData( + intent, + intent.latestEvent === EVENTS.SEEK ? previousSeq : latestSeq, + intent.latestEvent !== EVENTS.SEEK + ), + currentTime + } }; return intent.latestEvent === EVENTS.SEEK ? [stateFrame, seekFrame] @@ -320,13 +408,18 @@ export function reconcileQueuedRoomIntent(queue, { roomId, canControl = true, activeLobby = false, - desynced = false + desynced = false, + authoritativeLobby = false } = {}) { const source = Array.isArray(queue) ? queue : []; const blockedEvents = !canControl ? HOST_GATED_EVENTS - : (activeLobby || desynced ? FORCE_SYNC_EVENTS : null); + : new Set([ + ...(activeLobby || desynced ? FORCE_SYNC_EVENTS : []), + ...(authoritativeLobby ? [EVENTS.EPISODE_LOBBY, EVENTS.EPISODE_READY, EVENTS.EPISODE_LOBBY_CANCEL] : []) + ]); const reconciled = source.filter(entry => { + if (entry?.roomId && entry.roomId !== roomId) return false; if (isQueuedMediaIntent(entry) && entry.roomId === roomId) { return canControl && !activeLobby && !desynced; } @@ -366,7 +459,7 @@ export async function drainQueuedBatch(queue, { while (remaining.length > 0) { const entry = remaining[0]; - if (isQueuedMediaIntent(entry) && entry.roomId !== roomId) { + if (entry?.roomId && entry.roomId !== roomId) { remaining.shift(); droppedStaleIntents++; continue; diff --git a/extension/offline-media-intent.test.mjs b/extension/offline-media-intent.test.mjs index d7f3ead..df59459 100644 --- a/extension/offline-media-intent.test.mjs +++ b/extension/offline-media-intent.test.mjs @@ -34,7 +34,7 @@ describe('offline media intent coalescing', () => { 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.SEEK, data: { seq: 5, currentTime: 10, targetTime: 10 } }, { event: EVENTS.PLAY, data: { seq: 6, actionTimestamp: 100, currentTime: 10 } } ]); }); @@ -159,6 +159,17 @@ describe('offline media intent coalescing', () => { expect(maxQueuedSequence(restored)).toBe(4); }); + it('drops room-scoped barriers from another room and unknown persisted events', () => { + const restored = normalizePersistedEventQueue([ + { kind: 'event', roomId: 'room-b', event: EVENTS.FORCE_SYNC_EXECUTE, data: { seq: 1 } }, + { kind: 'event', roomId, event: 'unexpected_event', data: { secret: 'nope' } }, + { kind: 'event', roomId, event: EVENTS.EPISODE_READY, data: { seq: 2 } } + ], roomId); + expect(restored).toEqual([{ + kind: 'event', roomId, event: EVENTS.EPISODE_READY, data: { seq: 2 } + }]); + }); + 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); @@ -180,10 +191,28 @@ describe('offline media intent coalescing', () => { 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.queue).toEqual([{ + kind: 'event', + roomId, + event: EVENTS.EPISODE_READY, + data: { title: 'S01E02' } + }]); expect(result.hasPendingLocalIntent).toBe(false); }); + it('drops stale queued Episode Lobby coordination when ROOM_DATA has an authoritative lobby', () => { + let queue = enqueueQueuedEvent([], EVENTS.EPISODE_LOBBY, { expectedTitle: 'S02E01' }, { roomId }).queue; + queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { title: 'S02E01' }, { roomId }).queue; + queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_LOBBY_CANCEL, {}, { roomId }).queue; + const result = reconcileQueuedRoomIntent(queue, { + roomId, + activeLobby: true, + authoritativeLobby: true + }); + expect(result.queue).toEqual([]); + expect(result.discarded).toBe(3); + }); + 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 }); @@ -229,7 +258,7 @@ describe('offline media intent drain', () => { }); it('retains the whole logical intent after a partial send failure', async () => { - const queue = reserve(media(EVENTS.PLAY, { currentTime: 90, seq: 5 }), 6); + const queue = reserve(media(EVENTS.PLAY, { currentTime: 90, seq: 5, actionTimestamp: 500 }), 6); let calls = 0; const result = await drainQueuedBatch(queue, { roomId, @@ -240,6 +269,8 @@ describe('offline media intent drain', () => { expect(result.sentWireEvents).toBe(1); expect(result.queue).toEqual(queue); expect(materializeMediaIntent(result.queue[0]).map(frame => frame.data.seq)).toEqual([5, 6]); + expect(materializeMediaIntent(result.queue[0]).map(frame => frame.data.actionTimestamp)) + .toEqual([undefined, 500]); }); it('drops stale-room intent during drain and preserves unrelated events', async () => { @@ -251,8 +282,58 @@ describe('offline media intent drain', () => { 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 } }]); + expect(result.droppedStaleIntents).toBe(2); + expect(sent).toEqual([]); + }); + + it('repairs regressing sequences across malformed persisted intent entries', () => { + const restored = normalizePersistedEventQueue([ + { + kind: 'media-intent', + roomId, + intent: { + playbackState: 'playing', currentTime: 10, latestEvent: EVENTS.PLAY, + previousSeq: 99, latestSeq: 100, actionTimestamp: 1, mediaTitle: null, sourceEventCount: 1 + } + }, + { + kind: 'media-intent', + roomId, + intent: { + playbackState: 'paused', currentTime: 20, latestEvent: EVENTS.PAUSE, + previousSeq: 49, latestSeq: 50, actionTimestamp: 2, mediaTitle: null, sourceEventCount: 1 + } + } + ], roomId); + expect(materializeMediaIntent(restored[0]).map(frame => frame.data.seq)).toEqual([99, 100]); + expect(materializeMediaIntent(restored[1]).map(frame => frame.data.seq)).toEqual([101, 102]); + expect(maxQueuedSequence(restored)).toBe(102); + }); + + it('preserves a valid legacy single-frame intent during sequence repair', () => { + const restored = normalizePersistedEventQueue([{ + kind: 'media-intent', + roomId, + intent: { + playbackState: 'paused', currentTime: 20, latestEvent: EVENTS.PAUSE, + previousSeq: null, latestSeq: 50, actionTimestamp: 2, mediaTitle: null, sourceEventCount: 1 + } + }], roomId); + expect(materializeMediaIntent(restored[0])).toEqual([{ + event: EVENTS.PAUSE, + data: { seq: 50, actionTimestamp: 2, currentTime: 20 } + }]); + }); + + it('evicts a complete Force Sync transaction instead of orphaning EXECUTE at the cap', () => { + let queue = enqueueQueuedEvent([], EVENTS.FORCE_SYNC_PREPARE, { targetTime: 100, seq: 1 }, { roomId }).queue; + queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_EXECUTE, { seq: 2 }, { roomId }).queue; + for (let index = 0; index < 49; index++) { + queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_ACK, { seq: index + 3 }, { roomId }).queue; + } + expect(queue).toHaveLength(49); + expect(queue.some(entry => entry.event === EVENTS.FORCE_SYNC_PREPARE)).toBe(false); + expect(queue.some(entry => entry.event === EVENTS.FORCE_SYNC_EXECUTE)).toBe(false); }); it('reports logical and actual-wire queue sizes separately', () => { diff --git a/scripts/test-server-ws.mjs b/scripts/test-server-ws.mjs index 0efebb7..4d92c19 100644 --- a/scripts/test-server-ws.mjs +++ b/scripts/test-server-ws.mjs @@ -371,6 +371,35 @@ try { assert.equal(forceSyncState.playbackState, 'playing'); assert.equal(forceSyncState.currentTime, 700); + // The legacy wire exposes one room-wide prepared target. A later PREPARE + // replaces what every peer has just sought to; any currently authorized + // EXECUTE must commit that visible target instead of clearing it unmatched. + s(msa, 'force_sync_prepare', { targetTime: 800 }); + await w(msb, 'force_sync_prepare'); + s(msb, 'force_sync_prepare', { targetTime: 900 }); + await w(msa, 'force_sync_prepare'); + const beforeCompetingExecute = { ...mod.rooms.get(msrid).mediaState }; + s(msa, 'force_sync_execute', {}); + await w(msb, 'force_sync_execute'); + const competingForceState = mod.rooms.get(msrid).mediaState; + assert.equal(competingForceState.revision, beforeCompetingExecute.revision + 1); + assert.equal(competingForceState.currentTime, 900, + 'authorized EXECUTE commits the latest target visible to legacy peers'); + assert.equal(competingForceState.updatedBy, 'msa'); + + msa._m.length = msb._m.length = 0; + s(msa, 'force_sync_prepare', { targetTime: 1_000 }); + await w(msb, 'force_sync_prepare'); + s(msb, 'pause', { currentTime: 1_100 }); + await w(msa, 'pause'); + const supersedingMediaState = { ...mod.rooms.get(msrid).mediaState }; + msa._m.length = msb._m.length = 0; + s(msa, 'force_sync_execute', {}); + let orphanExecuteDropped = false; + try { await w(msb, 'force_sync_execute', 500); } catch { orphanExecuteDropped = true; } + assert.ok(orphanExecuteDropped, 'an action that supersedes PREPARE makes delayed EXECUTE invalid'); + assert.deepEqual(mod.rooms.get(msrid).mediaState, supersedingMediaState); + // Active Episode Lobby is additive ROOM_DATA state and does not rewrite mediaState. s(msa, 'episode_lobby', { expectedTitle: 'S02E03' }); await delay(40); @@ -675,6 +704,63 @@ try { close(); resetConnectionRate(); + // A valid PREPARE remains executable if the room changes from everyone to + // host-only before EXECUTE. An invalid PREPARE grants no such exemption. + const transitionRid = 'force-transition-'+Date.now(); + const transitionHost = await c(), transitionGuest = await c(); + await j(transitionHost, transitionRid, 'transition-host'); + await j(transitionGuest, transitionRid, 'transition-guest'); + transitionHost._m.length = transitionGuest._m.length = 0; + s(transitionGuest, 'force_sync_prepare', { targetTime: 321 }); + await w(transitionHost, 'force_sync_prepare'); + s(transitionHost, 'set_control_mode', { controlMode: 'host-only' }); + await w(transitionGuest, 'control_mode'); + transitionHost._m.length = transitionGuest._m.length = 0; + s(transitionGuest, 'force_sync_execute', {}); + await w(transitionHost, 'force_sync_execute'); + assert.equal(mod.rooms.get(transitionRid).mediaState.currentTime, 321); + + await delay(550); + s(transitionHost, 'set_peer_role', { peerId: 'transition-guest', controller: true }); + await w(transitionGuest, 'control_mode'); + transitionHost._m.length = transitionGuest._m.length = 0; + s(transitionGuest, 'force_sync_prepare', { targetTime: 'invalid' }); + let invalidPrepareRelayed = false; + try { await w(transitionHost, 'force_sync_prepare', 500); } catch { invalidPrepareRelayed = true; } + assert.ok(invalidPrepareRelayed, 'invalid PREPARE is not relayed'); + await delay(550); + s(transitionHost, 'set_peer_role', { peerId: 'transition-guest', controller: false }); + await w(transitionGuest, 'control_mode'); + transitionHost._m.length = transitionGuest._m.length = 0; + s(transitionGuest, 'force_sync_execute', {}); + let invalidExecuteGated = false; + try { await w(transitionHost, 'force_sync_execute', 500); } catch { invalidExecuteGated = true; } + assert.ok(invalidExecuteGated, 'invalid PREPARE grants no post-demotion EXECUTE exemption'); + close(); + resetConnectionRate(); + + // A transient disconnect clears the Host Control exemption but not the + // validated room target. If the peer rejoins with current authority, its + // recovered EXECUTE must keep live playback and canonical state aligned. + const forceReconnectRid = 'force-reconnect-'+Date.now(); + const forceReconnectHost = await c(), forceReconnectPeer = await c(); + await j(forceReconnectHost, forceReconnectRid, 'force-host'); + await j(forceReconnectPeer, forceReconnectRid, 'force-peer'); + forceReconnectHost._m.length = forceReconnectPeer._m.length = 0; + s(forceReconnectPeer, 'force_sync_prepare', { targetTime: 444 }); + await w(forceReconnectHost, 'force_sync_prepare'); + forceReconnectPeer.close(); + await delay(100); + const forceReconnectReplacement = await c(); + await j(forceReconnectReplacement, forceReconnectRid, 'force-peer'); + forceReconnectHost._m.length = forceReconnectReplacement._m.length = 0; + s(forceReconnectReplacement, 'force_sync_execute', {}); + await w(forceReconnectHost, 'force_sync_execute'); + assert.equal(mod.rooms.get(forceReconnectRid).mediaState.currentTime, 444); + assert.equal(mod.rooms.get(forceReconnectRid).mediaState.playbackState, 'playing'); + close(); + resetConnectionRate(); + // --- A guest's stray EXECUTE (no matching PREPARE they initiated) is still gated --- const grid = 'h1b-'+Date.now(); const go = await c(), gg = await c(); @@ -755,12 +841,12 @@ try { s(mxo,'play',{currentTime:1}); await w(mxn,'play'); s(mxo,'seek',{currentTime:99}); await w(mxn,'seek'); s(mxo,'force_sync_prepare',{targetTime:5}); await w(mxn,'force_sync_prepare'); - s(mxo,'episode_lobby',{expectedTitle:'S1E1'}); await w(mxn,'episode_lobby'); // New → old mxo._m.length = mxn._m.length = 0; + s(mxn,'force_sync_execute',{}); await w(mxo,'force_sync_execute'); + s(mxo,'episode_lobby',{expectedTitle:'S1E1'}); await w(mxn,'episode_lobby'); s(mxn,'pause',{currentTime:2}); await w(mxo,'pause'); s(mxn,'seek',{currentTime:50}); await w(mxo,'seek'); - s(mxn,'force_sync_execute',{}); await w(mxo,'force_sync_execute'); s(mxn,'episode_lobby_cancel',{}); await w(mxo,'episode_lobby_cancel'); close(); resetConnectionRate(); diff --git a/server/index.js b/server/index.js index 48500de..b1bd16c 100644 --- a/server/index.js +++ b/server/index.js @@ -286,10 +286,10 @@ function removePeerFromRoom(socketId, roomId, reason) { if (peerGone && room.controllers && room.peers.size > 0) { const wasController = room.controllers.has(peerId); room.controllers.delete(peerId); - // H-1: a leaving initiator strands the room's force-sync — release the - // slot so a future controller's PREPARE can take over cleanly. + // Release the post-demotion exemption, but retain the validated target: + // an authorized initiator may reconnect and finish the already-visible + // choreography. A newer PREPARE still replaces it normally. if (room.forceSyncInitiator === peerId) room.forceSyncInitiator = null; - if (room.forceSyncTarget?.initiatorPeerId === peerId) room.forceSyncTarget = null; if (room.hostPeerId === peerId) { // Owner left → reassign owner + fall back to 'everyone' so the room is // never stuck locked, and reset the controller set to just the new owner. @@ -596,15 +596,10 @@ io.on('connection', (socket) => { // a controller (the owner + any promoted co-hosts). Robust chokepoint: // independent of client behavior, kills spam. Heartbeats/ACKs pass. // - // H-1 exception: a demoted co-host's FORCE_SYNC_EXECUTE still has to - // land — otherwise their already-relayed PREPARE would leave the whole - // room stuck paused. Track the in-flight initiator on PREPARE and let - // their matching EXECUTE through regardless of current controllers set. - if (eventName === EVENTS.FORCE_SYNC_PREPARE && - room.controlMode === CONTROL_MODES.HOST_ONLY && - room.controllers && room.controllers.has(mapping.peerId)) { - room.forceSyncInitiator = mapping.peerId; - } + // H-1 exception: the latest valid PREPARE initiator's + // FORCE_SYNC_EXECUTE still has to land after demotion — + // otherwise the already-relayed room-wide choreography + // would leave peers paused. const isOwnForceSyncExecute = eventName === EVENTS.FORCE_SYNC_EXECUTE && room.forceSyncInitiator && mapping.peerId === room.forceSyncInitiator; if (!isOwnForceSyncExecute && @@ -614,11 +609,6 @@ io.on('connection', (socket) => { log('ROOM', `Dropped ${eventName} from guest ${mapping.peerId} in host-only room ${mapping.roomId.substring(0, 3)}***`); return; } - // Clear initiator tracking once the EXECUTE has been relayed. - if (eventName === EVENTS.FORCE_SYNC_EXECUTE && room.forceSyncInitiator) { - room.forceSyncInitiator = null; - } - // --- S-2 & S-3: Sanitize ALL relay fields (strings, numbers, booleans) --- const clamp = (val, max) => typeof val === 'string' ? val.substring(0, max) : undefined; const clampNum = (val, min, max) => typeof val === 'number' && Number.isFinite(val) ? Math.max(min, Math.min(max, val)) : undefined; @@ -661,6 +651,23 @@ io.on('connection', (socket) => { // Strip undefined keys for clean wire format Object.keys(relayPayload).forEach(k => relayPayload[k] === undefined && delete relayPayload[k]); + if (eventName === EVENTS.FORCE_SYNC_EXECUTE && !room.forceSyncTarget) { + log('ROOM', `Dropped force_sync_execute without a prepared target from ${mapping.peerId}`); + room.forceSyncInitiator = null; + return; + } + if (eventName === EVENTS.PLAY + || eventName === EVENTS.PAUSE + || eventName === EVENTS.SEEK + || eventName === EVENTS.EPISODE_LOBBY + || eventName === EVENTS.EPISODE_LOBBY_CANCEL) { + // A later room-driving action supersedes unfinished Force + // Sync choreography. Do not let a delayed EXECUTE commit + // an obsolete target after peers have moved elsewhere. + room.forceSyncInitiator = null; + room.forceSyncTarget = null; + } + // Canonical Media State v1: mutate only after rate limiting, // room mapping, Host Control authorization and sanitization. // Heartbeats remain observational and never enter this path. @@ -670,12 +677,25 @@ io.on('connection', (socket) => { senderPlaybackState: existing.playbackState }); if (eventName === EVENTS.FORCE_SYNC_PREPARE) { - room.forceSyncTarget = Number.isFinite(relayPayload.targetTime) - ? { initiatorPeerId: mapping.peerId, targetTime: relayPayload.targetTime } - : null; + // A malformed PREPARE must neither pause peers nor grant + // the initiator a later Host Control EXECUTE exemption. + if (!Number.isFinite(relayPayload.targetTime)) { + log('ROOM', `Dropped invalid force_sync_prepare from ${mapping.peerId}`); + return; + } + // One room-wide choreography is visible on the legacy + // wire. A newer PREPARE replaces the target every peer + // most recently received. Track its initiator in every + // control mode so an everyone -> host-only transition + // cannot strand that already-authorized transaction. + room.forceSyncInitiator = mapping.peerId; + room.forceSyncTarget = { + initiatorPeerId: mapping.peerId, + targetTime: relayPayload.targetTime + }; } else if (eventName === EVENTS.FORCE_SYNC_EXECUTE) { const forceSyncTarget = room.forceSyncTarget; - if (forceSyncTarget?.initiatorPeerId === mapping.peerId) { + if (forceSyncTarget) { commitForceSyncMediaState( room, forceSyncTarget.targetTime, @@ -683,6 +703,7 @@ io.on('connection', (socket) => { mediaStateNow ); } + room.forceSyncInitiator = null; room.forceSyncTarget = null; } diff --git a/tests/e2e/extension.spec.mjs b/tests/e2e/extension.spec.mjs index 7168c43..835aaa2 100644 --- a/tests/e2e/extension.spec.mjs +++ b/tests/e2e/extension.spec.mjs @@ -1182,10 +1182,11 @@ test('coalesces persisted offline media intent before canonical reconnect recove expect(restoredQueue.queuedLogicalEvents).toBeGreaterThanOrEqual(1); expect(restoredQueue.queuedWireEvents).toBeGreaterThanOrEqual(2); - await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async serverUrl => { + const retryResult = await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async serverUrl => { await chrome.storage.local.set({ serverUrl }); - chrome.alarms.create('keepAlive', { when: Date.now() + 50 }); + return chrome.runtime.sendMessage({ type: 'RETRY_CONNECT' }); }, `ws://127.0.0.1:${port}`)); + expect(retryResult).toMatchObject({ status: 'ok' }); await page.locator('#player').evaluate(video => { window.__koalaReconnectSeeks = [];