fix(sync): cancel superseded canonical apply

2026-08-30 12:29:27 +00:00 · 2026-08-26 00:41:54 +02:00
parent 5854c106b8
commit e5110504c1
5 changed files with 189 additions and 10 deletions
@@ -1889,6 +1889,13 @@ function supersedeCanonicalMediaRecovery(reason) {
    if (!pending || !markCanonicalMediaStateHandled(roomId, pending.mediaState.revision)) {
        return false;
    }
+    // Tracking cancellation alone is insufficient: content.js may still be
+    // awaiting an asynchronous player action. Invalidate that operation before
+    // a newer local or remote control is allowed to win.
+    sendMessageToCurrentContent({
+        type: 'CANCEL_CANONICAL_MEDIA_STATE',
+        reason
+    }).catch(() => {});
    addLog(`Canonical media state r${pending.mediaState.revision} superseded by ${reason}`, 'info');
    return true;
 }
@@ -51,7 +51,8 @@ describe('canonical ROOM_DATA recovery contract', () => {
        const contentHandlerStart = contentSource.indexOf("message.type === 'APPLY_CANONICAL_MEDIA_STATE'");
        const serverCommandStart = contentSource.indexOf("message.type === 'SERVER_COMMAND'", contentHandlerStart);
        const internalHandler = contentSource.slice(contentHandlerStart, serverCommandStart);
-        expect(internalHandler).toContain('applyCanonicalMediaState(message.mediaState)');
+        expect(internalHandler).toContain('const applyGeneration = beginCanonicalMediaApply()');
+        expect(internalHandler).toContain('applyCanonicalMediaState(message.mediaState, applyGeneration)');
        expect(internalHandler).not.toContain('CMD_ACK');
        expect(internalHandler).not.toContain('CONTENT_EVENT');
    });
@@ -89,6 +90,7 @@ describe('canonical ROOM_DATA recovery contract', () => {
        const supersede = functionBody(backgroundSource, 'supersedeCanonicalMediaRecovery', 'performPendingCanonicalMediaStateApply');
        expect(supersede).toContain('canonicalMediaStateTracker.getPending(roomId)');
        expect(supersede).toContain('markCanonicalMediaStateHandled(roomId, pending.mediaState.revision)');
+        expect(supersede).toContain("type: 'CANCEL_CANONICAL_MEDIA_STATE'");
        expect(backgroundSource).toContain('function isCanonicalSupersedingControl(event, data)');
        expect(backgroundSource).toContain('supersedeCanonicalMediaRecovery(`newer ${event}`)');
        expect(backgroundSource).toContain('supersedeCanonicalMediaRecovery(`local ${message.action}`)');
@@ -103,15 +105,19 @@ describe('canonical ROOM_DATA recovery contract', () => {
        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).toContain('await pollCanonicalMediaState(mediaState, startedAt, applyGeneration)');
        expect(apply.indexOf("status: 'applied'"))
-            .toBeGreaterThan(apply.indexOf('await pollCanonicalMediaState(mediaState, startedAt)'));
+            .toBeGreaterThan(apply.indexOf('await pollCanonicalMediaState(mediaState, startedAt, applyGeneration)'));
+        expect(apply).toContain('isCanonicalMediaApplyCurrent(applyGeneration)');
+        expect(apply).toContain('restoreSupersedingLocalState(video)');
        expect(apply).toContain('if (hcmDesynced)');
        expect(apply).toContain('isDifferentEpisode(mediaState.mediaTitle, localMediaTitle)');
        expect(apply).toContain("status: 'ignored_episode_mismatch'");
        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)');
+            .toContain('applyCanonicalMediaState(message.mediaState, applyGeneration).then(sendResponse)');
+        expect(contentSource).toContain("message.type === 'CANCEL_CANONICAL_MEDIA_STATE'");
+        expect(contentSource).toContain('cancelCanonicalMediaApply(EVENTS.SEEK, video)');
    });
 });
@@ -91,6 +91,8 @@
    // While a timer exists, matching native events are consumed and not relayed.
    // Timers self-clean after 300ms if the native event never fires.
    let _suppressTimers = {};
+    let canonicalMediaApplyGeneration = 0;
+    let canonicalSupersedingLocalState = null;

    function _setSuppress(state) {
        if (_suppressTimers[state]) clearTimeout(_suppressTimers[state]);
@@ -1229,12 +1231,61 @@
        }
    }

