From def5e1a380802b202461877c011db234364458c5 Mon Sep 17 00:00:00 2001 From: Timo <6156589+Shik3i@users.noreply.github.com> Date: Tue, 25 Aug 2026 21:21:18 +0200 Subject: [PATCH] fix(sync): harden canonical recovery --- docs/ARCHITECTURE.md | 18 ++-- docs/PROTOCOL.md | 28 ++++-- extension/background.js | 84 ++++++++++++++++- .../canonical-media-state-background.test.mjs | 29 ++++-- extension/content.js | 91 ++++++++++++++++--- scripts/test-server-ws.mjs | 45 ++++++++- server/index.js | 41 ++++++++- shared/README.md | 1 + shared/constants.js | 3 + tests/e2e/extension.spec.mjs | 69 ++++++++++++++ 10 files changed, 363 insertions(+), 46 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 031a327..2366d1f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -44,17 +44,23 @@ the optional `media-state-v1` capability. A joining/reconnecting client validate the room/revision, respects Host Control solo mode and Episode Lobby, then sends an 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: 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. +the 2-second drift tolerance, and programmatic-event suppression. Recovery only +completes after playback state and position verification. Transient failures +retry after 250, 750, 1500, and 3000 ms, while target, heartbeat, and content-boot +signals can retrigger a pending attempt within that bound. 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. A valid `PREPARE` is temporary room-wide choreography; the next authorized `EXECUTE` commits the latest target -visible to peers to canonical state before the shared Force Sync timeout. The +visible to peers to canonical state before the relay target TTL. That TTL is +longer than the client ACK timeout so its scheduled fallback can still land. The offline queue replays an adjacent `PREPARE`/`EXECUTE` pair in one paced batch and retains both if delivery fails. Per-sender -`seq`, peer heartbeats, and the reconnect queue remain separate mechanisms. +`seq`, peer heartbeats, and the reconnect queue remain separate mechanisms. The +relay rejects duplicate/regressing current-client media sequences before they +can diverge canonical truth from live receivers. ## 3.2 Offline Media Intent diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 4cbbd0e..ee49dae 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -121,21 +121,28 @@ Only accepted, sanitized room controls update canonical state: established playback state. - 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 before `FORCE_SYNC_TIMEOUT` expires. Expired targets are - cleared and cannot alter canonical state. The latest valid prepare is also - the only post-demotion execute exemption in Host Control mode. + target as playing before `FORCE_SYNC_TARGET_TTL` expires. This relay TTL is + intentionally longer than the client's `FORCE_SYNC_TIMEOUT` ACK wait so the + normal timeout fallback remains deliverable. Expired targets are cleared and + cannot alter canonical state. 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` -orders server-accepted room transitions. Neither replaces the other. +For current clients, the relay drops invalid, duplicate, or regressing `seq` +values on room-moving media commands before relay/canonical mutation. Legacy +clients without `seq` retain their existing behavior. Canonical `revision` +orders server-accepted room transitions; it does not replace per-sender order. -On join/reconnect, a capable extension applies a valid snapshot once through an -extension-internal recovery message. Existing seek/page-API and native-event +On join/reconnect, a capable extension attempts to apply a valid snapshot +through an extension-internal recovery message. Recovery is only marked handled +after playback state and position verification. Transient failures use bounded +retries after 250, 750, 1500, and 3000 ms and can also be retriggered by target, +heartbeat, or content-boot signals. 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 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. +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 the capability; old clients ignore the extra `room_data` field from a new relay. @@ -341,7 +348,8 @@ the latest validated room target retained from `force_sync_prepare`. In `host-only` mode, only controllers may send it. 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. +no exemption. The retained target expires after `FORCE_SYNC_TARGET_TTL`, which +includes a relay grace period beyond the client's ACK timeout. ## Episode Lobby diff --git a/extension/background.js b/extension/background.js index 44bde99..f4bcf1e 100644 --- a/extension/background.js +++ b/extension/background.js @@ -258,6 +258,18 @@ 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_RECOVERY_RETRY_DELAYS = Object.freeze([250, 750, 1500, 3000]); +const CANONICAL_RECOVERY_RETRYABLE = new Set([ + 'no_video', + 'apply_failed', + 'no_response', + 'pending_unreachable', + 'stale_target', + 'superseded' +]); +let canonicalRecoveryRetryTimer = null; +let canonicalRecoveryRetryAttempt = 0; +let canonicalRecoveryApplyInProgress = null; // --- Host Control Mode --- let controlMode = CONTROL_MODES.EVERYONE; // 'everyone' | 'host-only' @@ -285,7 +297,52 @@ function persistCanonicalMediaRecovery() { }).catch(() => {}); } +function resetCanonicalMediaRecoveryRetries() { + if (canonicalRecoveryRetryTimer) { + clearTimeout(canonicalRecoveryRetryTimer); + canonicalRecoveryRetryTimer = null; + } + canonicalRecoveryRetryAttempt = 0; +} + +function scheduleCanonicalMediaRecoveryRetry(reason) { + const roomId = currentRoom?.roomId; + const pending = canonicalMediaStateTracker.getPending(roomId); + if (!pending + || canonicalRecoveryRetryTimer + || canonicalRecoveryApplyInProgress + || canonicalRecoveryRetryAttempt >= CANONICAL_RECOVERY_RETRY_DELAYS.length) { + return false; + } + const expectedRevision = pending.mediaState.revision; + const delay = CANONICAL_RECOVERY_RETRY_DELAYS[canonicalRecoveryRetryAttempt++]; + canonicalRecoveryRetryTimer = setTimeout(() => { + canonicalRecoveryRetryTimer = null; + const latest = canonicalMediaStateTracker.getPending(roomId); + if (currentRoom?.roomId !== roomId || latest?.mediaState.revision !== expectedRevision) return; + addLog(`Retrying canonical media state r${expectedRevision} after ${reason}`, 'info'); + tryApplyPendingCanonicalMediaState().catch(error => { + addLog(`Canonical media state retry failed: ${error.message}`, 'warn'); + }); + }, delay); + return true; +} + +function requestCanonicalMediaRecoveryAttempt() { + if (canonicalRecoveryRetryTimer + || canonicalRecoveryApplyInProgress + || canonicalRecoveryRetryAttempt >= CANONICAL_RECOVERY_RETRY_DELAYS.length + || !canonicalMediaStateTracker.getPending(currentRoom?.roomId)) { + return false; + } + tryApplyPendingCanonicalMediaState().catch(error => { + addLog(`Canonical media state retry failed: ${error.message}`, 'warn'); + }); + return true; +} + function clearCanonicalMediaRecovery() { + resetCanonicalMediaRecoveryRetries(); canonicalMediaStateTracker.clear(); persistCanonicalMediaRecovery(); } @@ -728,6 +785,7 @@ function resolveServerUrl(settings) { function forceDisconnect({ preserveEventQueue = false } = {}) { connectionGeneration++; + resetCanonicalMediaRecoveryRetries(); if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; @@ -945,7 +1003,7 @@ function adoptReportingFrame(sender) { currentTargetHasVideo = true; stopMediaDiscoveryPoll(); chrome.storage.session.set({ currentTargetHasVideo }).catch(() => {}); - tryApplyPendingCanonicalMediaState().catch(() => {}); + requestCanonicalMediaRecoveryAttempt(); return true; } @@ -960,7 +1018,7 @@ function adoptReportingFrame(sender) { currentTargetDocumentId, currentTargetHasVideo }).catch(() => {}); - tryApplyPendingCanonicalMediaState().catch(() => {}); + requestCanonicalMediaRecoveryAttempt(); return true; } @@ -1817,11 +1875,12 @@ function stopPing() { function markCanonicalMediaStateHandled(roomId, revision) { if (!canonicalMediaStateTracker.markHandled(roomId, revision)) return false; + resetCanonicalMediaRecoveryRetries(); persistCanonicalMediaRecovery(); return true; } -async function tryApplyPendingCanonicalMediaState() { +async function performPendingCanonicalMediaStateApply() { const roomId = currentRoom?.roomId; const pending = canonicalMediaStateTracker.getPendingProjected(roomId); if (!pending || !roomId) return { status: 'none' }; @@ -1876,6 +1935,22 @@ async function tryApplyPendingCanonicalMediaState() { } } +async function tryApplyPendingCanonicalMediaState() { + if (canonicalRecoveryApplyInProgress) return canonicalRecoveryApplyInProgress; + const applyTask = performPendingCanonicalMediaStateApply(); + canonicalRecoveryApplyInProgress = applyTask; + let result; + try { + result = await applyTask; + } finally { + if (canonicalRecoveryApplyInProgress === applyTask) canonicalRecoveryApplyInProgress = null; + } + if (CANONICAL_RECOVERY_RETRYABLE.has(result?.status)) { + scheduleCanonicalMediaRecoveryRetry(result.status); + } + return result; +} + async function handleCanonicalRoomData(data, hasPendingLocalIntent) { canonicalMediaStateTracker.adoptRoom(data?.roomId || null); const canonicalSnapshot = canonicalMediaStateFromRoomData(data); @@ -1898,6 +1973,7 @@ async function handleCanonicalRoomData(data, hasPendingLocalIntent) { } if (received.status !== 'pending') return; + resetCanonicalMediaRecoveryRetries(); persistCanonicalMediaRecovery(); addLog(`Canonical media state received: r${mediaState.revision} ${mediaState.playbackState} @ ${mediaState.currentTime.toFixed(2)}s`, 'info'); @@ -4896,6 +4972,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { updateBadgeStatus(); } + requestCanonicalMediaRecoveryAttempt(); markRoomUseful(); getSettings().then(settings => { const sharedTitles = getSharedTitleFields(settings, message.payload?.mediaTitle); @@ -5222,6 +5299,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { return; } } + requestCanonicalMediaRecoveryAttempt(); // Content script re-injected, check if there's an active lobby if (episodeLobby) { sendResponse({ lobbyActive: true, expectedTitle: episodeLobby.expectedTitle }); diff --git a/extension/canonical-media-state-background.test.mjs b/extension/canonical-media-state-background.test.mjs index 699751d..2385d80 100644 --- a/extension/canonical-media-state-background.test.mjs +++ b/extension/canonical-media-state-background.test.mjs @@ -37,7 +37,7 @@ describe('canonical ROOM_DATA recovery contract', () => { }); it('uses a dedicated internal apply message without action/history/ACK machinery', () => { - const apply = functionBody(backgroundSource, 'tryApplyPendingCanonicalMediaState', 'handleCanonicalRoomData'); + const apply = functionBody(backgroundSource, 'performPendingCanonicalMediaStateApply', 'tryApplyPendingCanonicalMediaState'); expect(apply).toContain("type: 'APPLY_CANONICAL_MEDIA_STATE'"); expect(apply).not.toContain('routeToContent('); expect(apply).not.toContain('emit('); @@ -55,17 +55,23 @@ describe('canonical ROOM_DATA recovery contract', () => { it('keeps pending recovery room-scoped and retries on target lifecycle signals', () => { expect(backgroundSource).toContain("'canonicalMediaRecovery'"); expect(backgroundSource).toContain('canonicalMediaStateTracker.restore('); - expect(backgroundSource).toContain('tryApplyPendingCanonicalMediaState().catch(() => {})'); + expect(backgroundSource).toContain('CANONICAL_RECOVERY_RETRY_DELAYS'); + expect(backgroundSource).toContain('requestCanonicalMediaRecoveryAttempt()'); + expect(backgroundSource).toMatch(/message\.type === 'HEARTBEAT'[\s\S]*requestCanonicalMediaRecoveryAttempt\(\)/); + expect(backgroundSource).toMatch(/message\.type === 'CONTENT_BOOT'[\s\S]*requestCanonicalMediaRecoveryAttempt\(\)/); expect(backgroundSource).toMatch(/currentTargetHasVideo\) \{\s*await tryApplyPendingCanonicalMediaState\(\)/); expect(backgroundSource.match(/clearCanonicalMediaRecovery\(\)/g)?.length).toBeGreaterThanOrEqual(4); - const apply = functionBody(backgroundSource, 'tryApplyPendingCanonicalMediaState', 'handleCanonicalRoomData'); + const apply = functionBody(backgroundSource, 'performPendingCanonicalMediaStateApply', 'tryApplyPendingCanonicalMediaState'); expect(apply).toContain('getPendingProjected(roomId)'); expect(apply).toContain('targetActivationGeneration'); expect(apply).toContain("return { status: 'stale_target' }"); + const retry = functionBody(backgroundSource, 'scheduleCanonicalMediaRecoveryRetry', 'requestCanonicalMediaRecoveryAttempt'); + expect(retry).toContain('canonicalRecoveryRetryAttempt >= CANONICAL_RECOVERY_RETRY_DELAYS.length'); + expect(retry).toContain('latest?.mediaState.revision !== expectedRevision'); }); it('protects intentional desync, active Episode Lobby and queued reconnect intent', () => { - const apply = functionBody(backgroundSource, 'tryApplyPendingCanonicalMediaState', 'handleCanonicalRoomData'); + const apply = functionBody(backgroundSource, 'performPendingCanonicalMediaStateApply', 'tryApplyPendingCanonicalMediaState'); const roomData = functionBody(backgroundSource, 'handleCanonicalRoomData', 'handleServerEvent'); expect(apply).toContain('if (hcmDesynced)'); expect(apply).toContain('if (episodeLobby)'); @@ -75,13 +81,20 @@ describe('canonical ROOM_DATA recovery contract', () => { expect(backgroundSource).toContain('await flushEventQueue(replaySettings)'); }); - it('reuses existing seek abstractions, suppression and drift tolerance', () => { + it('awaits media actions and verifies playback plus drift before acknowledging recovery', () => { const apply = functionBody(contentSource, 'applyCanonicalMediaState', 'pollSeekReady'); expect(apply).toContain('Math.abs(drift) >= MIN_SEEK_DELTA'); expect(apply).toContain("_setSuppress('seek')"); - expect(apply).toContain('seekVideo(video, mediaState.currentTime)'); - expect(apply).toContain('tryMediaAction(EVENTS.PAUSE)'); - expect(apply).toContain('tryMediaAction(EVENTS.PLAY)'); + expect(apply).toContain('await tryMediaAction(EVENTS.SEEK'); + expect(apply).toContain('await tryMediaAction(EVENTS.PAUSE)'); + expect(apply).toContain('await tryMediaAction(EVENTS.PLAY)'); + expect(apply).toContain('await pollCanonicalMediaState(mediaState, startedAt)'); + expect(apply.indexOf("status: 'applied'")) + .toBeGreaterThan(apply.indexOf('await pollCanonicalMediaState(mediaState, startedAt)')); expect(apply).toContain('if (hcmDesynced)'); + const contentHandlerStart = contentSource.indexOf("message.type === 'APPLY_CANONICAL_MEDIA_STATE'"); + const serverCommandStart = contentSource.indexOf("message.type === 'SERVER_COMMAND'", contentHandlerStart); + expect(contentSource.slice(contentHandlerStart, serverCommandStart)) + .toContain('applyCanonicalMediaState(message.mediaState).then(sendResponse)'); }); }); diff --git a/extension/content.js b/extension/content.js index e07b636..198fa8a 100644 --- a/extension/content.js +++ b/extension/content.js @@ -1185,13 +1185,13 @@ // --- Helper: site-specific player actions, then native HTML5 fallback --- function tryMediaAction(action, data) { const video = findVideo(); - if (!video) return; + if (!video) return false; if (action === EVENTS.SEEK) { const target = data ? (data.targetTime !== undefined ? data.targetTime : data.currentTime) : undefined; if (!Number.isFinite(target)) { reportLog(`Media Action Error: Invalid seek payload - ${JSON.stringify(data)}`, 'error'); - return; + return false; } data = { ...data, targetTime: target }; } @@ -1199,28 +1199,71 @@ try { const actionFix = getActivePlayerActionFix(); if (tryPlayerActionFix(actionFix, action, video, data)) { - return; + return true; } // Fallback for native HTML5 if (action === EVENTS.PLAY) { _setSuppress('playing'); - video.play().catch((e) => { - reportLog(`Playback prevented: ${e.message}`, 'warn'); - _clearSuppress('playing'); - }); + const playResult = video.play(); + if (playResult && typeof playResult.then === 'function') { + return playResult.then(() => true).catch((e) => { + reportLog(`Playback prevented: ${e.message}`, 'warn'); + _clearSuppress('playing'); + return false; + }); + } + return true; } else if (action === EVENTS.PAUSE) { _setSuppress('paused'); video.pause(); + return true; } else if (action === EVENTS.SEEK) { seekVideo(video, data.targetTime, data.delta); + return true; } - } catch (e) { + return false; + } catch (e) { reportLog(`Media Action Error: ${e.message}`, 'error'); + return false; } } - function applyCanonicalMediaState(mediaState) { + function pollCanonicalMediaState(mediaState, startedAt, timeoutMs = 2500) { + return new Promise((resolve) => { + const interval = 100; + const finishAt = Date.now() + timeoutMs; + const timer = setInterval(() => { + if (destroyed) { + clearInterval(timer); + seekPollTimers.delete(timer); + resolve(null); + return; + } + const video = findVideo(); + const currentTime = video ? getSyncCurrentTime(video) : null; + const projectedTime = mediaState.currentTime + + (mediaState.playbackState === 'playing' + ? Math.max(0, Date.now() - startedAt) / 1000 + : 0); + const playbackMatches = video + && (mediaState.playbackState === 'playing' ? !video.paused : video.paused); + const drift = currentTime === null ? null : projectedTime - currentTime; + if (playbackMatches && drift !== null && Math.abs(drift) < MIN_SEEK_DELTA) { + clearInterval(timer); + seekPollTimers.delete(timer); + resolve({ currentTime, drift }); + } else if (Date.now() >= finishAt) { + clearInterval(timer); + seekPollTimers.delete(timer); + resolve(null); + } + }, interval); + seekPollTimers.add(timer); + }); + } + + async function applyCanonicalMediaState(mediaState) { if (!mediaState || typeof mediaState !== 'object' || !Number.isSafeInteger(mediaState.revision) || mediaState.revision < 1 || (mediaState.playbackState !== 'playing' && mediaState.playbackState !== 'paused') @@ -1238,23 +1281,40 @@ const currentTime = getSyncCurrentTime(video); const drift = currentTime === null ? null : mediaState.currentTime - currentTime; const shouldSeek = drift === null || Math.abs(drift) >= MIN_SEEK_DELTA; + const startedAt = Date.now(); try { // Paused recovery pauses before seeking; playing recovery seeks before // starting. Both paths reuse the same site/page-API abstractions and // native-event suppression as ordinary remote commands. if (mediaState.playbackState === 'paused' && !video.paused) { - tryMediaAction(EVENTS.PAUSE); + if (!await tryMediaAction(EVENTS.PAUSE)) { + return { status: 'apply_failed', reason: 'pause_action_failed' }; + } } if (shouldSeek) { _setSuppress('seek'); - seekVideo(video, mediaState.currentTime); + if (!await tryMediaAction(EVENTS.SEEK, { targetTime: mediaState.currentTime })) { + return { status: 'apply_failed', reason: 'seek_action_failed' }; + } } if (mediaState.playbackState === 'playing' && video.paused) { - tryMediaAction(EVENTS.PLAY); + if (!await tryMediaAction(EVENTS.PLAY)) { + return { status: 'apply_failed', reason: 'play_action_failed' }; + } + } + const verified = await pollCanonicalMediaState(mediaState, startedAt); + if (!verified) { + reportLog(`Canonical media state r${mediaState.revision} could not be verified`, 'warn'); + return { status: 'apply_failed', reason: 'verification_timeout' }; } scheduleProactiveHeartbeat(); - return { status: 'applied', revision: mediaState.revision, drift, sought: shouldSeek }; + return { + status: 'applied', + revision: mediaState.revision, + drift: verified.drift, + sought: shouldSeek + }; } catch (error) { reportLog(`Canonical media state apply failed: ${error.message}`, 'warn'); return { status: 'apply_failed' }; @@ -1362,7 +1422,10 @@ } if (message.type === 'APPLY_CANONICAL_MEDIA_STATE') { - sendResponse(applyCanonicalMediaState(message.mediaState)); + applyCanonicalMediaState(message.mediaState).then(sendResponse).catch(error => { + reportLog(`Canonical media state apply failed: ${error.message}`, 'warn'); + sendResponse({ status: 'apply_failed', reason: 'unexpected_error' }); + }); return true; } diff --git a/scripts/test-server-ws.mjs b/scripts/test-server-ws.mjs index 8cad6d4..ea40de8 100644 --- a/scripts/test-server-ws.mjs +++ b/scripts/test-server-ws.mjs @@ -10,7 +10,7 @@ import { materializeMediaIntent, reserveLatestMediaIntentSequence } from '../extension/offline-media-intent.js'; -import { FORCE_SYNC_TIMEOUT } from '../shared/constants.js'; +import { FORCE_SYNC_TARGET_TTL, FORCE_SYNC_TIMEOUT } from '../shared/constants.js'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const require = createRequire(path.join(__dirname, '..', 'server', 'package.json')); @@ -404,10 +404,20 @@ try { 'authorized EXECUTE commits the latest target visible to legacy peers'); assert.equal(competingForceState.updatedBy, 'msa'); + // The initiator's normal ACK timeout must still fit inside relay target + // retention. This is the exact fallback boundary used by background.js. + s(msa, 'force_sync_prepare', { targetTime: 925 }); + await w(msb, 'force_sync_prepare'); + mod.rooms.get(msrid).forceSyncTarget.preparedAt = Date.now() - FORCE_SYNC_TIMEOUT; + s(msa, 'force_sync_execute', {}); + await w(msb, 'force_sync_execute'); + assert.equal(mod.rooms.get(msrid).mediaState.currentTime, 925, + 'relay grace accepts EXECUTE at the client ACK-timeout boundary'); + s(msa, 'force_sync_prepare', { targetTime: 950 }); await w(msb, 'force_sync_prepare'); const beforeExpiredExecute = { ...mod.rooms.get(msrid).mediaState }; - mod.rooms.get(msrid).forceSyncTarget.preparedAt = Date.now() - FORCE_SYNC_TIMEOUT - 1; + mod.rooms.get(msrid).forceSyncTarget.preparedAt = Date.now() - FORCE_SYNC_TARGET_TTL - 1; msa._m.length = msb._m.length = 0; s(msa, 'force_sync_execute', {}); let expiredExecuteDropped = false; @@ -467,6 +477,37 @@ try { assert.equal(mod.rooms.get(msgateRid).mediaState.revision, gatedBaseline.revision + 1); assert.equal(mod.rooms.get(msgateRid).mediaState.currentTime, 800); + // Current receivers ignore duplicate/regressing seq. The relay must make the + // same decision before canonical mutation so late joiners see the same truth. + msgHost._m.length = msgGuest._m.length = 0; + s(msgHost, 'play', { currentTime: 820, seq: 10 }); + await w(msgGuest, 'play'); + const sequencedBaseline = { ...mod.rooms.get(msgateRid).mediaState }; + const sequencedPeerBaseline = { + ...Array.from(mod.rooms.get(msgateRid).peerData.values()) + .find(peer => peer.peerId === 'msg-host') + }; + s(msgHost, 'pause', { currentTime: 1, playbackState: 'paused', seq: 10 }); + let duplicateSequenceDropped = false; + try { await w(msgGuest, 'pause', 500); } catch { duplicateSequenceDropped = true; } + assert.ok(duplicateSequenceDropped, 'duplicate media seq is not relayed'); + s(msgHost, 'seek', { targetTime: 5, seq: 9 }); + let staleSequenceDropped = false; + try { await w(msgGuest, 'seek', 500); } catch { staleSequenceDropped = true; } + assert.ok(staleSequenceDropped, 'regressing media seq is not relayed'); + assert.deepEqual(mod.rooms.get(msgateRid).mediaState, sequencedBaseline, + 'regressing media seq cannot revise canonical state'); + assert.deepEqual( + Array.from(mod.rooms.get(msgateRid).peerData.values()) + .find(peer => peer.peerId === 'msg-host'), + sequencedPeerBaseline, + 'duplicate/regressing media seq cannot alter peer state used by later canonical updates'); + const msgLateJoiner = await c(); + const msgLateRoom = await j(msgLateJoiner, msgateRid, 'msg-late'); + assert.equal(msgLateRoom.mediaState.revision, sequencedBaseline.revision); + assert.equal(msgLateRoom.mediaState.currentTime, sequencedBaseline.currentTime, + 'late joiner receives the same state accepted by live receivers'); + const validationBaseline = { ...mod.rooms.get(msgateRid).mediaState }; for (const invalidPayload of [{ targetTime: null }, { targetTime: '50' }, { targetTime: {} }, {}]) { s(msgHost, 'seek', invalidPayload); diff --git a/server/index.js b/server/index.js index 31ab60a..1541765 100644 --- a/server/index.js +++ b/server/index.js @@ -4,7 +4,7 @@ import { fileURLToPath } from 'url'; import { Server } from 'socket.io'; import crypto from 'crypto'; import dotenv from 'dotenv'; -import { EVENTS, ERROR_CODES, OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION, CONTROL_MODES, CAPABILITIES, FORCE_SYNC_TIMEOUT, MAX_MEDIA_TIME } from '../shared/constants.js'; +import { EVENTS, ERROR_CODES, OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION, CONTROL_MODES, CAPABILITIES, FORCE_SYNC_TARGET_TTL, MAX_MEDIA_TIME } from '../shared/constants.js'; import { createChatEnvelope } from './chat.js'; import { commitForceSyncMediaState, @@ -176,6 +176,18 @@ const HOST_ONLY_GATED_EVENTS = new Set([ EVENTS.EPISODE_LOBBY_CANCEL ]); +// Current clients sequence room-moving media commands. The relay mirrors the +// receiver-side stale guard so a frame ignored by live peers cannot become the +// canonical snapshot shown to a later joiner. Legacy clients omit seq entirely +// and retain their pre-feature behavior. +const SEQUENCED_ROOM_EVENTS = new Set([ + EVENTS.PLAY, + EVENTS.PAUSE, + EVENTS.SEEK, + EVENTS.FORCE_SYNC_PREPARE, + EVENTS.FORCE_SYNC_EXECUTE +]); + // Features this relay supports, advertised to clients in ROOM_DATA so they can // enable matching UI/behavior only when the server actually backs it. Append a // flag here when a new server-gated feature ships (e.g. co-host promotion). @@ -615,6 +627,29 @@ io.on('connection', (socket) => { const validState = (val) => (val === 'playing' || val === 'paused') ? val : undefined; const validBool = (val) => typeof val === 'boolean' ? val : undefined; + const hasSequenceField = data.seq !== undefined; + const sequence = Number.isSafeInteger(data.seq) && data.seq >= 0 + ? data.seq + : undefined; + if (SEQUENCED_ROOM_EVENTS.has(eventName)) { + if (hasSequenceField && sequence === undefined) { + log('ROOM', `Dropped ${eventName} with invalid seq from ${mapping.peerId}`); + return; + } + if (sequence !== undefined) { + if (socket.data.mediaSequencePeerId !== mapping.peerId) { + socket.data.mediaSequencePeerId = mapping.peerId; + socket.data.lastMediaSequence = null; + } + if (Number.isSafeInteger(socket.data.lastMediaSequence) + && sequence <= socket.data.lastMediaSequence) { + log('ROOM', `Dropped stale ${eventName} from ${mapping.peerId} (seq ${sequence} <= ${socket.data.lastMediaSequence})`); + return; + } + socket.data.lastMediaSequence = sequence; + } + } + const existing = room.peerData.get(socket.id) || { peerId: mapping.peerId }; room.peerData.set(socket.id, { ...existing, @@ -632,7 +667,7 @@ io.on('connection', (socket) => { // --- S-3: Construct clean relay payload — never forward raw client data --- const relayPayload = { senderId: mapping.peerId, - seq: clampNum(data.seq, 0, Number.MAX_SAFE_INTEGER), + seq: sequence, currentTime: data.currentTime === null ? null : clampNum(data.currentTime, 0, MAX_MEDIA_TIME), targetTime: clampNum(data.targetTime, 0, MAX_MEDIA_TIME), playbackState: validState(data.playbackState), @@ -656,7 +691,7 @@ io.on('connection', (socket) => { const forceSyncTarget = room.forceSyncTarget; const targetExpired = forceSyncTarget && (!Number.isFinite(forceSyncTarget.preparedAt) - || mediaStateNow - forceSyncTarget.preparedAt > FORCE_SYNC_TIMEOUT); + || mediaStateNow - forceSyncTarget.preparedAt > FORCE_SYNC_TARGET_TTL); if (!forceSyncTarget || targetExpired) { log('ROOM', `Dropped force_sync_execute ${targetExpired ? 'with an expired target' : 'without a prepared target'} from ${mapping.peerId}`); room.forceSyncInitiator = null; diff --git a/shared/README.md b/shared/README.md index f00c52e..a2b851f 100644 --- a/shared/README.md +++ b/shared/README.md @@ -64,6 +64,7 @@ For the complete event list, read the `EVENTS` object in [`constants.js`](consta - `HEARTBEAT_INTERVAL`: content heartbeat interval in milliseconds. - `FORCE_SYNC_TIMEOUT`: max wait for force-sync ACKs. +- `FORCE_SYNC_TARGET_TTL`: relay retention for a prepared Force Sync target; includes post-timeout delivery grace. - `EPISODE_LOBBY_TIMEOUT`: max wait for episode-lobby readiness. - `MAX_MEDIA_TIME`: shared relay/extension upper bound, in seconds, for synchronized media positions. diff --git a/shared/constants.js b/shared/constants.js index 74b6c72..69e3fb2 100644 --- a/shared/constants.js +++ b/shared/constants.js @@ -99,4 +99,7 @@ export const MAX_MEDIA_TIME = 86400; export const HEARTBEAT_INTERVAL = 15000; // 15s export const FORCE_SYNC_TIMEOUT = 8500; // 8.5s timeout for force sync ACKs (must be > content.js poll timeout of 8s) +// Relay retention must outlive the client's ACK wait. Otherwise the timeout +// fallback EXECUTE arrives exactly when the relay expires its prepared target. +export const FORCE_SYNC_TARGET_TTL = FORCE_SYNC_TIMEOUT + 2000; export const EPISODE_LOBBY_TIMEOUT = 60000; // 60s timeout for episode lobby diff --git a/tests/e2e/extension.spec.mjs b/tests/e2e/extension.spec.mjs index 7b09594..bf0a60f 100644 --- a/tests/e2e/extension.spec.mjs +++ b/tests/e2e/extension.spec.mjs @@ -388,6 +388,75 @@ test('applies canonical recovery without echoing media commands or activity', as expect(historyAfter).toEqual(historyBefore); }); +test('recovers relay ROOM_DATA through background retries into the packed player', async ({ context, extensionId, baseURL }) => { + test.setTimeout(45_000); + const relay = await import('../../server/index.js'); + let legacy = null; + try { + await relay.startServer(0, '127.0.0.1'); + const port = relay.httpServer.address().port; + const roomId = `e2e-canonical-room-data-${Date.now()}`; + legacy = await connectLegacyRelayClient(port); + await joinLegacyRelayRoom(legacy, roomId, 'canonical-source'); + sendLegacyRelayEvent(legacy, 'play', { currentTime: 6, seq: 1, actionTimestamp: 1 }); + await expect.poll(() => relay.rooms.get(roomId)?.mediaState) + .toMatchObject({ revision: 1, playbackState: 'playing', currentTime: 6 }); + + const url = `${baseURL}/pages/simple-player.html`; + const page = await context.newPage(); + await page.goto(url); + await page.waitForFunction(() => window.__fixtureReady === true); + await selectTargetTab(context, extensionId, url); + await page.locator('#player').evaluate(video => { + const nativePlay = video.play.bind(video); + window.__koalaCanonicalPlayAttempts = 0; + Object.defineProperty(video, 'play', { + configurable: true, + value: () => { + window.__koalaCanonicalPlayAttempts++; + if (window.__koalaCanonicalPlayAttempts === 1) { + return Promise.reject(new DOMException('audit autoplay rejection', 'NotAllowedError')); + } + return nativePlay(); + } + }); + }); + legacy.messages.length = 0; + + 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: 'canonical-receiver' + })); + + await expect.poll(() => getExtensionState(context, extensionId, { type: 'GET_STATUS' })) + .toMatchObject({ status: 'connected', roomId, queuedLogicalEvents: 0 }); + await expect.poll(() => page.evaluate(() => window.__koalaCanonicalPlayAttempts)) + .toBeGreaterThanOrEqual(2); + await expect.poll(() => page.locator('#player').evaluate(video => ({ + paused: video.paused, + currentTime: video.currentTime + }))).toMatchObject({ paused: false }); + await expect.poll(() => page.locator('#player').evaluate(video => video.currentTime)) + .toBeGreaterThan(5); + + await page.waitForTimeout(700); + const recoveryEchoes = 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(recoveryEchoes).toEqual([]); + } finally { + try { legacy?.close(); } catch { /* already closed */ } + await relay.stopServerForTests(); + } +}); + test('@race reinjects after the target tab navigates', async ({ context, extensionId, baseURL }) => { const first = `${baseURL}/pages/iframe-player.html`; const page = await context.newPage();