danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Player revamp Phase 3a: extract stream state into testable per-session store

Browse source 
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.
This commit is contained in:
BoulderBadgeDad 2026-05-30 08:59:15 -07:00
parent c3aea58b03
commit ca90c6ae6f
6 changed files with 467 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

140
core/streaming/state.py Normal file
View file

@ -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()))

6
revamp_plan.md
View file

@ -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.
---

22
tests/streaming/test_prepare.py
View file

@ -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}

126
tests/streaming/test_stream_state_callsite_compat.py Normal file
View file

@ -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"

155
tests/streaming/test_stream_state_store.py Normal file
View file

@ -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

31
web_server.py
View file

@ -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,