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) <noreply@anthropic.com>
This commit is contained in:
@@ -11,6 +11,21 @@ def _hashed_password(password: str) -> str:
|
|||||||
|
|
||||||
_AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"}
|
_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:
|
def _flatten_audio(dest: str) -> int:
|
||||||
"""streamrip writes into a nested `Artist - Album [...]/` subfolder. Move any audio
|
"""streamrip writes into a nested `Artist - Album [...]/` subfolder. Move any audio
|
||||||
@@ -77,7 +92,7 @@ class StreamripClient:
|
|||||||
return cfg
|
return cfg
|
||||||
|
|
||||||
def search_album(self, artist: str, album: str) -> list[dict]:
|
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]:
|
async def _search(self, query: str) -> list[dict]:
|
||||||
from streamrip.client import QobuzClient
|
from streamrip.client import QobuzClient
|
||||||
@@ -107,7 +122,7 @@ class StreamripClient:
|
|||||||
return results
|
return results
|
||||||
|
|
||||||
def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
|
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
|
# 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.
|
# so the tag stage can organize it. A partial download -> short count -> needs_attention.
|
||||||
return {"track_count": _flatten_audio(dest), "path": dest}
|
return {"track_count": _flatten_audio(dest), "path": dest}
|
||||||
|
|||||||
@@ -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)
|
||||||
Reference in New Issue
Block a user