diff --git a/web/src/player/hooks/usePlaybackSession.test.ts b/web/src/player/hooks/usePlaybackSession.test.ts index eca191dbf0..0e464c67d1 100644 --- a/web/src/player/hooks/usePlaybackSession.test.ts +++ b/web/src/player/hooks/usePlaybackSession.test.ts @@ -4897,6 +4897,12 @@ describe("usePlaybackSession deferred push authority", () => { const collisionAudio = [ { codec: "ac3", channels: 2, layout: "stereo", language: "fra", default: true }, ]; + // Two entries so a fold that lands is observable even when the incumbent menu + // is non-empty and no identity move is involved. + const richerCollisionAudio = [ + { codec: "ac3", channels: 2, layout: "stereo", language: "fra", default: true }, + { codec: "ac3", channels: 2, layout: "stereo", language: "ita", default: false }, + ]; it("rejects an outgoing source commit even when it matches the outgoing rendered file", async () => { const startBodies: Array<{ file_id: number }> = []; @@ -5320,12 +5326,14 @@ describe("usePlaybackSession deferred push authority", () => { unmount(); }); - it("drops a same-file inventory after the live source already rotated", async () => { + it("drops a same-file sibling inventory after the live source already rotated", async () => { // The plan owns file 8 without naming a candidate (v2 wire). A poll moves // the live source to candidate A. During a replacement that also settles on - // file 8 without a candidate, an A inventory is queued. The outgoing live - // URI is A even though the plan's captured URI is null, so file equality - // must not admit A. + // file 8 without a candidate, a C inventory is queued. C is a same-file + // sibling of the live outgoing A: the URI-less plan cannot vouch for it and + // the concrete live URI contradicts it, so it is a genuine collision and is + // dropped. (A same-file push that agrees with the live source is admitted; + // see the URI-less acceptance regression below.) const startBodies: Array<{ file_id: number }> = []; let releaseSwitch: ((response: Response) => void) | undefined; const switchResponse = new Promise((resolve) => { @@ -5370,14 +5378,16 @@ describe("usePlaybackSession deferred push authority", () => { act(() => result.current.switchVersion(9, 0)); await waitFor(() => expect(startBodies).toHaveLength(2)); - // The outgoing live candidate A is deferred behind the pending replacement. + // Candidate C on the same file is deferred behind the pending replacement. + // It contradicts the live outgoing A, so the URI-less winner cannot vouch + // for it and it must not be folded. act(() => result.current.applyInventoryUpdate({ session_id: "session-1", - inventory_revision: "inv:live-a", + inventory_revision: "inv:sibling-c", inventory_status: "verified", effective_media_file_id: 8, - effective_virtual_uri: "virtual://movie/x?result=A", + effective_virtual_uri: "virtual://movie/x?result=C", audio_tracks: outgoingAudio, }), ); @@ -5405,13 +5415,102 @@ describe("usePlaybackSession deferred push authority", () => { }); await waitFor(() => expect(result.current.sessionId).toBe("session-2")); - // The outgoing A revision is not corroborated by the URI-less winner and is - // dropped; the plan's own (empty, candidate-less) identity is what stands. + // Candidate C is a same-file sibling of the live outgoing A, so the + // URI-less winner cannot vouch for it and it is dropped; the plan's own + // (empty, candidate-less) identity is what stands. expect(result.current.effectiveVirtualUri).toBeNull(); expect(result.current.planAudioTracks).toEqual([]); unmount(); }); + it("admits a same-file inventory push against a URI-less settled plan", async () => { + // The v2 start wire omits the candidate URI, so the settled replacement plan + // names only file 8. The verified inventory push names that same file with + // the candidate the plan leaves out; it is the only carrier of the live + // identity and must be folded rather than refused for the plan's silence. + const startBodies: Array<{ file_id: number }> = []; + let releaseSwitch: ((response: Response) => void) | undefined; + const switchResponse = new Promise((resolve) => { + releaseSwitch = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + const body = JSON.parse(String(init?.body)) as { file_id: number }; + startBodies.push(body); + if (startBodies.length === 2) return switchResponse; + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => result.current.switchVersion(8, 0)); + await waitFor(() => expect(startBodies).toHaveLength(2)); + + // A verified inventory for candidate B on the replacement's file 8 is + // deferred behind the pending switch. + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:uri-less", + inventory_status: "verified", + effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=B", + audio_tracks: rotatedAudio, + }), + ); + + await act(async () => { + releaseSwitch?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-2", + playback_plan: fixturePlanV3({ + session_id: "session-2", + plan_id: "plan:uri-less-2", + plan_attempt_key: "v3:uri-less-2", + requested_media_file_id: 8, + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await switchResponse; + }); + + await waitFor(() => expect(result.current.mediaFileId).toBe(8)); + // The push names the plan's own file, so it is folded; the plan itself + // carries no candidate URI. + expect(result.current.plan?.effective_virtual_uri).toBeUndefined(); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + it("holds an older inventory behind a newer source commit during a replan", async () => { // A replanning session defers an inventory for candidate C, then a newer // source commit for candidate B arrives while the same replan is still in @@ -5851,7 +5950,12 @@ describe("usePlaybackSession deferred push authority", () => { unmount(); }); - it("does not refold a rejected inventory revision after an accepted rotation", async () => { + it("refolds a refused inventory revision once a later rotation matches it", async () => { + // A revision refused by a flush is not the same statement as a revision that + // was applied: the refusal is about the settled plan at that moment, and the + // live source can still move onto the candidate the revision names later in + // the same generation. Recording the refusal would swallow that redelivery + // and leave the track menus stale for the rest of the session. let releaseReplan: ((response: Response) => void) | undefined; const heldReplan = new Promise((resolve) => { releaseReplan = resolve; @@ -5892,15 +5996,16 @@ describe("usePlaybackSession deferred push authority", () => { }); await waitFor(() => expect(result.current.replanning).toBe(true)); - // A file-only inventory revision is held behind the replan. The settled - // replacement names candidate B by URI, so a file-only revision cannot be - // tied to it and is refused (its revision recorded as handled). + // A revision for candidate C is held behind the replan. The settled + // replacement names candidate B, so C is a same-file collision and is + // refused rather than folded. act(() => result.current.applyInventoryUpdate({ session_id: "session-1", - inventory_revision: "inv:rejected", + inventory_revision: "inv:redelivered", inventory_status: "verified", effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=C", audio_tracks: collisionAudio, }), ); @@ -5913,8 +6018,8 @@ describe("usePlaybackSession deferred push authority", () => { outcome: "playable", session_id: "session-1", playback_plan: fixturePlanV3({ - plan_id: "plan:reject-again", - plan_attempt_key: "v3:reject-again", + plan_id: "plan:redeliver0001", + plan_attempt_key: "v3:redeliver0001", effective_media_file_id: 8, effective_virtual_uri: "virtual://movie/x?result=B", audio_tracks: [], @@ -5924,39 +6029,1011 @@ describe("usePlaybackSession deferred push authority", () => { await heldReplan; }); - await waitFor(() => expect(result.current.plan?.plan_id).toBe("plan:reject-again")); + await waitFor(() => expect(result.current.plan?.plan_id).toBe("plan:redeliver0001")); expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); expect(result.current.planAudioTracks).toEqual([]); - // An accepted source rotation moves the menu to candidate D with an empty - // declared inventory. That identity change must not forget the revision the - // flush already handled. + // A rotation moves the live source onto C with an empty inventory. The + // revision the flush refused is still unrecorded, so the redelivery now + // matches the live identity and must fold. act(() => result.current.applyCommittedSource( { effectiveMediaFileId: 8, - effectiveVirtualUri: "virtual://movie/x?result=D", + effectiveVirtualUri: "virtual://movie/x?result=C", inventoryStatus: "verified", }, [], ), ); - expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=D"); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=C"); expect(result.current.planAudioTracks).toEqual([]); - // Redelivering the refused revision is a no-op, not a second fold. The - // rejected payload carries a track the accepted source does not have, so a - // forgotten revision would show a stale track under candidate D. act(() => result.current.applyInventoryUpdate({ session_id: "session-1", - inventory_revision: "inv:rejected", + inventory_revision: "inv:redelivered", inventory_status: "verified", effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=C", audio_tracks: collisionAudio, }), ); - expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=D"); + expect(result.current.planAudioTracks).toEqual(collisionAudio); + + // Now that the revision has genuinely been applied, its duplicate is + // suppressed: a redelivery with a different inventory must not overwrite. + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:redelivered", + inventory_status: "verified", + effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=C", + audio_tracks: rotatedAudio, + }), + ); + expect(result.current.planAudioTracks).toEqual(collisionAudio); + unmount(); + }); + + it("admits a deferred URI-only inventory push against a URI-less settled plan", async () => { + // A partial push may name only the candidate URI and no file id. It is + // deferred behind a replan whose replacement plan is URI-less (v2 wire) and + // the live source is URI-less too, so the URI is the only key the push + // offers. It must be folded rather than dropped for the missing file. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 8, + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 8, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:uri-only", + inventory_status: "verified", + effective_virtual_uri: "virtual://movie/x?result=B", + audio_tracks: rotatedAudio, + }), + ); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + plan_id: "plan:uri-only0001", + plan_attempt_key: "v3:uri-only0001", + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await heldReplan; + }); + + await waitFor(() => expect(result.current.plan?.plan_id).toBe("plan:uri-only0001")); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("drops a file-only inventory under a concrete live candidate on the same file", async () => { + // The plan is URI-less on file 8 and a poll has resolved the live source to + // candidate A. A later file-only inventory revision names file 8 but no + // candidate, so it cannot be tied to A; folding it would overwrite the + // poll's proven inventory with an unproven one. + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 8, + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 8, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => result.current.applyAudioInventory(rotatedAudio, 8, "virtual://movie/x?result=A")); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=A"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:file-only", + inventory_status: "verified", + effective_media_file_id: 8, + audio_tracks: richerCollisionAudio, + }), + ); + + // The concrete live candidate A stands; the file-only revision is refused + // even though its richer list would otherwise replace the menu. + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=A"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("drops a file-only inventory against a URI-ful settled plan via the direct path", async () => { + // The settled plan names candidate B by URI. A file-only inventory revision + // (the shape a refused-then-redelivered revision takes) names the plan's file + // but no candidate, so the direct path must not fold it over B's inventory. + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=B", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 8, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:file-only-uri-ful", + inventory_status: "verified", + effective_media_file_id: 8, + audio_tracks: richerCollisionAudio, + }), + ); + + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(outgoingAudio); + unmount(); + }); + + it("refuses a file-only inventory deferred before a poll folded a same-file candidate", async () => { + // A FILE-ONLY inventory revision is queued behind a replan. Its captured + // outgoing identity is URI-less, so the arrival-time baseline alone would + // admit it. While it waits, a poll (which bypasses the adoption barrier) + // folds candidate B on the same file. The flush must judge the queued push + // against the candidate the menus now carry, not only the arrival capture: + // otherwise the unproven file-only revision overwrites B's proven inventory. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 8, + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 8, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + // The file-only inventory is held; its captured outgoing identity is + // URI-less, and the richer list would replace a one-entry menu. + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:file-only-deferred", + inventory_status: "verified", + effective_media_file_id: 8, + audio_tracks: richerCollisionAudio, + }), + ); + // The poll folds candidate B under the same file while the push is queued. + act(() => result.current.applyAudioInventory(rotatedAudio, 8, "virtual://movie/x?result=B")); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "adaptation_unavailable", + terminal: { + reason: "video_conversion_unsupported", + message: "No executor can transcode this source.", + retryable: false, + }, + }), + ); + await heldReplan; + }); + + // The plan was refused, so it still stands URI-less on file 8; the live source + // is the folded B. The file-only revision must not overwrite B's inventory. + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("refuses a file-only inventory deferred while a concrete candidate is live", async () => { + // A poll moves the live source to candidate B on file 8, then a switch starts + // and a file-only inventory revision for file 8 is deferred. The replacement + // settles URI-less on file 8, so file equality alone would admit it; the + // concrete arrival-time live candidate B must refuse it instead. + const startBodies: Array<{ file_id: number }> = []; + let releaseSwitch: ((response: Response) => void) | undefined; + const switchResponse = new Promise((resolve) => { + releaseSwitch = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + const body = JSON.parse(String(init?.body)) as { file_id: number }; + startBodies.push(body); + if (startBodies.length === 2) return switchResponse; + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + // The poll folds candidate B on file 8 before the switch barrier opens. + act(() => result.current.applyAudioInventory(rotatedAudio, 8, "virtual://movie/x?result=B")); + act(() => result.current.switchVersion(9, 0)); + await waitFor(() => expect(startBodies).toHaveLength(2)); + + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:file-only-live", + inventory_status: "verified", + effective_media_file_id: 8, + audio_tracks: richerCollisionAudio, + }), + ); + + await act(async () => { + releaseSwitch?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-2", + playback_plan: fixturePlanV3({ + session_id: "session-2", + plan_id: "plan:file-only-live2", + plan_attempt_key: "v3:file-only-live2", + requested_media_file_id: 9, + // The replacement falls back onto file 8, URI-less. + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await switchResponse; + }); + + await waitFor(() => expect(result.current.sessionId).toBe("session-2")); + // The replacement plan is URI-less on file 8, and the file-only revision is + // refused against the concrete candidate that was live when it arrived, so + // its richer track list is not folded and the plan's own (empty) identity + // and inventory stand. + expect(result.current.effectiveVirtualUri).toBeNull(); + expect(result.current.planAudioTracks).toEqual([]); + unmount(); + }); + + it("keeps the newer of two same-file inventories replayed in one flush", async () => { + // Two concrete inventory revisions for the same file are queued behind a + // replan: C first, then B. The settled plan is URI-less on file 8, so both + // pass the identity gate and arrival order must decide — B, sent later, wins. + // If the flush judged each entry against the menu identity an earlier entry + // had already moved, C would veto B and invert arrival order. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:order-c", + inventory_status: "verified", + effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=C", + audio_tracks: collisionAudio, + }), + ); + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:order-b", + inventory_status: "verified", + effective_media_file_id: 8, + effective_virtual_uri: "virtual://movie/x?result=B", + audio_tracks: rotatedAudio, + }), + ); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + plan_id: "plan:order-b0000001", + plan_attempt_key: "v3:order-b0000001", + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await heldReplan; + }); + + await waitFor(() => expect(result.current.mediaFileId).toBe(8)); + // The later inventory (B) wins; the earlier one (C) must not veto it. + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("admits a queued push naming the replaced plan's explicitly winning candidate", async () => { + // The session was on (file 7, candidate A) and the viewer explicitly picked + // a row the server resolved to the same file's candidate B. A probe push for + // B is queued behind the switch. The outgoing candidate A must not veto the + // plan the server just selected: a replaced URI-ful winner is decided by + // exact identity before any ambiguity guard. + const startBodies: Array<{ file_id: number }> = []; + let releaseSwitch: ((response: Response) => void) | undefined; + const switchResponse = new Promise((resolve) => { + releaseSwitch = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + const body = JSON.parse(String(init?.body)) as { file_id: number }; + startBodies.push(body); + if (startBodies.length === 2) return switchResponse; + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => result.current.switchVersion(99, 0)); + await waitFor(() => expect(startBodies).toHaveLength(2)); + + act(() => + result.current.applyCommittedSource( + { + effectiveMediaFileId: 7, + effectiveVirtualUri: "virtual://movie/x?result=B", + inventoryStatus: "verified", + }, + rotatedAudio, + ), + ); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=A"); + + await act(async () => { + releaseSwitch?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-2", + playback_plan: fixturePlanV3({ + session_id: "session-2", + plan_id: "plan:retain-b000001", + plan_attempt_key: "v3:retain-b000001", + requested_media_file_id: 99, + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=B", + audio_tracks: [], + }), + }), + ); + await switchResponse; + }); + + await waitFor(() => + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"), + ); + // The winner names candidate B outright, so its corroborating inventory is + // folded rather than vetoed by the outgoing candidate A. + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("keeps the newer same-file source commit over an earlier deferred source commit", async () => { + // The transport commits to candidate C, then to candidate B, on the same + // file while a replan is in flight. Both are source commits. The settled + // winner is URI-less, so arrival order decides: the queue's own earlier + // commit is not an external authority the newer one must answer to. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + act(() => + result.current.applyCommittedSource( + { + effectiveMediaFileId: 8, + effectiveVirtualUri: "virtual://movie/x?result=C", + inventoryStatus: "verified", + }, + collisionAudio, + ), + ); + act(() => + result.current.applyCommittedSource( + { + effectiveMediaFileId: 8, + effectiveVirtualUri: "virtual://movie/x?result=B", + inventoryStatus: "verified", + }, + rotatedAudio, + ), + ); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + plan_id: "plan:source-order01", + plan_attempt_key: "v3:source-order01", + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await heldReplan; + }); + + await waitFor(() => expect(result.current.mediaFileId).toBe(8)); + // The newer commit (B) must remain the final identity and inventory; the + // earlier same-file commit (C) is not an authority that can veto it. + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("does not fold a URI-only inventory for the outgoing file onto a replaced file", async () => { + // The plan owns file 7 with no candidate URI. A URI-only inventory push for + // its candidate is queued during a switch, and the replacement settles on a + // different file 8, also URI-less. The push proves no relation to file 8, so + // it must not inherit the replacement's file id. + const startBodies: Array<{ file_id: number }> = []; + let releaseSwitch: ((response: Response) => void) | undefined; + const switchResponse = new Promise((resolve) => { + releaseSwitch = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + const body = JSON.parse(String(init?.body)) as { file_id: number }; + startBodies.push(body); + if (startBodies.length === 2) return switchResponse; + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => result.current.switchVersion(8, 0)); + await waitFor(() => expect(startBodies).toHaveLength(2)); + + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:uri-only-outgoing", + inventory_status: "verified", + effective_virtual_uri: "virtual://movie/x?result=C", + audio_tracks: rotatedAudio, + }), + ); + + await act(async () => { + releaseSwitch?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-2", + playback_plan: fixturePlanV3({ + session_id: "session-2", + plan_id: "plan:cross-file-01", + plan_attempt_key: "v3:cross-file-01", + requested_media_file_id: 8, + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await switchResponse; + }); + + await waitFor(() => expect(result.current.sessionId).toBe("session-2")); + // No proven relation to the settled file, so the push is refused and a + // current-source refresh must re-establish the candidate instead. + expect(result.current.effectiveVirtualUri).toBeNull(); + expect(result.current.planAudioTracks).toEqual([]); + unmount(); + }); + + it("does not let a source commit overwrite a poll folded after it was queued", async () => { + // The review's ordering: a URI-bearing source commit queued while a replan + // owns the session names file 8 candidate C. Before the replan settles, a + // catalog poll (which bypasses the adoption barrier) folds the newer candidate + // B on the same file. The replacement plan then lands URI-less on file 8, + // overwriting the menu mirror with the plan's own candidate-less identity, so + // `appliedIdentity` alone no longer names B. The commit's arrival outgoing is + // cross-file (file 7 candidate A), so an arrival-order-only judgement against + // the queue would admit C and install it; the poll candidate remembered since + // the fold must still veto the stale commit. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + // The transport's older commit for candidate C is held behind the replan. + act(() => + result.current.applyCommittedSource( + { + effectiveMediaFileId: 8, + effectiveVirtualUri: "virtual://movie/x?result=C", + inventoryStatus: "verified", + }, + collisionAudio, + ), + ); + // A poll then folds the newer candidate B on the same file; the menus move to + // B while the commit still waits. + act(() => result.current.applyAudioInventory(rotatedAudio, 8, "virtual://movie/x?result=B")); + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + plan_id: "plan:poll-stale01", + plan_attempt_key: "v3:poll-stale01", + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await heldReplan; + }); + + // The replacement plan is URI-less on file 8 and cannot vouch for C, and the + // poll's B was folded after C was queued. The stale commit must not install C + // over the newer poll; the plan's own candidate-less identity stands. + await waitFor(() => expect(result.current.plan?.plan_id).toBe("plan:poll-stale01")); + expect(result.current.effectiveVirtualUri).toBeNull(); + expect(result.current.planAudioTracks).toEqual([]); + unmount(); + }); + + it("keeps a poll's candidate over a source commit queued after it", async () => { + // The mirror of the previous ordering: the poll folds candidate B first, so + // that candidate is the commit's arrival outgoing baseline. A later sibling + // commit for candidate C on the same file is a genuine same-file collision and + // is refused, whether or not it is treated as newer than an applied poll. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 7, + effective_virtual_uri: "virtual://movie/x?result=A", + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 7, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + act(() => result.current.applyAudioInventory(rotatedAudio, 8, "virtual://movie/x?result=B")); + // The commit for C arrives after the poll; its captured outgoing is B. + act(() => + result.current.applyCommittedSource( + { + effectiveMediaFileId: 8, + effectiveVirtualUri: "virtual://movie/x?result=C", + inventoryStatus: "verified", + }, + collisionAudio, + ), + ); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "adaptation_unavailable", + terminal: { + reason: "video_conversion_unsupported", + message: "No executor can transcode this source.", + retryable: false, + }, + }), + ); + await heldReplan; + }); + + expect(result.current.effectiveVirtualUri).toBe("virtual://movie/x?result=B"); + expect(result.current.planAudioTracks).toEqual(rotatedAudio); + unmount(); + }); + + it("refuses a deferred push with neither URI nor file against a URI-less winner", async () => { + // A malformed or partial push may name no source at all. It cannot be tied to + // the settled adoption, so it must not fold its audio list over a URI-less + // winner that the wire already names by file alone. + let releaseReplan: ((response: Response) => void) | undefined; + const heldReplan = new Promise((resolve) => { + releaseReplan = resolve; + }); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/playback/start")) { + return jsonResponse( + { + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + effective_media_file_id: 8, + audio_tracks: outgoingAudio, + }), + }, + { status: 201 }, + ); + } + if (url.endsWith("/playback/session-1/replan")) return heldReplan; + if (url.endsWith("/playback/route-events")) return new Response(null, { status: 202 }); + if (init?.method === "DELETE") return new Response(null, { status: 204 }); + throw new Error(`Unexpected request: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const { result, unmount } = renderHook( + () => usePlaybackSession("request-1", [], [], 8, 0, false, "auto"), + { wrapper }, + ); + await waitFor(() => expect(result.current.plan).not.toBeNull()); + + act(() => { + void result.current.refreshSubtitles(120); + }); + await waitFor(() => expect(result.current.replanning).toBe(true)); + + // No file id and no URI: an identity-less push that cannot be tied to any + // adoption. Its richer list must not replace the URI-less winner's menu. + act(() => + result.current.applyInventoryUpdate({ + session_id: "session-1", + inventory_revision: "inv:bogus-identity", + inventory_status: "verified", + audio_tracks: richerCollisionAudio, + }), + ); + + await act(async () => { + releaseReplan?.( + jsonResponse({ + protocol_version: 3, + server_features: ["playback_plan_v3"], + outcome: "playable", + session_id: "session-1", + playback_plan: fixturePlanV3({ + plan_id: "plan:bogus-identity", + plan_attempt_key: "v3:bogus-identity", + effective_media_file_id: 8, + audio_tracks: [], + }), + }), + ); + await heldReplan; + }); + + await waitFor(() => expect(result.current.plan?.plan_id).toBe("plan:bogus-identity")); + expect(result.current.effectiveVirtualUri).toBeNull(); expect(result.current.planAudioTracks).toEqual([]); unmount(); }); diff --git a/web/src/player/hooks/usePlaybackSession.ts b/web/src/player/hooks/usePlaybackSession.ts index 1c9ffaff63..743cd47524 100644 --- a/web/src/player/hooks/usePlaybackSession.ts +++ b/web/src/player/hooks/usePlaybackSession.ts @@ -169,6 +169,13 @@ interface SourceIdentity { interface DeferredPushBase { /** Local, monotonically increasing arrival order. */ seq: number; + /** + * The source-identity transition clock at arrival. A later external transition + * (a poll fold, or an adopted plan) raises the clock above this; a value still + * at or below it is the state the entry already saw at capture and is already + * represented by its `outgoing` baseline. + */ + arrivalTransitionClock: number; /** * Whether the wire shape named any source at all. An identity-less push * cannot be tied to an adoption, so it may not cross a session change the @@ -180,6 +187,13 @@ interface DeferredPushBase { plan: PlanV3 | null; /** The live source identity in force at arrival (rotations included). */ outgoing: SourceIdentity; + /** + * True when `outgoing` was itself produced by an earlier deferred source + * commit rather than by the settled plan or an external poll. Such a baseline + * is just the transport's previous move captured from the same queue, so it is + * not authority a newer source commit must answer to. + */ + outgoingFromDeferredSource: boolean; /** The adoption generation (`loadSequence`) that owned the session. */ generation: number; /** The session id at arrival. */ @@ -279,38 +293,117 @@ function identityNamesSameSource(a: SourceIdentity, b: SourceIdentity): boolean * Whether it was replaced at all is decided first, because a refused plan is * still the pre-rotation plan and its URI would otherwise veto a legitimate * rotation. Against a replaced plan that names a candidate URI, that URI is the - * only candidate it vouches for, and an exact match is retained however it - * lines up with the outgoing source (a replacement can retain the effective - * candidate). + * only candidate it vouches for, and an exact match is retained even when the + * live outgoing source names a different candidate (a replacement can retain + * the effective candidate) — so a replaced URI-ful winner is decided by exact + * identity *before* any ambiguity guard, or an outgoing candidate the session + * was leaving would veto the plan the server just selected. + * + * When the replaced plan is URI-less (the v2 wire omits the candidate) file + * equality is still the only key the wire offers, so a candidate-bearing push + * naming the plan's own file is admitted — the payload routinely carries the + * candidate the plan omits, and refusing it would leave the track menus stale + * for the whole session. A URI-only push (no file id) cannot be tied to the + * settled file by itself; it is admitted only when the arrival-time live source + * was already on the winner's file, which corroborates that the candidate + * belongs to the adoption the flush kept. Otherwise the push may be inventory + * for the file the session was leaving, and folding it would inherit an + * unrelated replacement file. + * + * A same-file source the plan cannot resolve stays refused whenever a concrete + * candidate is already current on that file: a file-only push under any + * concrete candidate, and a concrete push whose URI disagrees with one. That is + * checked against two baselines — `outgoing`, the live source captured when the + * push arrived, and `applied`, the identity the menus carry when the flush runs + * — because a poll bypasses the adoption barrier and can fold a concrete + * candidate after the push was queued, including when the plan is later refused + * rather than replaced. + * + * A URI-bearing source commit is the transport's own move, so arrival order + * decides among the queue's own commits: when such a commit supersedes a + * previous deferred source commit, `outgoingSuperseded` drops that queue-produced + * `outgoing` baseline. It does not license the commit to overwrite external + * state: a candidate a poll applied after the entry was queued is newer than the + * commit and stays authoritative. `appliedPollNewer` carries that candidate — the + * menus may have moved on since — and is checked as an extra baseline. A + * candidate the menus carried before the entry was queued is either the arrival + * state already captured by `outgoing`, or a sibling the arrival collision + * already rejects, so it is not re-checked. * - * When the replaced plan is URI-less (the v2 wire omits the candidate) it - * cannot prove which of two same-file candidates it owns, so file equality is - * never proof. A `source_committed` naming a different file than the live - * outgoing source is the only rotation that can be adopted, and it must land on - * the plan's own file. When the plan was refused rather than replaced, a push - * that still names the live source is the same source and is kept — the source - * never moved — while a source commit naming another file is a rotation the - * transport already made and is kept too. + * When the plan was refused rather than replaced, the live source is the + * authority and `allowRotation` decides whether a cross-file source commit may + * move it: a push that still names the live source is kept — the source never + * moved — while a source commit naming another file is a rotation the transport + * already made and is kept too. */ function deferredIdentityIsAdmissible( identity: SourceIdentity, plan: PlanV3, arrivalPlan: PlanV3 | null, outgoing: SourceIdentity, + applied: SourceIdentity, allowRotation: boolean, + /** + * The concrete candidate a poll applied *after* this entry was queued, or null + * when none did. A URI-bearing source commit is judged by arrival order against + * its own queue, but this state is newer than the commit and still vetoes it. + */ + appliedPollNewer: SourceIdentity | null, + outgoingSuperseded = false, ): boolean { if (!identityNamesSource(identity)) return false; // Whether the settled plan is a different adoption or the byte-for-byte - // arrival plan. This must be decided before the winner's URI is consulted: a - // refused plan still names the pre-rotation source, so its URI must not veto - // a rotation the transport already committed to. - if (!planWasReplaced(arrivalPlan, plan)) { - // The plan was refused, not replaced. The live source is authoritative: a - // push that names it exactly is a same-source update, and a source commit - // may also prove the transport rotated to another file. The refused plan's - // own URI describes the source before that rotation, so it is not applied - // as authority here. + // arrival plan. A refused plan still names the pre-rotation source, so its + // URI must not veto a rotation the transport already committed to. + const replaced = planWasReplaced(arrivalPlan, plan); + + // The winner was replaced and names a concrete candidate: its URI is the only + // candidate it vouches for, and an exact match wins outright. This must be + // decided before the collision guards below, which would otherwise let the + // outgoing candidate the session was leaving veto the explicitly winning plan. + if (replaced && plan.effective_virtual_uri != null) { + return identityMatchesPlan(identity, plan); + } + + // A concrete candidate already current on the push's file is authority the + // push cannot outweigh: a file-only push carries no candidate, and a push + // naming a different candidate is a sibling collision. A cross-file baseline + // is a different source and does not apply. + // + // A URI-bearing source commit is the transport's own move, so arrival order + // decides among the queue's own commits: `outgoingSuperseded` drops the + // queue-produced `outgoing` baseline of an earlier deferred source commit. The + // live `outgoing` baseline is still consulted — it is the source the transport + // was on at arrival, and a differing concrete candidate on the same file is a + // real collision. What arrival order does *not* license is overwriting a + // candidate a poll folded *after* the entry was queued: that `appliedPollNewer` + // state is newer than the commit, so it is added as a veto even for a commit. + // An older applied candidate predates the entry and is either the arrival state + // already captured by `outgoing` or a sibling the arrival collision already + // rejects, so it is not re-checked. + const judgedByArrival = allowRotation && identity.uri != null; + const baselines: SourceIdentity[] = [outgoing]; + if (!judgedByArrival) baselines.push(applied); + else if (appliedPollNewer != null) baselines.push(appliedPollNewer); + for (const baseline of baselines) { + if (outgoingSuperseded && baseline === outgoing) continue; + if (baseline.uri == null) continue; + // An unknown file on either side is treated as the same file: neither can + // prove a different source, so a differing concrete URI is a collision. + const sameFile = + baseline.fileId == null || identity.fileId == null || baseline.fileId === identity.fileId; + if (!sameFile) continue; + if (identity.uri == null || identity.uri !== baseline.uri) return false; + } + + // The plan was refused, not replaced. `allowRotation` applies to this branch + // only; the replaced-plan branches decide from the winner. + if (!replaced) { + // The live source is authoritative: a push that names it exactly is a + // same-source update, and a source commit may also prove the transport + // rotated to another file. The refused plan's own URI describes the source + // before that rotation, so it is not applied as authority here. if (identityNamesSameSource(identity, outgoing)) return true; return ( allowRotation && @@ -319,20 +412,17 @@ function deferredIdentityIsAdmissible( ); } - // The plan was replaced and won. When it names a candidate URI, that URI is - // the only candidate it vouches for, and an exact match is retained however - // it lines up with the outgoing source (a replacement can retain the - // effective candidate). - if (plan.effective_virtual_uri != null) return identityMatchesPlan(identity, plan); - // A URI-only push cannot be tied to the URI-less plan's file. - if (identity.fileId == null) return false; - - // The URI-less winner cannot prove same-file ownership: a different file - // than the live outgoing is the only evidence that can outweigh the plan's - // candidate silence, and only a source commit carries it. - if (!allowRotation) return false; - if (outgoing.fileId != null && identity.fileId === outgoing.fileId) return false; - return identity.fileId === plan.effective_media_file_id; + // The winner is URI-less. The baseline loop above already refused a same-file + // collision, so a push naming the plan's own file is admitted; a push naming + // another file is a rotation the URI-less plan cannot vouch for. + if (identity.fileId != null) { + return identity.fileId === plan.effective_media_file_id; + } + // A URI-only push offers no file key. Admit it only when the arrival-time live + // source was on the winner's file, which ties the candidate to the settled + // adoption; otherwise it may belong to the file being replaced and must wait + // for a current-source refresh instead of inheriting the replacement's id. + return outgoing.fileId != null && outgoing.fileId === plan.effective_media_file_id; } /** A start body plus the optional force-relink flag the retry path adds. */ @@ -760,6 +850,25 @@ export function usePlaybackSession( fileId: state.mediaFileId, uri: state.effectiveVirtualUri, }); + // The identity the most recent deferred source commit moved the live mirror + // to, when nothing external has transitioned it since. A later deferred source + // commit must not treat its own predecessor from the same queue as an + // authority to answer to, while a poll fold is external and stays + // authoritative; `transitionSourceIdentity` clears this on every such fold. + const lastDeferredSourceIdentityRef = useRef(null); + // A monotonic clock raised on every *external* source-identity transition (a + // poll fold or an adopted plan), never by the deferred queue's own replay. A + // deferred push records the clock at arrival; a later poll fold raises it, so + // the flush can tell a candidate folded after the entry was queued from one + // the menus already carried at capture. See `captureDeferredPush`. + const identityTransitionClockRef = useRef(0); + // The live candidate a poll folded on the last external transition, with the + // clock reading of that transition. Between an entry's arrival and its flush a + // poll can move the menus to a candidate, and the winning plan's own adoption + // can then overwrite the menu mirror before the flush runs. Remembering the + // poll's candidate lets the flush veto a stale commit that would otherwise + // overwrite it; the clock says whether the fold was newer than the entry. + const lastPollAppliedRef = useRef<{ clock: number; identity: SourceIdentity } | null>(null); // The identity the menus render, mirrored outside the state updater. A poll // or rotation can fold before React has processed an earlier push in the same // tick, so the next transition is computed against this synchronous mirror; @@ -921,23 +1030,38 @@ export function usePlaybackSession( menuIdentityRef.current = { fileId, uri }; // A fold of the menu identity is also a commit of the live source, so the // outgoing baseline for the next rotation follows it. A deferred push - // moves the live mirror separately in `captureDeferredPush`. + // moves the live mirror separately in `captureDeferredPush`. Any external + // transition supersedes the last deferred source commit, so it is no + // longer a baseline a later queued commit may ignore. liveSourceIdentityRef.current = { fileId, uri }; + lastDeferredSourceIdentityRef.current = null; + identityTransitionClockRef.current += 1; return changed; }, [], ); + /** + * Records a candidate a catalog poll folded, with the transition clock of the + * fold. `deferredIdentityIsAdmissible` consults it so a source commit queued + * before the poll cannot overwrite the poll's newer candidate, even after the + * winning plan's own adoption has since overwritten the menu mirror. + */ + const recordPollApplied = useCallback((identity: SourceIdentity) => { + lastPollAppliedRef.current = { clock: identityTransitionClockRef.current, identity }; + }, []); + /** * The revision bookkeeping scope for the current session generation. * * A revision is the server's opaque digest of the whole inventory *and* the - * effective source it describes, so within one session a redelivery is the - * same payload and folding it twice is wrong — including a redelivery of a - * revision the flush refused. The scope is keyed by the load generation and - * resets only when that moves; an identity rotation must not clear it, or a - * redelivered file-only rejection would fold once a later source moved the - * menu. + * effective source it describes, so within one session redelivering a revision + * that was already folded is the same payload and folding it twice is wrong. + * Only genuinely applied revisions are recorded; a revision the flush refused + * is deliberately left unrecorded, so a redelivery that the plan (or a later + * rotation) vouches for still folds. The scope is keyed by the load generation + * and resets only when that moves; an identity rotation must not clear it, or a + * revision folded before the rotation would fold again once the source moved. */ const revisionScopeRef = useRef<{ generation: number; revisions: Set }>({ generation: -1, @@ -2179,7 +2303,14 @@ export function usePlaybackSession( // the live source changed, so the menu must follow it. Commit the // transition synchronously and outside the updater; the revision scope is // generation-keyed and is deliberately not cleared here. - if (identityChanged) transitionSourceIdentity(nextFileId, nextUri); + if (identityChanged) { + transitionSourceIdentity(nextFileId, nextUri); + // A catalog poll is external state. Record the candidate it moved to, + // with the transition clock, so a deferred source commit queued before + // this fold cannot overwrite it even if a later plan adoption replaces + // the menu mirror before the flush runs. + recordPollApplied({ fileId: nextFileId, uri: nextUri }); + } setState((current) => { if (identityChanged) { return { @@ -2209,7 +2340,7 @@ export function usePlaybackSession( }; }); }, - [transitionSourceIdentity], + [recordPollApplied, transitionSourceIdentity], ); const foldCommittedSource = useCallback( @@ -2271,12 +2402,18 @@ export function usePlaybackSession( */ const captureDeferredPush = useCallback((input: DeferredPushInput) => { const live = liveSourceIdentityRef.current; + const lastDeferredSource = lastDeferredSourceIdentityRef.current; const base: DeferredPushBase = { seq: deferredPushSeqRef.current++, + arrivalTransitionClock: identityTransitionClockRef.current, hasIdentity: identityNamesSource(input.identity), identity: input.identity, plan: planRef.current, outgoing: { fileId: live.fileId, uri: live.uri }, + outgoingFromDeferredSource: + lastDeferredSource != null && + lastDeferredSource.fileId === live.fileId && + lastDeferredSource.uri === live.uri, generation: loadSequenceRef.current, sessionId: sessionIdRef.current, }; @@ -2298,6 +2435,7 @@ export function usePlaybackSession( fileId: input.identity.fileId ?? live.fileId, uri: input.identity.uri ?? live.uri, }; + lastDeferredSourceIdentityRef.current = liveSourceIdentityRef.current; } }, []); @@ -2494,6 +2632,19 @@ export function usePlaybackSession( ) { return; } + // A file-only push cannot be tied to a candidate. When the live source + // already names a concrete candidate on that same file, the file alone is + // not proof of ownership and must not re-key the menus under that + // candidate. This mirrors `deferredIdentityIsAdmissible` so a refused + // file-only revision cannot slip back in through the direct path. + if ( + payload.effective_media_file_id != null && + payload.effective_virtual_uri == null && + liveIdentity.uri != null && + payload.effective_media_file_id === liveIdentity.fileId + ) { + return; + } foldInventoryUpdate(payload); }, [captureDeferredPush, foldInventoryUpdate], @@ -2508,8 +2659,10 @@ export function usePlaybackSession( * the replacement plan lands rather than being lost. The queue is replayed in * arrival order. The settled plan is the adoption that won: a push captured * under another adoption generation or session is dropped, and a push whose - * identity the plan cannot vouch for is dropped. A stale inventory revision - * still marks itself folded so a later re-delivery does not re-open the menus. + * identity the plan cannot vouch for is dropped. A refused inventory revision + * is left unrecorded, so a redelivery that the plan (or a later rotation) does + * vouch for still folds; a genuinely applied revision is recorded by the fold + * itself, so duplicate deliveries remain no-ops. */ const flushDeferredPushes = useCallback(() => { // A start/replan owns the session while its adoption is in flight, and a @@ -2528,20 +2681,25 @@ export function usePlaybackSession( const queue = deferredPushesRef.current.slice().sort((a, b) => a.seq - b.seq); if (queue.length === 0) return; deferredPushesRef.current = []; - const markRevisionFolded = (entry: DeferredPush) => { - // The settled plan is authoritative and its own inventory is on screen; - // mark a stale revision as folded so a re-delivery is a no-op. Every - // rejection path runs this so bookkeeping does not depend on which - // identity check happened to discard the entry. Recording it here does - // not fold the push; it only remembers that this generation has handled - // the revision, so a redelivered file-only rejection cannot fold later. - if (entry.kind === "inventory" && entry.payload.inventory_revision != null) { - inventoryRevisionRef.current = entry.payload.inventory_revision; - syncRevisionScope().revisions.add(entry.payload.inventory_revision); - } + // The identity the menus carry when the flush starts. It captures movements + // an outside push made while these entries waited (a poll folds without the + // adoption barrier); it is snapshotted once so the entries replayed below do + // not become a veto baseline for each other, which would invert arrival + // order when two concrete candidates are queued. + const appliedIdentity = menuIdentityRef.current; + const handleRefusedPush = (entry: DeferredPush) => { // A deferred source commit had moved the live baseline when it was // captured. If nothing later moved it on, put it back on the winning plan // so a source the flush just refused is not left recorded as live. + // + // A refused inventory revision is deliberately *not* recorded as folded: + // the refusal is a statement about this flush's settled plan, not about + // the revision, and the identity can move later in the same generation + // (a rotation onto a candidate the revision names). Recording it here + // would swallow the redelivery that finally matches, leaving the menus + // stale for the rest of the session. Only a genuinely applied revision is + // recorded, by `foldInventoryUpdate`, so duplicates of applied revisions + // stay suppressed while a refused one can still apply once it matches. if ( entry.kind === "source" && identityNamesSameSource(liveSourceIdentityRef.current, entry.identity) @@ -2550,6 +2708,7 @@ export function usePlaybackSession( fileId: plan.effective_media_file_id, uri: plan.effective_virtual_uri ?? null, }; + lastDeferredSourceIdentityRef.current = null; } }; for (const entry of queue) { @@ -2568,20 +2727,34 @@ export function usePlaybackSession( ) { continue; } - // The settled plan is the only authority. Its own identity wins over the - // outgoing one (a replacement may retain the effective candidate); when - // it cannot prove which same-file candidate it owns, the outgoing - // identity is never folded and file equality is not proof. + // The settled plan is the primary authority. When it names a candidate + // URI, that URI wins over the outgoing one (a replacement may retain the + // effective candidate) and an exact match is kept. When it is URI-less, + // the push's own file is the key, except for a same-file source the plan + // cannot resolve. `appliedIdentity` is the menu identity at flush start, so + // a push deferred before a poll folded a concrete candidate cannot + // overwrite it. A URI-bearing source commit is judged by arrival order, but + // only against state older than it: a poll that folded a candidate after + // this entry was queued is external and newer and still vetoes the commit. + // See deferredIdentityIsAdmissible. + const pollApplied = lastPollAppliedRef.current; + const appliedPollNewer = + pollApplied != null && pollApplied.clock > entry.arrivalTransitionClock + ? pollApplied.identity + : null; if ( !deferredIdentityIsAdmissible( entry.identity, plan, entry.plan, entry.outgoing, + appliedIdentity, entry.kind === "source", + appliedPollNewer, + entry.kind === "source" && entry.outgoingFromDeferredSource, ) ) { - markRevisionFolded(entry); + handleRefusedPush(entry); continue; } if (entry.kind === "source") { @@ -2590,7 +2763,7 @@ export function usePlaybackSession( foldInventoryUpdate(entry.payload); } } - }, [foldCommittedSource, foldInventoryUpdate, syncRevisionScope]); + }, [foldCommittedSource, foldInventoryUpdate]); flushDeferredPushesRef.current = flushDeferredPushes; const updatePlaybackState = useCallback((positionSeconds: number, playing: boolean) => {