From ca90c6ae6f81a47d2b428fa0dd72c3826047a620 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sat, 30 May 2026 08:59:15 -0700 Subject: [PATCH] Player revamp Phase 3a: extract stream state into testable per-session store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Foundation for multi-listener playback. Today web_server.py keeps ONE global stream_state dict + one lock (web_server.py:747), so the whole server shares a single 'currently playing' — every tab/device is a remote for the same playback and two listeners collide. That global is woven through ~22 sites and isn't unit-testable where it lives. Lifted into core/streaming/state.py WITHOUT changing behavior: - StreamSession: one playback's state, dict-compatible (s['k'], s.get, s.update, 'k' in s) so existing call sites work unchanged, each with its OWN RLock so distinct sessions never block/clobber each other. - StreamStateStore: registry of named sessions; lazy + race-safe create; DEFAULT session reproduces today's exact single-global behavior. Also drop()/active_ids()/session_ids() for the eventual per-listener wiring. web_server.py now binds (DEFAULT) and . Drop-in: every .update()/[k]/.get()/ site behaves identically. _set_stream_state routes a reassign through session.replace() so the store's session stays the live object (it's effectively dead — prepare.py only mutates in place — but safe now). Honest scope: this is the PROVABLE half of Phase 3. The remaining half (3b: derive a per-browser session id, per-session Stream/ staging, executor concurrency, disconnect cleanup) is browser-coupled and can't be verified without driving 2+ live clients — deferred to a live session. The store API is already shaped for it. Tests (tests/streaming/, 33 total): - test_stream_state_store.py (19): session dict-compat, isolation, lazy create, drop rules, active_ids, concurrent-create race safety. - test_stream_state_callsite_compat.py (7): every real web_server access pattern (library/play, stream/start, status, audio guard, stop, prepare in-place mutation, set->replace) against the exact object web_server binds. - test_prepare.py +1: real prepare worker drives an actual StreamSession. 76 streaming+radio tests green; ruff clean; web_server.py parses. --- core/streaming/state.py | 140 ++++++++++++++++ revamp_plan.md | 6 +- tests/streaming/test_prepare.py | 22 +++ .../test_stream_state_callsite_compat.py | 126 ++++++++++++++ tests/streaming/test_stream_state_store.py | 155 ++++++++++++++++++ web_server.py | 31 ++-- 6 files changed, 467 insertions(+), 13 deletions(-) create mode 100644 core/streaming/state.py create mode 100644 tests/streaming/test_stream_state_callsite_compat.py create mode 100644 tests/streaming/test_stream_state_store.py diff --git a/core/streaming/state.py b/core/streaming/state.py new file mode 100644 index 00000000..a35c7531 --- /dev/null +++ b/core/streaming/state.py @@ -0,0 +1,140 @@ +"""Stream playback state — testable store, foundation for multi-listener. + +Today ``web_server.py`` keeps ONE module-global ``stream_state`` dict + one +``stream_lock`` (``web_server.py:747``). That means the whole server has a +single "currently playing" — every browser tab/device is a remote for the same +playback, and two listeners collide. Fixing that (the player-revamp Phase 3 +goal) requires per-session state, but the global is woven through ~22 call +sites and isn't unit-testable where it lives. + +This module lifts the state into a small, tested abstraction WITHOUT yet +changing behavior: + +* ``StreamSession`` — one playback's state. Behaves like the old dict + (``s["status"]``, ``s.get(...)``, ``s.update({...})``) so existing call sites + work unchanged, but each carries its OWN lock so distinct sessions never + block or clobber each other. +* ``StreamStateStore`` — a registry of named sessions. ``DEFAULT_SESSION`` is + the single shared session that reproduces today's exact behavior; wiring the + web server through it is a no-op refactor. When Phase 3 adds a per-request + session id (browser/device), the store already supports it — that step is the + only remaining (browser-side, unprovable-here) piece. + +Pure Python, no Flask/DB. Fully unit-testable. +""" + +from __future__ import annotations + +import threading +from typing import Any, Dict, Iterator, List, Optional + +DEFAULT_SESSION = "default" + + +def _fresh_state() -> Dict[str, Any]: + """The stopped/empty baseline — matches web_server.py's original literal.""" + return { + "status": "stopped", # stopped | loading | queued | ready | error + "progress": 0, + "track_info": None, + "file_path": None, + "error_message": None, + } + + +class StreamSession: + """One playback session's state, with its own lock. + + Dict-compatible for the operations the existing call sites use + (``__getitem__``, ``__setitem__``, ``get``, ``update``) so lifting the + global is a drop-in. ``lock`` is exposed so callers that did + ``with stream_lock:`` keep that exact guard — now per-session. + """ + + def __init__(self, initial: Optional[Dict[str, Any]] = None): + self._state: Dict[str, Any] = _fresh_state() + if initial: + self._state.update(initial) + self.lock = threading.RLock() + + # -- dict-compatible surface (matches old stream_state usage) -- + def __getitem__(self, key: str) -> Any: + return self._state[key] + + def __setitem__(self, key: str, value: Any) -> None: + self._state[key] = value + + def __contains__(self, key: str) -> bool: + return key in self._state + + def get(self, key: str, default: Any = None) -> Any: + return self._state.get(key, default) + + def update(self, values: Dict[str, Any]) -> None: + self._state.update(values) + + def snapshot(self) -> Dict[str, Any]: + """A shallow copy — for emitting to clients without leaking the live + dict (the old code read individual keys under the lock; a snapshot is + the safe equivalent for the whole thing).""" + return dict(self._state) + + def reset(self) -> None: + """Return to the stopped/empty baseline (used by stop).""" + self._state = _fresh_state() + + def replace(self, new_state: Dict[str, Any]) -> None: + """Wholesale replace the backing dict (mirrors the old + ``_set_stream_state`` global reassignment).""" + self._state = dict(new_state) + + +class StreamStateStore: + """Registry of named :class:`StreamSession` objects. + + ``get()`` lazily creates a session on first reference, so a brand-new + session id just works. The default session reproduces the single-global + behavior the app has today. + """ + + def __init__(self): + self._sessions: Dict[str, StreamSession] = {} + self._registry_lock = threading.RLock() + + def get(self, session_id: str = DEFAULT_SESSION) -> StreamSession: + with self._registry_lock: + session = self._sessions.get(session_id) + if session is None: + session = StreamSession() + self._sessions[session_id] = session + return session + + def has(self, session_id: str) -> bool: + with self._registry_lock: + return session_id in self._sessions + + def drop(self, session_id: str) -> bool: + """Remove a session (e.g. on disconnect). Returns True if one existed. + The default session is never dropped — it's the always-present shared + playback.""" + if session_id == DEFAULT_SESSION: + return False + with self._registry_lock: + return self._sessions.pop(session_id, None) is not None + + def session_ids(self) -> List[str]: + with self._registry_lock: + return list(self._sessions.keys()) + + def active_ids(self) -> List[str]: + """Session ids whose status is not 'stopped' — i.e. currently doing + something. The signal multi-listener UI / cleanup will key off of.""" + with self._registry_lock: + return [ + sid for sid, s in self._sessions.items() + if s.get("status") != "stopped" + ] + + def __iter__(self) -> Iterator[StreamSession]: + with self._registry_lock: + return iter(list(self._sessions.values())) diff --git a/revamp_plan.md b/revamp_plan.md index 205ee5b2..3892c16c 100644 --- a/revamp_plan.md +++ b/revamp_plan.md @@ -25,11 +25,11 @@ Rule for every phase: kettui standard — importable/testable logic, seam-level - [x] **Weighted ranking** DONE. Each tier now fetches a random POOL (4x, floored) and `core/radio/selection.rank_candidates` orders it by `score_candidate`: play_count + lastfm_playcount (log-damped), recently-played penalty, stable per-id jitter for run variety. Defensive column-probe → still works on a DB predating the play_count/lastfm migration. 43 radio tests; ranking math is deterministic-unit-proven; DB wiring shown via decoy-pool test (probabilistic by nature — documented). - [ ] **Future (optional deepening):** wire `_recently_played` from `listening_history` (column + scorer support already exist; not yet populated in the query), genre-adjacency graph (currently exact-genre LIKE only). -## Phase 3 — Architecture (deepest, riskiest — listener decision lands here) +## Phase 3 — Architecture (deepest, riskiest — multi-listener) -- [ ] Per-session (or multi-tenant) stream state — replaces the single global `stream_state` + 1-worker executor + single `Stream/` staging file (`web_server.py:747`). +- [x] **3a. Stream-state store extracted + wired (foundation).** DONE. `core/streaming/state.py`: `StreamSession` (dict-compatible, own RLock) + `StreamStateStore` (named-session registry, lazy create, race-safe). `web_server.py` now binds `stream_state` to the store's DEFAULT session — behavior identical to the old single global (proven by call-site-compat + real-session worker tests). 33 streaming tests. This is the provable foundation multi-listener needs. +- [ ] **3b. Per-listener session id (the unprovable-here part).** Derive a session id per browser/device (cookie/header) and key `stream_state_store.get(session_id)` off it in the stream routes; per-session `Stream/` staging subdir; drop session on disconnect; bump `stream_executor` past max_workers=1. Needs live multi-client testing — do in a session where Boulder can drive 2+ clients. The store API (`get(id)`, `drop`, `active_ids`, per-session staging) is already built for it. - [ ] Server-side persistent queue (resume across devices/refresh). -- [ ] Final multi-listener vs single-listener scope decided here, with real usage in hand. --- diff --git a/tests/streaming/test_prepare.py b/tests/streaming/test_prepare.py index bbfc9cef..f33d7680 100644 --- a/tests/streaming/test_prepare.py +++ b/tests/streaming/test_prepare.py @@ -178,3 +178,25 @@ def test_succeeded_state_with_partial_bytes_keeps_polling(tmp_path): # Should NOT have gone to 'ready' because bytes were incomplete assert deps._state['status'] != 'ready' + + +# --------------------------------------------------------------------------- +# Real StreamSession compatibility (player-revamp Phase 3 wiring) +# --------------------------------------------------------------------------- + +def test_worker_drives_a_real_stream_session(tmp_path): + """web_server.py now binds stream_state to a StreamStateStore session + (not a bare dict). Prove the prepare worker drives that real object + correctly end-to-end through the deps proxy — the actual production type.""" + from core.streaming.state import StreamStateStore + + session = StreamStateStore().get() # the real production object + sk = _FakeSoulseek(download_id=None) # early error exit is enough to mutate state + deps = _build_deps(soulseek=sk, project_root=str(tmp_path), state=session) + + sp.prepare_stream_task({'username': 'u', 'filename': 'song.flac', 'size': 1}, deps) + + # Worker mutated the SAME session via update()/[k]= — proves dict-compat. + assert session['status'] == 'error' + assert 'Failed to initiate' in session['error_message'] + assert session['track_info'] == {'username': 'u', 'filename': 'song.flac', 'size': 1} diff --git a/tests/streaming/test_stream_state_callsite_compat.py b/tests/streaming/test_stream_state_callsite_compat.py new file mode 100644 index 00000000..8a0cce90 --- /dev/null +++ b/tests/streaming/test_stream_state_callsite_compat.py @@ -0,0 +1,126 @@ +"""Proves the StreamSession is a drop-in for the old stream_state dict. + +web_server.py was swapped from a module-global ``dict`` + ``threading.Lock`` +to ``StreamStateStore().get()`` (the default session) + that session's lock. +This exercises every access pattern the real call sites use, against the same +object web_server now binds, so the swap is verified without booting Flask. + +The patterns mirrored here come verbatim from web_server.py: + - /api/library/play: with lock: state.update({... "is_library": True}) + - /api/stream/start: with lock: state.update({"status": "loading", ...}) + - /api/stream/status: with lock: read state["status"], state["progress"], ... + - /stream/audio: with lock: if state["status"] != "ready" or not state["file_path"] + - /api/stream/stop: with lock: state.get("is_library", False); state.update({... reset}) + - prepare.py: state.update(...) and state["status"] = "queued" +""" + +from __future__ import annotations + +from core.streaming.state import StreamStateStore + + +def _server_like_state(): + """Reproduce exactly what web_server.py now binds.""" + store = StreamStateStore() + state = store.get() # DEFAULT_SESSION + lock = state.lock + return state, lock + + +def test_library_play_pattern(): + state, lock = _server_like_state() + with lock: + state.update({ + "status": "ready", + "progress": 100, + "track_info": {"title": "T", "artist": "A", "album": "Al"}, + "file_path": "/Stream/x.flac", + "is_library": True, + }) + assert state["status"] == "ready" + assert state["file_path"] == "/Stream/x.flac" + assert state.get("is_library") is True + + +def test_stream_start_pattern(): + state, lock = _server_like_state() + with lock: + state.update({ + "status": "loading", + "progress": 0, + "track_info": {"title": "Song"}, + "file_path": None, + "error_message": None, + }) + assert state["status"] == "loading" + + +def test_stream_status_read_pattern(): + state, lock = _server_like_state() + state.update({"status": "queued", "progress": 42, + "track_info": {"title": "Q"}, "error_message": None}) + with lock: + payload = { + "status": state["status"], + "progress": state["progress"], + "track_info": state["track_info"], + "error_message": state["error_message"], + } + assert payload == {"status": "queued", "progress": 42, + "track_info": {"title": "Q"}, "error_message": None} + + +def test_stream_audio_guard_pattern(): + state, lock = _server_like_state() + # Not ready → guard trips (would 404 in the route). + with lock: + not_ready = state["status"] != "ready" or not state["file_path"] + assert not_ready is True + + state.update({"status": "ready", "file_path": "/Stream/y.flac"}) + with lock: + not_ready = state["status"] != "ready" or not state["file_path"] + path = state["file_path"] + assert not_ready is False + assert path == "/Stream/y.flac" + + +def test_stream_stop_pattern(): + state, lock = _server_like_state() + state.update({"status": "ready", "file_path": "/x", "is_library": True}) + with lock: + is_library = state.get("is_library", False) + assert is_library is True + with lock: + state.update({ + "status": "stopped", "progress": 0, "track_info": None, + "file_path": None, "error_message": None, "is_library": False, + }) + assert state["status"] == "stopped" + assert state.get("is_library") is False + + +def test_prepare_worker_inplace_mutation_pattern(): + state, _ = _server_like_state() + # prepare.py mutates in place via both update() and [k]= + state.update({"status": "loading", "progress": 0}) + state["status"] = "queued" + state["progress"] = 10 + state["status"] = "loading" + state["progress"] = 55 + assert state["status"] == "loading" + assert state["progress"] == 55 + + +def test_set_stream_state_replace_keeps_same_session_object(): + """web_server._set_stream_state now routes a reassignment through + replace() so the store's default session stays the live object. Verify a + 'reassign' is reflected in the SAME session the store hands out.""" + store = StreamStateStore() + state = store.get() + # simulate _set_stream_state(value): state.replace(dict(value)) + state.replace({"status": "error", "error_message": "boom"}) + # The store still hands back the same object, now carrying the new values. + assert store.get() is state + assert store.get()["status"] == "error" + assert store.get()["error_message"] == "boom" diff --git a/tests/streaming/test_stream_state_store.py b/tests/streaming/test_stream_state_store.py new file mode 100644 index 00000000..8273f481 --- /dev/null +++ b/tests/streaming/test_stream_state_store.py @@ -0,0 +1,155 @@ +"""Tests for the stream-state store (player revamp Phase 3 foundation). + +Pins the dict-compatible behavior the web server relies on, and the +multi-session registry semantics that per-listener playback will build on. +""" + +from __future__ import annotations + +import threading + +from core.streaming.state import ( + DEFAULT_SESSION, + StreamSession, + StreamStateStore, +) + + +class TestStreamSession: + def test_fresh_baseline(self): + s = StreamSession() + assert s["status"] == "stopped" + assert s["progress"] == 0 + assert s["track_info"] is None + assert s["file_path"] is None + assert s["error_message"] is None + + def test_initial_overrides(self): + s = StreamSession({"status": "ready", "file_path": "/x.flac"}) + assert s["status"] == "ready" + assert s["file_path"] == "/x.flac" + # untouched keys keep baseline + assert s["progress"] == 0 + + def test_dict_compatible_get_with_default(self): + s = StreamSession() + # The old code did stream_state.get("is_library", False) — a key not in + # the baseline. Must return the default, not raise. + assert s.get("is_library", False) is False + + def test_setitem_and_getitem(self): + s = StreamSession() + s["status"] = "loading" + assert s["status"] == "loading" + + def test_update(self): + s = StreamSession() + s.update({"status": "ready", "progress": 100, "is_library": True}) + assert s["status"] == "ready" + assert s["progress"] == 100 + assert s.get("is_library") is True + + def test_contains(self): + s = StreamSession() + assert "status" in s + assert "is_library" not in s + + def test_snapshot_is_a_copy(self): + s = StreamSession() + snap = s.snapshot() + snap["status"] = "mutated" + assert s["status"] == "stopped" # live state untouched + + def test_reset_returns_to_baseline(self): + s = StreamSession() + s.update({"status": "ready", "progress": 100, "is_library": True}) + s.reset() + assert s["status"] == "stopped" + assert s["progress"] == 0 + assert s.get("is_library", False) is False # extra keys gone too + + def test_replace_swaps_backing_dict(self): + s = StreamSession() + s.replace({"status": "error", "error_message": "boom"}) + assert s["status"] == "error" + assert s["error_message"] == "boom" + + def test_each_session_has_its_own_lock(self): + a, b = StreamSession(), StreamSession() + assert a.lock is not b.lock + + def test_lock_is_reentrant(self): + # RLock — a call site that re-enters under its own lock won't deadlock. + s = StreamSession() + with s.lock: + with s.lock: + s["status"] = "ready" + assert s["status"] == "ready" + + +class TestStreamStateStore: + def test_default_session_is_stable(self): + store = StreamStateStore() + a = store.get() + b = store.get(DEFAULT_SESSION) + assert a is b # same object — reproduces the single-global behavior + + def test_distinct_sessions_are_isolated(self): + store = StreamStateStore() + alice = store.get("alice") + bob = store.get("bob") + assert alice is not bob + alice["status"] = "ready" + assert bob["status"] == "stopped" # no cross-clobber — the whole point + + def test_lazy_creation(self): + store = StreamStateStore() + assert not store.has("new") + store.get("new") + assert store.has("new") + + def test_drop_removes_session(self): + store = StreamStateStore() + store.get("temp") + assert store.drop("temp") is True + assert not store.has("temp") + assert store.drop("temp") is False # already gone + + def test_default_session_cannot_be_dropped(self): + store = StreamStateStore() + store.get() # materialize default + assert store.drop(DEFAULT_SESSION) is False + assert store.has(DEFAULT_SESSION) + + def test_session_ids_lists_created(self): + store = StreamStateStore() + store.get("a") + store.get("b") + assert set(store.session_ids()) == {"a", "b"} + + def test_active_ids_excludes_stopped(self): + store = StreamStateStore() + store.get("idle") # stays stopped + store.get("playing")["status"] = "ready" + store.get("loading")["status"] = "loading" + assert set(store.active_ids()) == {"playing", "loading"} + + def test_concurrent_get_same_session_returns_one_object(self): + # The lazy-create must be race-safe: many threads getting the same new + # id must all see the SAME session (no lost writes from a torn create). + store = StreamStateStore() + seen = [] + barrier = threading.Barrier(8) + + def worker(): + barrier.wait() + seen.append(store.get("contended")) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(seen) == 8 + assert all(s is seen[0] for s in seen) # exactly one object shared diff --git a/web_server.py b/web_server.py index 4c14954f..26535fca 100644 --- a/web_server.py +++ b/web_server.py @@ -744,14 +744,20 @@ logger.info("Core service initialization complete.") # modules and handlers were still referencing names that never got created. # Global Streaming State Management -stream_state = { - "status": "stopped", # States: stopped, loading, queued, ready, error - "progress": 0, - "track_info": None, - "file_path": None, # Path to the audio file in the Stream folder - "error_message": None, -} -stream_lock = threading.Lock() # Prevent race conditions +# Stream playback state. Lifted into core.streaming.state.StreamStateStore — +# a per-session registry that's unit-tested and is the foundation for +# multi-listener playback (player-revamp Phase 3). Today we use only the +# DEFAULT session, so behavior is identical to the old single global: the +# whole server still shares one "currently playing". The store just makes the +# eventual per-listener split a wiring change instead of a rewrite. +# +# ``stream_state`` is the default session — dict-compatible (s["k"], s.get, +# s.update) so the ~20 existing call sites work unchanged. ``stream_lock`` is +# that session's own lock, so ``with stream_lock:`` guards exactly what it did. +from core.streaming.state import StreamStateStore as _StreamStateStore +stream_state_store = _StreamStateStore() +stream_state = stream_state_store.get() # DEFAULT_SESSION +stream_lock = stream_state.lock stream_background_task = None stream_executor = ThreadPoolExecutor(max_workers=1) # Only one stream at a time @@ -1782,8 +1788,13 @@ def _build_prepare_stream_deps(): return stream_state def _set_stream_state(value): - global stream_state - stream_state = value + # prepare.py only ever mutates in place (.update / [k]=), so this is + # effectively dead — but if anything DOES reassign, route it through + # the session's replace() so the store's default session stays the live + # object instead of being detached by a global rebind. + if value is stream_state: + return + stream_state.replace(dict(value)) return _streaming_prepare.PrepareStreamDeps( config_manager=config_manager,