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.
This commit is contained in:
Zach Clendenen
2026-09-25 12:53:04 -05:00
parent a18c8493c1
commit c9f1d5e33b
8 changed files with 281 additions and 27 deletions
+3
View File
@@ -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):
+37 -11
View File
@@ -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.
+10 -1
View File
@@ -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.
+150 -2
View File
@@ -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
):
@@ -15,5 +15,6 @@ export type StreamingContainerSchema = {
supports_memory_cards: boolean;
supports_save_picker: boolean;
supports_live_states: boolean;
import_kinds: Array<'save' | 'state'>;
};
+1
View File
@@ -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,
+62 -6
View File
@@ -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<VueWrapper> {
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]);
});
});
+17 -7
View File
@@ -303,7 +303,7 @@ const selectedState = ref<UserStateSchema | null>(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<SaveSchema[]>(() => {
const allSaves = computed<SaveSchema[]>(() => {
if (!rom.value) return [];
return [...(rom.value.user_saves ?? [])].sort(
(a, b) =>
@@ -316,11 +316,18 @@ const pickableSaves = computed<SaveSchema[]>(() => {
const nativeRestorableSaves = computed<SaveSchema[]>(() => {
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<SaveSchema[]>(() =>
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<SaveSchema | null>(
@@ -353,11 +360,14 @@ const nativeStreamStates = computed<UserStateSchema[]>(() => {
);
});
// 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<UserStateSchema[]>(() =>
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<UserStateSchema[]>(() => {
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