From 53c0dcc2562351154948bd50e1a298aa34c162bd Mon Sep 17 00:00:00 2001 From: Jonathan Date: Mon, 13 Jul 2026 23:04:25 +0200 Subject: [PATCH] fix(worker): run streamrip on one persistent event loop (fixes Qobuz download failures) Each search/download called asyncio.run, creating and closing a fresh event loop. streamrip's module-global download-concurrency Semaphore binds to the loop on first use, so the 2nd+ download reused a Semaphore bound to a closed loop -> "Semaphore is bound to a different event loop", tracks failed, album arrived short -> "incomplete download" / needs_attention. Run all streamrip coroutines on one persistent loop. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/adapters/_streamrip.py | 19 +++++++++++++++++-- worker/tests/test_streamrip_loop.py | 17 +++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) create mode 100644 worker/tests/test_streamrip_loop.py diff --git a/worker/lyra_worker/adapters/_streamrip.py b/worker/lyra_worker/adapters/_streamrip.py index d02d9fd..60da230 100644 --- a/worker/lyra_worker/adapters/_streamrip.py +++ b/worker/lyra_worker/adapters/_streamrip.py @@ -11,6 +11,21 @@ def _hashed_password(password: str) -> str: _AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"} +# streamrip holds module-global asyncio primitives (a download-concurrency Semaphore) that +# bind to the running event loop on first use. `asyncio.run` creates a NEW loop per call and +# closes it, orphaning that Semaphore, so the 2nd+ download fails with "Semaphore ... is bound +# to a different event loop" and its tracks silently fail (-> incomplete download). Running every +# streamrip coroutine on ONE persistent loop keeps those globals valid across downloads. The +# worker processes one job at a time, so a single shared loop is safe. +_loop: "asyncio.AbstractEventLoop | None" = None + + +def _run(coro): + global _loop + if _loop is None or _loop.is_closed(): + _loop = asyncio.new_event_loop() + return _loop.run_until_complete(coro) + def _flatten_audio(dest: str) -> int: """streamrip writes into a nested `Artist - Album [...]/` subfolder. Move any audio @@ -77,7 +92,7 @@ class StreamripClient: return cfg def search_album(self, artist: str, album: str) -> list[dict]: - return asyncio.run(self._search(f"{artist} {album}")) + return _run(self._search(f"{artist} {album}")) async def _search(self, query: str) -> list[dict]: from streamrip.client import QobuzClient @@ -107,7 +122,7 @@ class StreamripClient: return results def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: - asyncio.run(self._download(source_ref, dest, on_progress)) + _run(self._download(source_ref, dest, on_progress)) # Count the files actually written and flatten streamrip's nested output into dest # so the tag stage can organize it. A partial download -> short count -> needs_attention. return {"track_count": _flatten_audio(dest), "path": dest} diff --git a/worker/tests/test_streamrip_loop.py b/worker/tests/test_streamrip_loop.py new file mode 100644 index 0000000..d2cc227 --- /dev/null +++ b/worker/tests/test_streamrip_loop.py @@ -0,0 +1,17 @@ +import asyncio + +from lyra_worker.adapters._streamrip import _run + + +def test_run_reuses_one_persistent_loop_across_calls(): + # The event-loop fix: unlike asyncio.run (which creates AND closes a fresh loop per call), + # _run runs every streamrip coroutine on ONE persistent loop — so streamrip's loop-bound + # global Semaphore stays valid across downloads instead of orphaning on a closed loop. + async def current_loop(): + return asyncio.get_running_loop() + + loop_a = _run(current_loop()) + loop_b = _run(current_loop()) + + assert loop_a is loop_b # same loop reused, not a fresh one each call + assert not loop_a.is_closed() # still open (asyncio.run would have closed it)