From c9f1d5e33be54213a7105edfcb45da61e2764d95 Mon Sep 17 00:00:00 2001 From: Zach Clendenen Date: Fri, 25 Sep 2026 12:53:04 -0500 Subject: [PATCH] feat(streaming): offer only the foreign picks a broker imports, and honour the picked state over an older save's exit state The streaming config now ships import_kinds per container, from the broker's cached import-spec, so the picker hides foreign saves and states a broker would refuse instead of failing them after the launch starts. On DuckStation and RPCS3, a save pick other than the newest archive no longer lets that archive's own exit state stand in for the picked state: the state rides the import archive, or the resume is reported lost where the broker cannot import it. --- backend/endpoints/responses/streaming.py | 3 + backend/endpoints/streaming.py | 48 ++++-- backend/handler/streaming/webstation.py | 11 +- backend/tests/endpoints/test_streaming.py | 152 +++++++++++++++++- .../models/StreamingContainerSchema.ts | 1 + frontend/src/stores/streaming.test.ts | 1 + frontend/src/v2/views/Player/Stream.test.ts | 68 +++++++- frontend/src/v2/views/Player/Stream.vue | 24 ++- 8 files changed, 281 insertions(+), 27 deletions(-) diff --git a/backend/endpoints/responses/streaming.py b/backend/endpoints/responses/streaming.py index 8b072ff5c..5e6c09def 100644 --- a/backend/endpoints/responses/streaming.py +++ b/backend/endpoints/responses/streaming.py @@ -31,6 +31,9 @@ class StreamingContainerSchema(BaseModel): supports_memory_cards: bool supports_save_picker: bool supports_live_states: bool + import_kinds: list[Literal["save", "state"]] + """Which foreign-emulator picks the broker declares it can import, empty + when it declares none or cannot be asked.""" class StreamingConfigSchema(BaseModel): diff --git a/backend/endpoints/streaming.py b/backend/endpoints/streaming.py index 376b345e9..42b0aa3b9 100644 --- a/backend/endpoints/streaming.py +++ b/backend/endpoints/streaming.py @@ -293,12 +293,22 @@ async def get_config(request: Request) -> StreamingConfigSchema: """Return streaming configuration to the frontend""" # One row per platform: a pool is a backend concern, the frontend picks a # platform and the claim decides which container serves it. + # The record carries the platform's label and capabilities, so a + # platform hidden from this caller must not be listed here either. + visible = [ + c + for c in first_claim_targets() + if access.platform_is_visible(request, c.platform) + ] + # Concurrently, so one unreachable broker costs one timeout, not one each. + specs = await asyncio.gather( + *( + asyncio.to_thread(webstation.import_spec, c, c.emulator, c.platform) + for c in visible + ) + ) safe_containers: list[StreamingContainerSchema] = [] - for c in first_claim_targets(): - # The record carries the platform's label and capabilities, so a - # platform hidden from this caller must not be listed here either. - if not access.platform_is_visible(request, c.platform): - continue + for c, spec in zip(visible, specs, strict=True): safe_containers.append( StreamingContainerSchema( platform=c.platform, @@ -319,6 +329,8 @@ async def get_config(request: Request) -> StreamingConfigSchema: # Whether the in-game Save and Load buttons reach a broker that # honours them, which an exit-state emulator's does not. supports_live_states=c.supports_live_states, + # So the picker only offers a foreign pick the claim will take. + import_kinds=spec.pickable_kinds() if spec is not None else [], ) ) @@ -842,9 +854,10 @@ async def claim_session( container = await _win_container(request, candidates, session, platform) session_key = container.key - # A save archive only carries the newest exit state, so an older native - # pick on an archive-resume container (DuckStation, RPCS3) needs importing. - older_exit_state = False + # On an archive-resume container (DuckStation, RPCS3) the save archive + # carries its own exit state, and only the newest archive carries the newest + # capture, so any other state and save pairing needs the state imported. + state_off_archive = False if ( resume_state is not None and not resume_foreign @@ -853,12 +866,22 @@ async def claim_session( newest_states = await asyncio.to_thread( states.user_states_for_emulator, request.user.id, rom.id, container.emulator ) - older_exit_state = not newest_states or newest_states[0].id != resume_state.id + state_off_archive = not newest_states or newest_states[0].id != resume_state.id + if not state_off_archive and picked_save is not None: + newest_save = None + if not save_foreign: + newest_save = await asyncio.to_thread( + saves.newest_restorable, + request.user.id, + rom.id, + container.emulator, + ) + state_off_archive = newest_save is None or newest_save.id != picked_save.id # The pre-win checks asked the pool's reference; the won container's own # import-spec is what decides, and one answer covers both picks. spec = None - if resume_foreign or older_exit_state or save_foreign: + if resume_foreign or state_off_archive or save_foreign: spec = await asyncio.to_thread( webstation.import_spec, container, container.emulator, container.platform ) @@ -876,7 +899,7 @@ async def claim_session( status_code=400, detail="This container cannot restore the picked save" ) resume_via_import = ( - (resume_foreign or older_exit_state) + (resume_foreign or state_off_archive) and spec is not None and spec.accepts("state") and spec.state_channel == "archive" @@ -938,6 +961,9 @@ async def claim_session( resume_import: launch.ResumeImport = "none" if resume_via_import: resume_import = "imported" if state_imported else "lost" + elif state_off_archive: + # The archive would resume a state the player did not pick. + resume_import = "lost" # Detached because an activate blocks through pkg and archive extraction, # minutes on a large title, which no player can cancel out of. diff --git a/backend/handler/streaming/webstation.py b/backend/handler/streaming/webstation.py index 0c7a4af87..adb97674e 100644 --- a/backend/handler/streaming/webstation.py +++ b/backend/handler/streaming/webstation.py @@ -26,7 +26,7 @@ import math import time import urllib.error from dataclasses import dataclass -from typing import Any +from typing import Any, Literal from urllib.parse import quote from config import STREAMING_LAUNCH_TIMEOUT, STREAMING_SAVE_TIMEOUT @@ -54,6 +54,15 @@ class ImportSpec: return self.state_slot return None + def pickable_kinds(self) -> list[Literal["save", "state"]]: + """The launch picks a foreign save or state can resume through here.""" + kinds: list[Literal["save", "state"]] = [] + if self.accepts("save"): + kinds.append("save") + if self.resume_slot() is not None: + kinds.append("state") + return kinds + # Per container, so one broker's answer never speaks for another. A 404 or 422 # holds for the worker's life, an answer for one claim's checks, a failure not at all. diff --git a/backend/tests/endpoints/test_streaming.py b/backend/tests/endpoints/test_streaming.py index e2b0c832b..7098cc1b8 100644 --- a/backend/tests/endpoints/test_streaming.py +++ b/backend/tests/endpoints/test_streaming.py @@ -509,6 +509,47 @@ def test_get_config_reports_save_picker_support(client, access_token, rom: Rom): assert supported["gba"] is False +def test_get_config_reports_the_foreign_picks_each_broker_imports( + client, access_token, rom: Rom +): + """The picker only offers a foreign pick the claim would take, so a broker + that declares no imports, or cannot be asked, reports none.""" + importing = {**_webstation_for(rom), "emulator": "duckstation"} + silent = {**_webstation_for(rom), "platform": "ps2", "emulator": "pcsx2"} + spec = webstation.ImportSpec( + kinds=frozenset({"save", "state"}), state_channel="archive", state_slot=0 + ) + + def answer(container, _emulator, _platform): + return spec if container.platform == rom.platform_slug else None + + with _streaming(importing, silent): + with patch("handler.streaming.webstation.import_spec", side_effect=answer): + response = client.get("/api/streaming/config", headers=_auth(access_token)) + assert response.status_code == 200 + kinds = {c["platform"]: c["import_kinds"] for c in response.json()["containers"]} + assert kinds[rom.platform_slug] == ["save", "state"] + assert kinds["ps2"] == [] + + +@pytest.mark.parametrize( + ("kinds", "channel", "slot", "expected"), + [ + ({"save", "state"}, "archive", 0, ["save", "state"]), + ({"state"}, "none", 0, []), + ({"state"}, "push", None, []), + ({"save", "memcard"}, "none", None, ["save"]), + ], +) +def test_import_spec_pickable_kinds_match_what_the_claim_accepts( + kinds, channel, slot, expected +): + spec = webstation.ImportSpec( + kinds=frozenset(kinds), state_channel=channel, state_slot=slot + ) + assert spec.pickable_kinds() == expected + + def test_clears_stale_saves_overrides_the_emulator_default(client, access_token, rom): """The default mirrors a flag that lives in the broker's repo, so an operator on a fork or a newer broker can say what theirs actually does.""" @@ -7192,7 +7233,8 @@ def test_an_older_exit_state_stays_off_the_import_path_without_broker_support( client, access_token, rom: Rom, admin_user: User ): """A broker that predates imports would never place an `.import/` member, - so the archive and its own exit state go out untouched instead.""" + so the archive goes out untouched and the resume is reported lost rather + than claiming the archive's own exit state was the pick.""" older = db_state_handler.add_state( _state_for( rom, @@ -7224,10 +7266,116 @@ def test_an_older_exit_state_stays_off_the_import_path_without_broker_support( r = _claim(client, access_token, rom.id, state_id=older.id) assert r.status_code == 202 hydrate_import.assert_not_called() - assert run_launch_mock.call_args.kwargs["resume_import"] == "none" + assert run_launch_mock.call_args.kwargs["resume_import"] == "lost" assert run_launch_mock.call_args.kwargs["archive_path"] == "/romm/saves/archive.zip" +def _duckstation_pairing(rom: Rom, user: User) -> tuple[Save, Save, State]: + """An older and a newer DuckStation archive, plus the newest capture.""" + older, newer = ( + db_save_handler.add_save( + _save_for( + rom, user, f"Game [duckstation {tag}].saves.zip", "duckstation", tag + ) + ) + for tag in ("a", "b") + ) + state = db_state_handler.add_state( + _state_for(rom, user, "SLUS-00594_resume.sav", "duckstation") + ) + return older, newer, state + + +def _clearing_duckstation(rom: Rom) -> dict: + return { + **_webstation_for(rom), + "emulator": "duckstation", + "clears_stale_saves": True, + } + + +def test_an_older_save_with_the_newest_state_imports_the_picked_state( + client, access_token, rom: Rom, admin_user: User +): + """The older archive only carries its own exit state, so the picked state + rides the import archive on top of it instead of being silently swapped.""" + older, _, state = _duckstation_pairing(rom, admin_user) + activate = MagicMock(return_value={"url": "/room/x"}) + upload = MagicMock(return_value="rom-1.zip") + spec = webstation.ImportSpec( + kinds=frozenset({"state"}), state_channel="archive", state_slot=0 + ) + with _streaming(_clearing_duckstation(rom)): + with ( + patch("handler.streaming.webstation.activate", activate), + patch("handler.streaming.webstation.import_spec", return_value=spec), + patch( + "handler.streaming.imports.fs_asset_handler.read_file", + new=AsyncMock(side_effect=lambda path: path.encode()), + ), + patch("handler.streaming.imports.webstation.upload_archive", upload), + patch("handler.streaming.background.spawn_sync_task"), + ): + with _pushes() as sent: + r = _claim( + client, access_token, rom.id, state_id=state.id, save_id=older.id + ) + assert r.status_code == 202 + with zipfile.ZipFile(io.BytesIO(upload.call_args.args[2])) as zf: + assert any(name.startswith(".import/state/") for name in zf.namelist()) + assert _launch_ready(sent)["resume"] is True + + +def test_an_older_save_with_the_newest_state_reports_the_resume_lost_without_imports( + client, access_token, rom: Rom, admin_user: User +): + """Without an import channel the older archive's own exit state would load, + so the launch must not report the picked state as resumed.""" + older, _, state = _duckstation_pairing(rom, admin_user) + run_launch_mock = AsyncMock() + with _streaming(_clearing_duckstation(rom)): + with ( + patch("handler.streaming.webstation.import_spec", return_value=None), + patch( + "handler.streaming.saves.hydrate_saves_to_webstation", + new=AsyncMock(return_value="/romm/saves/archive.zip"), + ), + patch("handler.streaming.launch.run_launch", run_launch_mock), + ): + r = _claim( + client, access_token, rom.id, state_id=state.id, save_id=older.id + ) + assert r.status_code == 202 + assert run_launch_mock.call_args.kwargs["resume_import"] == "lost" + + +def test_the_newest_save_with_the_newest_state_resumes_from_the_archive( + client, access_token, rom: Rom, admin_user: User +): + """Naming the newest archive explicitly is the default pairing, so the + archive's own exit state is the pick and nothing is imported.""" + _, newer, state = _duckstation_pairing(rom, admin_user) + hydrate_import = AsyncMock() + run_launch_mock = AsyncMock() + with _streaming(_clearing_duckstation(rom)): + with ( + patch("handler.streaming.webstation.import_spec") as spec, + patch("handler.streaming.imports.hydrate_import_archive", hydrate_import), + patch( + "handler.streaming.saves.hydrate_saves_to_webstation", + new=AsyncMock(return_value="/romm/saves/archive.zip"), + ), + patch("handler.streaming.launch.run_launch", run_launch_mock), + ): + r = _claim( + client, access_token, rom.id, state_id=state.id, save_id=newer.id + ) + assert r.status_code == 202 + spec.assert_not_called() + hydrate_import.assert_not_called() + assert run_launch_mock.call_args.kwargs["resume_import"] == "none" + + def test_an_exit_state_resume_with_no_archive_reports_the_resume_lost( client, access_token, rom: Rom, admin_user: User ): diff --git a/frontend/src/__generated__/models/StreamingContainerSchema.ts b/frontend/src/__generated__/models/StreamingContainerSchema.ts index 0121f48bd..b5715953d 100644 --- a/frontend/src/__generated__/models/StreamingContainerSchema.ts +++ b/frontend/src/__generated__/models/StreamingContainerSchema.ts @@ -15,5 +15,6 @@ export type StreamingContainerSchema = { supports_memory_cards: boolean; supports_save_picker: boolean; supports_live_states: boolean; + import_kinds: Array<'save' | 'state'>; }; diff --git a/frontend/src/stores/streaming.test.ts b/frontend/src/stores/streaming.test.ts index 604aeae14..b4b2cbaa2 100644 --- a/frontend/src/stores/streaming.test.ts +++ b/frontend/src/stores/streaming.test.ts @@ -28,6 +28,7 @@ describe("platformCapabilities disc flags", () => { supports_memory_cards: false, supports_save_picker: false, supports_live_states: true, + import_kinds: [], capabilities: { max_slots: 0, has_autosave: true, diff --git a/frontend/src/v2/views/Player/Stream.test.ts b/frontend/src/v2/views/Player/Stream.test.ts index 052f4d5a1..077dd3bb1 100644 --- a/frontend/src/v2/views/Player/Stream.test.ts +++ b/frontend/src/v2/views/Player/Stream.test.ts @@ -2,9 +2,9 @@ import { RBtn } from "@v2/lib"; import { flushPromises, mount, type VueWrapper } from "@vue/test-utils"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { defineComponent, type Slots, type VNodeChild } from "vue"; -import type { SaveSchema } from "@/__generated__"; +import type { SaveSchema, StateSchema } from "@/__generated__"; import type { DetailedRom } from "@/stores/roms"; -import { saveFixture } from "@/utils/assets.fixtures"; +import { saveFixture, stateFixture } from "@/utils/assets.fixtures"; import AssetPreview from "@/v2/components/Player/AssetPreview.vue"; import SaveDataPanel from "@/v2/components/Player/SaveDataPanel.vue"; import AssetList from "@/v2/components/shared/AssetList.vue"; @@ -211,7 +211,7 @@ const ARCHIVES = [ save(1, "Pool [retroarch 2026-09-14 00-20-28].saves.zip"), ]; -function romWith(saves: SaveSchema[]): DetailedRom { +function romWith(saves: SaveSchema[], states: StateSchema[] = []): DetailedRom { return { id: 3, name: "Archer Maclean's 3D Pool (USA)", @@ -221,7 +221,7 @@ function romWith(saves: SaveSchema[]): DetailedRom { fs_name: "Archer Maclean's 3D Pool (USA).gba", files: [], user_saves: saves, - all_user_states: [], + all_user_states: states, user_screenshots: [], metadatum: {}, } as unknown as DetailedRom; @@ -240,7 +240,9 @@ afterEach(() => { async function launch(opts: { picker: boolean; saves?: SaveSchema[]; + states?: StateSchema[]; liveStates?: boolean; + imports?: ("save" | "state")[]; }): Promise { mocks.container = { name: "WEBSTATION-DEV", @@ -248,10 +250,13 @@ async function launch(opts: { protocol: "webstation", supports_save_picker: opts.picker, supports_live_states: opts.liveStates ?? true, + import_kinds: opts.imports ?? [], supports_memory_cards: false, supports_multiplayer: false, }; - mocks.getRom.mockResolvedValue({ data: romWith(opts.saves ?? ARCHIVES) }); + mocks.getRom.mockResolvedValue({ + data: romWith(opts.saves ?? ARCHIVES, opts.states), + }); const wrapper = mount(Stream, { shallow: true, global: { @@ -347,10 +352,11 @@ describe("Stream save picker", () => { expect(mocks.claimSession.mock.calls[0][2]).toBe(1); }); - it("includes bare (non-archive) save files in the picker", async () => { + it("includes bare (non-archive) save files where the broker imports saves", async () => { const wrapper = await launch({ picker: true, saves: [save(9, "Pool.srm"), ...ARCHIVES], + imports: ["save"], }); expect( @@ -370,6 +376,7 @@ describe("Stream save picker", () => { }), ...ARCHIVES, ], + imports: ["save"], }); expect( @@ -378,6 +385,20 @@ describe("Stream save picker", () => { expect(saveList(wrapper)!.props("selectedId")).toBe(3); }); + it("hides another emulator's archives where the broker declares no save import", async () => { + const wrapper = await launch({ + picker: true, + saves: [ + save(9, "Pool [pcsx2 a].saves.zip", { emulator: "pcsx2" }), + ...ARCHIVES, + ], + }); + + expect( + (saveList(wrapper)!.props("assets") as SaveSchema[]).map((s) => s.id), + ).toEqual([3, 2, 1]); + }); + it("sends a foreign-emulator pick on the claim", async () => { const wrapper = await launch({ picker: true, @@ -389,6 +410,7 @@ describe("Stream save picker", () => { }), ...ARCHIVES, ], + imports: ["save"], }); await saveList(wrapper)!.vm.$emit( @@ -1055,3 +1077,37 @@ describe("Stream join", () => { expect(mocks.joinSession).toHaveBeenCalledWith("gba", undefined); }); }); + +describe("Stream state picker", () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.claimSession.mockResolvedValue(CLAIM); + }); + + const STATES = [ + stateFixture({ id: 5, emulator: "retroarch" }), + stateFixture({ id: 6, emulator: "duckstation" }), + ]; + + function pickableStateIds(wrapper: VueWrapper): number[] { + return ( + wrapper.vm as unknown as { pickableStates: StateSchema[] } + ).pickableStates.map((s) => s.id); + } + + it("offers only this emulator's states where the broker declares no state import", async () => { + const wrapper = await launch({ picker: false, states: STATES }); + + expect(pickableStateIds(wrapper)).toEqual([5]); + }); + + it("offers another emulator's states where the broker imports them", async () => { + const wrapper = await launch({ + picker: false, + states: STATES, + imports: ["state"], + }); + + expect(pickableStateIds(wrapper)).toEqual([5, 6]); + }); +}); diff --git a/frontend/src/v2/views/Player/Stream.vue b/frontend/src/v2/views/Player/Stream.vue index 9a3a5c9b9..7bcbf7a1a 100644 --- a/frontend/src/v2/views/Player/Stream.vue +++ b/frontend/src/v2/views/Player/Stream.vue @@ -303,7 +303,7 @@ const selectedState = ref(null); // Every save, whichever emulator wrote it, newest capture first: created_at, // since the updated_at user_saves arrives on moves with a rehash. -const pickableSaves = computed(() => { +const allSaves = computed(() => { if (!rom.value) return []; return [...(rom.value.user_saves ?? [])].sort( (a, b) => @@ -316,11 +316,18 @@ const pickableSaves = computed(() => { const nativeRestorableSaves = computed(() => { const emulator = emulatorKey(container.value?.emulator); if (!emulator) return []; - return pickableSaves.value.filter( + return allSaves.value.filter( (s) => emulatorKey(s.emulator) === emulator && s.file_name.endsWith(".zip"), ); }); +// A foreign save is only on offer where the broker declares it can import one. +const pickableSaves = computed(() => + container.value?.import_kinds.includes("save") + ? allSaves.value + : nativeRestorableSaves.value, +); + // The one the broker restores before boot when the claim names none: this // emulator's own, so the fallback restore always succeeds. const newestSave = computed( @@ -353,11 +360,14 @@ const nativeStreamStates = computed(() => { ); }); -// Every state regardless of which emulator wrote it, so the picker can offer -// a foreign pick; the broker routes one through the declared-import path. -const pickableStates = computed(() => - rom.value ? (rom.value.all_user_states ?? []) : [], -); +// Every state regardless of which emulator wrote it where the broker declares +// it can import one, which routes a foreign pick through the import path. +const pickableStates = computed(() => { + if (!rom.value) return []; + return container.value?.import_kinds.includes("state") + ? (rom.value.all_user_states ?? []) + : nativeStreamStates.value; +}); // Every capture is kept, so a heavy save-stater ends up with a history the // horizontal strip buries. Grid and list trade thumbnail size for how many