-    function pollCanonicalMediaState(mediaState, startedAt, timeoutMs = 2500) {
+    function beginCanonicalMediaApply() {
+        canonicalMediaApplyGeneration++;
+        canonicalSupersedingLocalState = null;
+        return canonicalMediaApplyGeneration;
+    }
+
+    function cancelCanonicalMediaApply(action = null, video = null) {
+        canonicalMediaApplyGeneration++;
+        if ((action === EVENTS.PLAY || action === EVENTS.PAUSE || action === EVENTS.SEEK) && video) {
+            canonicalSupersedingLocalState = {
+                playbackState: action === EVENTS.PLAY
+                    ? 'playing'
+                    : (action === EVENTS.PAUSE ? 'paused' : (video.paused ? 'paused' : 'playing')),
+                currentTime: action === EVENTS.SEEK ? getSyncCurrentTime(video) : null
+            };
+        }
+        return canonicalMediaApplyGeneration;
+    }
+
+    function isCanonicalMediaApplyCurrent(generation) {
+        return !destroyed && generation === canonicalMediaApplyGeneration;
+    }
+
+    async function restoreSupersedingLocalState(video) {
+        const state = canonicalSupersedingLocalState;
+        if (!state || !video) return;
+
+        if (state.playbackState === 'paused' && !video.paused) {
+            _setSuppress('paused');
+            video.pause();
+        }
+        if (Number.isFinite(state.currentTime)) {
+            const currentTime = getSyncCurrentTime(video);
+            if (currentTime === null || Math.abs(currentTime - state.currentTime) >= MIN_SEEK_DELTA) {
+                _setSuppress('seek');
+                seekVideo(video, state.currentTime);
+            }
+        }
+        if (state.playbackState === 'playing' && video.paused) {
+            _setSuppress('playing');
+            try {
+                await video.play();
+            } catch (error) {
+                _clearSuppress('playing');
+                reportLog(`Could not restore locally superseding playback: ${error.message}`, 'warn');
+            }
+        }
+    }
+
+    function pollCanonicalMediaState(mediaState, startedAt, applyGeneration, timeoutMs = 2500) {
        return new Promise((resolve) => {
            const interval = 100;
            const finishAt = Date.now() + timeoutMs;
            const timer = setInterval(() => {
-                if (destroyed) {
+                if (!isCanonicalMediaApplyCurrent(applyGeneration)) {
                    clearInterval(timer);
                    seekPollTimers.delete(timer);
                    resolve(null);
@@ -1263,7 +1314,7 @@
        });
    }

-    async function applyCanonicalMediaState(mediaState) {
+    async function applyCanonicalMediaState(mediaState, applyGeneration) {
        if (!mediaState || typeof mediaState !== 'object'
            || !Number.isSafeInteger(mediaState.revision) || mediaState.revision < 1
            || (mediaState.playbackState !== 'playing' && mediaState.playbackState !== 'paused')
@@ -1276,6 +1327,7 @@
                && typeof mediaState.mediaTitle !== 'string')) {
            return { status: 'invalid' };
        }
+        if (!isCanonicalMediaApplyCurrent(applyGeneration)) return { status: 'superseded' };
        if (hcmDesynced) return { status: 'ignored_desynced' };
        const localMediaTitle = getMediaTitle();
        if (_autoSyncEnabled && isDifferentEpisode(mediaState.mediaTitle, localMediaTitle)) {
@@ -1290,6 +1342,10 @@
        const drift = currentTime === null ? null : mediaState.currentTime - currentTime;
        const shouldSeek = drift === null || Math.abs(drift) >= MIN_SEEK_DELTA;
        const startedAt = Date.now();
+        const superseded = async () => {
+            await restoreSupersedingLocalState(video);
+            return { status: 'superseded', revision: mediaState.revision };
+        };

        try {
            // Paused recovery pauses before seeking; playing recovery seeks before
@@ -1299,19 +1355,25 @@
                if (!await tryMediaAction(EVENTS.PAUSE)) {
                    return { status: 'apply_failed', reason: 'pause_action_failed' };
                }
+                if (!isCanonicalMediaApplyCurrent(applyGeneration)) return superseded();
            }
+            if (!isCanonicalMediaApplyCurrent(applyGeneration)) return superseded();
            if (shouldSeek) {
                _setSuppress('seek');
                if (!await tryMediaAction(EVENTS.SEEK, { targetTime: mediaState.currentTime })) {
                    return { status: 'apply_failed', reason: 'seek_action_failed' };
                }
+                if (!isCanonicalMediaApplyCurrent(applyGeneration)) return superseded();
            }
+            if (!isCanonicalMediaApplyCurrent(applyGeneration)) return superseded();
            if (mediaState.playbackState === 'playing' && video.paused) {
                if (!await tryMediaAction(EVENTS.PLAY)) {
                    return { status: 'apply_failed', reason: 'play_action_failed' };
                }
+                if (!isCanonicalMediaApplyCurrent(applyGeneration)) return superseded();
            }
-            const verified = await pollCanonicalMediaState(mediaState, startedAt);
+            const verified = await pollCanonicalMediaState(mediaState, startedAt, applyGeneration);
+            if (!isCanonicalMediaApplyCurrent(applyGeneration)) return superseded();
            if (!verified) {
                reportLog(`Canonical media state r${mediaState.revision} could not be verified`, 'warn');
                return { status: 'apply_failed', reason: 'verification_timeout' };
@@ -1429,8 +1491,15 @@
            return true;
        }

+        if (message.type === 'CANCEL_CANONICAL_MEDIA_STATE') {
+            cancelCanonicalMediaApply();
+            sendResponse({ status: 'cancelled' });
+            return true;
+        }
+
        if (message.type === 'APPLY_CANONICAL_MEDIA_STATE') {
-            applyCanonicalMediaState(message.mediaState).then(sendResponse).catch(error => {
+            const applyGeneration = beginCanonicalMediaApply();
+            applyCanonicalMediaState(message.mediaState, applyGeneration).then(sendResponse).catch(error => {
                reportLog(`Canonical media state apply failed: ${error.message}`, 'warn');
                sendResponse({ status: 'apply_failed', reason: 'unexpected_error' });
            });
@@ -1440,6 +1509,9 @@
        if (message.type === 'SERVER_COMMAND') {
            const { action, payload } = message;
            let actionCompleted = false;
+            if ([EVENTS.PLAY, EVENTS.PAUSE, EVENTS.SEEK, EVENTS.FORCE_SYNC_PREPARE, EVENTS.FORCE_SYNC_EXECUTE].includes(action)) {
+                cancelCanonicalMediaApply();
+            }

            // Host Control Mode: while watching on our own (desynced), don't apply
            // host commands. Only ACK FORCE_SYNC_PREPARE — that's the one the host's
@@ -1783,6 +1855,10 @@
            return;
        }

+        if (action === EVENTS.PLAY || action === EVENTS.PAUSE) {
+            cancelCanonicalMediaApply(action, video);
+        }
+
        // Suppress only SEEK during visibility grace period (tab re-focus ghost jump).
        // Play/Pause pass through — user may want to immediately pause after tabbing back.
        if (Date.now() < visibilityGraceUntil && action === EVENTS.SEEK) return;
@@ -1936,6 +2012,13 @@
            return;
        }

+        // `seeking` normally captured the local state synchronously. Keep this
+        // as a fallback for players that emit only `seeked`, without replacing
+        // the earlier paused/playing state after an async stale action resolves.
+        if (!canonicalSupersedingLocalState) {
+            cancelCanonicalMediaApply(EVENTS.SEEK, video);
+        }
+
        // Step 4: Debounce rapid consecutive seeks (e.g. scrubbing)
        // — wait 300ms for the user to settle before relaying
        if (seekDebounceTimer) clearTimeout(seekDebounceTimer);
@@ -1953,6 +2036,18 @@
        }, 300);
    };

+    const handleSeeking = event => {
+        if (!isCurrentVideoEvent(event)) return;
+        const video = event.currentTarget;
+        const current = getSyncCurrentTime(video);
+        if (current === null) return;
+        if (expectedSeekTime !== null && Math.abs(current - expectedSeekTime) < 1.0) return;
+        if (Date.now() < visibilityGraceUntil) return;
+        const delta = lastReportedSeekTime !== null ? Math.abs(current - lastReportedSeekTime) : null;
+        if (delta !== null && delta < MIN_SEEK_DELTA) return;
+        cancelCanonicalMediaApply(EVENTS.SEEK, video);
+    };
+

    let lastVideoSrc = undefined;

@@ -1967,6 +2062,7 @@
        if (handlers) {
            video.removeEventListener('play', handlers.play);
            video.removeEventListener('pause', handlers.pause);
+            if (handlers.seeking) video.removeEventListener('seeking', handlers.seeking);
            video.removeEventListener('seeked', handlers.seeked);
            video.removeEventListener('loadeddata', handlers.loadeddata);
            if (handlers.waiting) video.removeEventListener('waiting', handlers.waiting);
@@ -1999,9 +2095,17 @@
            const existing = video._koalaHandlers;
            if (existing) detachVideoListeners(video);
            activeVideo = video;
-            video._koalaHandlers = { play: handlePlay, pause: handlePause, seeked: handleSeeked, loadeddata: handleLoadedData, waiting: handleWaiting };
+            video._koalaHandlers = {
+                play: handlePlay,
+                pause: handlePause,
+                seeking: handleSeeking,
+                seeked: handleSeeked,
+                loadeddata: handleLoadedData,
+                waiting: handleWaiting
+            };
            video.addEventListener('play', handlePlay);
            video.addEventListener('pause', handlePause);
+            video.addEventListener('seeking', handleSeeking);
            video.addEventListener('seeked', handleSeeked);
            video.addEventListener('loadeddata', handleLoadedData);
            video.addEventListener('waiting', handleWaiting);
@@ -9,6 +9,7 @@ const backgroundSource = fs.readFileSync(path.join(extensionDir, 'background.js'
 function functionBody(name, nextName) {
    const start = backgroundSource.indexOf(`function ${name}(`);
    const end = backgroundSource.indexOf(`function ${nextName}(`, start + 1);
+    expect(start).toBeGreaterThan(-1);
    return backgroundSource.slice(start, end === -1 ? backgroundSource.length : end);
 }

@@ -427,6 +427,67 @@ test('applies canonical recovery without echoing media commands or activity', as
    expect(historyAfter).toEqual(historyBefore);
 });

+test('local media input cancels an in-flight canonical recovery', async ({ context, extensionId, baseURL }) => {
+    const url = `${baseURL}/pages/simple-player.html`;
+    const page = await context.newPage();
+    await page.goto(url);
+    await page.waitForFunction(() => window.__fixtureReady === true);
+
+    const { tabId } = await selectTargetTab(context, extensionId, url);
+    await expect.poll(() => page.locator('#player').getAttribute('data-koala-attached')).toBe('true');
+    await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async selectedTabId => {
+        await chrome.scripting.executeScript({
+            target: { tabId: selectedTabId },
+            world: 'ISOLATED',
+            func: () => {
+                const video = document.querySelector('#player');
+                if (!video) throw new Error('canonical local-supersession fixture video missing');
+                video.pause();
+                video.currentTime = 0;
+                const nativePlay = video.play.bind(video);
+                video.dataset.koalaDelayedPlayAttempts = '0';
+                Object.defineProperty(video, 'play', {
+                    configurable: true,
+                    value: () => {
+                        const attempts = Number(video.dataset.koalaDelayedPlayAttempts || '0') + 1;
+                        video.dataset.koalaDelayedPlayAttempts = String(attempts);
+                        return nativePlay().then(() => new Promise(resolve => setTimeout(resolve, 400)));
+                    }
+                });
+            }
+        });
+    }, tabId));
+
+    let applyResponse = null;
+    const applyPromise = applyCanonicalMediaState(context, extensionId, tabId, {
+        revision: 20,
+        playbackState: 'playing',
+        currentTime: 6,
+        updatedBy: 'peer-a'
+    }).then(response => {
+        applyResponse = response;
+        return response;
+    });
+    await expect.poll(async () => ({
+        attempts: await page.locator('#player').evaluate(video =>
+            Number(video.dataset.koalaDelayedPlayAttempts || '0')),
+        response: applyResponse
+    })).toMatchObject({ attempts: 1, response: null });
+    await expect.poll(() => page.locator('#player').evaluate(video => video.paused)).toBe(false);
+
+    await page.locator('#player').evaluate(video => {
+        video.pause();
+        video.currentTime = 10;
+    });
+    await expect.poll(() => page.locator('#player').evaluate(video => video.currentTime))
+        .toBeGreaterThan(9);
+
+    await expect(applyPromise).resolves.toMatchObject({ status: 'superseded' });
+    await page.waitForTimeout(700);
+    expect(await page.locator('#player').evaluate(video => video.paused)).toBe(true);
+    expect(await page.locator('#player').evaluate(video => video.currentTime)).toBeGreaterThan(9);
+});
+
 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');