rommapp/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-10-12 02:21:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(streaming): put every container in a pool, claim from the first

Browse Source 
A platform's containers are now grouped into pools of interchangeable
members (same emulator, memory card sync, save picker and protocol), so a
lone container is a pool of one instead of a skipped non-member. Game claims,
status, heartbeats and joins still walk only the first pool, which matches
the containers they reached before.

The "not a pool member" warning fired on every lookup, one container at a
time. It now fires once per config and names every container a game claim
can never reach. Named lookups (desktop, admin release, named heartbeat)
already reach a container in any pool.
This commit is contained in:
Zach Clendenen
2026-09-17 11:42:18 -05:00
parent c62af75621
commit 9c1e9f4d75
4 changed files with 120 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/endpoints/streaming.py
+2 -2
View File
@@ -625,7 +625,7 @@ async def claim_session(
The ROM's filesystem path is derived server-side from its database row -
the client only supplies a ROM id, never a path.
Returns 404 if the ROM doesn't exist or no container serves its platform.
Returns 409 if every container serving the platform is occupied.
Returns 409 if every container in the platform's first pool is occupied.
Returns 428 if the container's pre-existing memory card needs a decision.
"""
rom = db_rom_handler.get_rom(req.rom_id)
@@ -926,7 +926,7 @@ async def heartbeat_session(
making rather than watching a dead stream.
`container` names the claim to refresh, needed for a desktop on a container
the platform's pool leaves out.
outside the platform's first pool.
"""
user_id = request.user.id
# A named claim answers for itself: another session the caller holds on the
backend/handler/streaming/access.py
+2 -2
View File
@@ -108,8 +108,8 @@ async def find_session_for_user(
def named_container(platform: str, container_key: str) -> ResolvedContainer:
"""A container serving a platform, found by key so one the pool leaves out
stays reachable. Raises 404 when the key names no such container."""
"""A container serving a platform, found by key so one in a later pool stays
reachable. Raises 404 when the key names no such container."""
candidate = entry_for_platform(containers_by_key().get(container_key, []), platform)
if candidate is None:
raise HTTPException(
backend/handler/streaming/config.py
+50 -25
View File
@@ -3,13 +3,15 @@
The raw YAML is loose: a container may serve one platform or a map of them, and
may leave its broker host to be derived. Resolution is memoized until the config
changes, so an unusable container is reported once rather than once per lookup.
A platform's containers form pools of interchangeable members, and a game claim
only walks the first.
"""
from __future__ import annotations
import json
import secrets
from collections.abc import Sequence
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse, urlunparse
@@ -518,38 +520,61 @@ def resolve_containers() -> tuple[ResolvedContainer, ...]:
for row, platform in _platform_entries(dict(entry)):
resolved.append(_resolve_one(row, platform, entry.get("label")))
_warn_about_later_pools(resolved)
_cache_fingerprint = fingerprint
_cached = tuple(resolved)
return _cached
def containers_for_platform(platform: str) -> list[ResolvedContainer]:
"""Every container serving a platform, in config order.
def _pools(containers: Iterable[ResolvedContainer]) -> list[list[ResolvedContainer]]:
"""Group containers into pools of interchangeable members. A pool sits at its
first member's config position, and members keep config order."""
pools: list[list[ResolvedContainer]] = []
for container in containers:
pool = next((p for p in pools if p[0].interchangeable_with(container)), None)
if pool is None:
pools.append([container])
else:
pool.append(container)
return pools
More than one entry is a pool and the claim takes the first free one, so a
container that disagrees with the head on emulator or card sync is not a
pool member and is left out.
"""
lower = platform.lower()
candidates: list[ResolvedContainer] = []
for container in resolve_containers():
if container.platform.lower() != lower:
continue
if not container.key:
# Nothing to dial, so a claim would have nowhere to go. The fleet
# view still lists it, which is where the operator sees why.
continue
if candidates and not candidates[0].interchangeable_with(container):
def _warn_about_later_pools(resolved: Sequence[ResolvedContainer]) -> None:
"""Name every container a game claim can never reach, once per config."""
by_platform: dict[str, list[ResolvedContainer]] = {}
for container in resolved:
if container.key:
by_platform.setdefault(container.platform.lower(), []).append(container)
for containers in by_platform.values():
later = [c.key for pool in _pools(containers)[1:] for c in pool]
if later:
log.warning(
"container for platform '%s' disagrees with the first one on "
"emulator, memory card sync or protocol, so it is not a pool "
"member, skipping: %s",
platform,
container.key,
"containers for platform '%s' disagree on emulator, memory card "
"sync, save picker or protocol, so game claims only use the first "
"pool; never claimed for a game: %s",
containers[0].platform,
", ".join(later),
)
continue
candidates.append(container)
return candidates
def pools_for_platform(platform: str) -> list[list[ResolvedContainer]]:
"""Every container serving a platform, grouped into pools. A container that
matches no other is a pool of one."""
lower = platform.lower()
return _pools(
container
for container in resolve_containers()
# Nothing to dial, so a claim would have nowhere to go. The fleet view
# still lists it, which is where the operator sees why.
if container.key and container.platform.lower() == lower
)
def containers_for_platform(platform: str) -> list[ResolvedContainer]:
"""The platform's first pool, which a game claim walks for the first free
container, in config order."""
pools = pools_for_platform(platform)
return pools[0] if pools else []
def containers_by_key() -> dict[str, list[ResolvedContainer]]:
backend/tests/endpoints/test_streaming.py
+66 -33
View File
@@ -54,6 +54,7 @@ from handler.streaming.config import (
ResolvedContainer,
_derive_broker_host,
emulator_display_label,
pools_for_platform,
reset_cache,
resolve_entry,
)
@@ -536,7 +537,7 @@ def test_clears_stale_saves_has_no_picker_to_gate_on_a_legacy_container(caplog):
assert "clears_stale_saves" in caplog.text
def test_a_container_that_disagrees_on_clearing_saves_is_not_a_pool_member(caplog):
def test_a_container_that_disagrees_on_clearing_saves_is_a_pool_of_its_own(caplog):
"""The picker is advertised from the head of the pool, so a member that
keeps its own newer files would take the pick and silently discard it."""
first = {
@@ -562,7 +563,7 @@ def test_a_container_that_disagrees_on_clearing_saves_is_not_a_pool_member(caplo
finally:
romm_logger.removeHandler(caplog.handler)
assert [c.clears_stale_saves for c in candidates] == [True]
assert "not a pool" in caplog.text
assert "never claimed for a game" in caplog.text
def test_legacy_containers_still_pool_under_an_inert_clearing_flag():
@@ -1223,6 +1224,21 @@ def test_pool_409s_only_once_every_container_is_held(
assert "2 containers" in r3.json()["detail"]["message"]
def test_a_claim_never_lands_in_a_later_pool(
client, access_token, viewer_access_token, rom: Rom
):
"""Only the first pool matches the settings a claim is validated against,
so a later pool stays out of reach even when it is the only one free."""
later = {**_pool_member(rom, 1), "emulator": "other"}
with _streaming(_pool_member(rom, 0), later):
r1 = _claim_ok(client, access_token, rom.id)
r2 = _claim_ok(client, viewer_access_token, rom.id)
assert _session_raw(later) is None
assert r1.status_code == 202
assert r1.json()["container"] == _key_of(_pool_member(rom, 0))
assert r2.status_code == 409
def test_a_pool_does_not_roll_its_own_holder_onto_a_second_container(
client, access_token, rom: Rom
):
@@ -1374,9 +1390,9 @@ def test_admin_release_rejects_a_container_that_serves_another_platform(
assert r.status_code == 404
def _outside_the_ps2_pool() -> dict:
def _in_a_later_ps2_pool() -> dict:
"""A second webstation that disagrees with `_webstation()` on the ps2
emulator, so the ps2 pool leaves it out."""
emulator, so it forms a later ps2 pool."""
return _webstation(
host="http://192.168.1.11:3000",
broker_host="http://192.168.1.11:8000",
@@ -1384,14 +1400,14 @@ def _outside_the_ps2_pool() -> dict:
)
def test_admin_release_ends_a_session_on_a_container_outside_the_pool(
def test_admin_release_ends_a_session_on_a_container_in_a_later_pool(
client, access_token
):
"""A container that disagrees with the pool on an emulator is left out of it,
but an admin can still end its session by naming it."""
outside = _outside_the_ps2_pool()
with _streaming(_webstation(), outside):
key = _key_of(outside)
"""A game claim never reaches a later pool, but an admin can still end a
session there by naming the container."""
later = _in_a_later_ps2_pool()
with _streaming(_webstation(), later):
key = _key_of(later)
assert key not in [c.key for c in streaming.containers_for_platform("ps2")]
assert _desktop(client, access_token, key)[0].status_code == 200
with patch("handler.streaming.commands.stop", return_value=None):
@@ -1402,25 +1418,25 @@ def test_admin_release_ends_a_session_on_a_container_outside_the_pool(
)
assert r.status_code == 200
assert r.json()["status"] == "released"
assert _session_raw(outside) is None
assert _session_raw(later) is None
def test_heartbeat_naming_the_container_refreshes_a_desktop_outside_the_pool(
def test_heartbeat_naming_the_container_refreshes_a_desktop_in_a_later_pool(
client, access_token
):
"""The platform's pool scan cannot see this container, so only the named key
keeps its desktop claim fresh."""
outside = _outside_the_ps2_pool()
with _streaming(_webstation(), outside):
key = _key_of(outside)
"""The platform's lookup only walks the first pool, so only the named key
keeps this desktop claim fresh."""
later = _in_a_later_ps2_pool()
with _streaming(_webstation(), later):
key = _key_of(later)
assert _desktop(client, access_token, key)[0].status_code == 200
_age_session_on(outside, session_store._STREAMING_SESSION_STALE_SECONDS + 60)
_age_session_on(later, session_store._STREAMING_SESSION_STALE_SECONDS + 60)
r = client.post(
"/api/streaming/sessions/ps2/heartbeat",
params={"container": key},
headers=_auth(access_token),
)
session = json.loads(_session_raw(outside))
session = json.loads(_session_raw(later))
assert r.status_code == 200
assert r.json()["status"] == "active"
assert not session_store.session_is_stale(session)
@@ -1429,9 +1445,9 @@ def test_heartbeat_naming_the_container_refreshes_a_desktop_outside_the_pool(
def test_heartbeat_naming_a_released_container_reports_ended(client, access_token):
"""A session the caller holds on another of the platform's containers must
not answer for the named one."""
outside = _outside_the_ps2_pool()
with _streaming(_webstation(), outside):
key = _key_of(outside)
later = _in_a_later_ps2_pool()
with _streaming(_webstation(), later):
key = _key_of(later)
assert (
_desktop(client, access_token, _key_of(_webstation()))[0].status_code == 200
)
@@ -1455,17 +1471,17 @@ def test_heartbeat_naming_a_released_container_reports_ended(client, access_toke
def test_heartbeat_naming_a_container_leaves_another_users_session_alone(
client, access_token, viewer_access_token
):
outside = _outside_the_ps2_pool()
with _streaming(_webstation(), outside):
key = _key_of(outside)
later = _in_a_later_ps2_pool()
with _streaming(_webstation(), later):
key = _key_of(later)
assert _desktop(client, access_token, key)[0].status_code == 200
_age_session_on(outside, session_store._STREAMING_SESSION_STALE_SECONDS + 60)
_age_session_on(later, session_store._STREAMING_SESSION_STALE_SECONDS + 60)
r = client.post(
"/api/streaming/sessions/ps2/heartbeat",
params={"container": key},
headers=_auth(viewer_access_token),
)
session = json.loads(_session_raw(outside))
session = json.loads(_session_raw(later))
assert r.status_code == 200
assert r.json()["status"] == "ended"
assert session_store.session_is_stale(session)
@@ -1511,9 +1527,9 @@ def test_heartbeat_refreshes_the_session_on_the_container_that_holds_it(
assert after > before
def test_a_container_that_disagrees_on_the_emulator_is_not_a_pool_member(caplog):
def test_a_container_that_disagrees_on_the_emulator_is_a_later_pool(caplog):
"""Pool members file states and cards in one place, so an entry naming a
different emulator is a separate setup rather than a spare container."""
different emulator is a pool of its own rather than a spare container."""
first = {
"platform": "ps2",
"host": "http://192.168.1.10:3000",
@@ -1523,16 +1539,33 @@ def test_a_container_that_disagrees_on_the_emulator_is_not_a_pool_member(caplog)
second = {**first, "host": "http://192.168.1.11:3000"}
second["broker_host"] = "http://192.168.1.11:8000"
second["emulator"] = "play"
third = {**first, "host": "http://192.168.1.12:3000"}
third["broker_host"] = "http://192.168.1.12:8000"
romm_logger = logging.getLogger("romm")
romm_logger.addHandler(caplog.handler)
try:
with _streaming(first, second):
with _streaming(first, second, third):
with caplog.at_level(logging.WARNING, logger="romm"):
pools = pools_for_platform("ps2")
candidates = streaming.containers_for_platform("ps2")
finally:
romm_logger.removeHandler(caplog.handler)
assert [c.emulator for c in candidates] == ["pcsx2"]
assert "not a pool" in caplog.text
assert [[c.emulator for c in pool] for pool in pools] == [
["pcsx2", "pcsx2"],
["play"],
]
assert candidates == pools[0]
# Once per config, not once per lookup.
assert caplog.text.count("never claimed for a game") == 1
assert _key_of(second) in caplog.text
def test_a_lone_container_is_a_pool_of_one(rom: Rom):
with _streaming(_pool_member(rom, 0)):
pools = pools_for_platform(rom.platform_slug)
assert [[c.key for c in pool] for pool in pools] == [
[_key_of(_pool_member(rom, 0))]
]
def test_webstation_pool_members_at_different_subfolders_are_still_a_pool(caplog):
@@ -1564,7 +1597,7 @@ def test_webstation_pool_members_at_different_subfolders_are_still_a_pool(caplog
"http://192.168.1.10:8000",
"http://192.168.1.11:8000",
]
assert "not a pool" not in caplog.text
assert "never claimed for a game" not in caplog.text
def test_a_proxied_host_disagreeing_with_its_subfolder_cannot_be_claimed(caplog):