From 398ac43554caad018082d2bccacf7b6ff4a9f736 Mon Sep 17 00:00:00 2001 From: Ismael Leon Date: Thu, 20 Aug 2026 19:09:45 -0600 Subject: [PATCH] Fix yt-dlp stream stalls and unreaped processes Replace the loose (Popen, kill_stream, classify_stream_error) trio with an AudioStream class that owns the child process end to end. Two problems it fixes: stderr was a pipe nobody read while a track played. classify_stream_error only drained it after playback ended, so a chatty yt-dlp (fragment retries, HTTP warnings, throttling notices) would fill the ~64 KB kernel buffer, block on write, and stop producing audio. FFmpeg then starved and playback stalled with nothing reported to the user. stderr now goes to a temporary file, which never blocks. classify_error() reads only the tail, since the real error is written last and can sit behind megabytes of progress noise. The child was killed but not fully released. Pipes were left for the GC to close and TimeoutExpired was swallowed. close() now kills, reaps, captures the error text while the file is still open, and closes both handles - and is idempotent. The live host had accumulated a yt-dlp zombie that survived 19 days on exactly this path. close() blocks briefly, so the player calls it through an executor rather than from the event loop, in both the playback loop and destroy(). Verified end to end against real yt-dlp: SoundCloud 2.86s to first byte, YouTube 2.42s, 256 KB streamed each, process reaped, pipes closed. The new flood test reproduces the stall directly - 1 MB of stderr, 16x the pipe buffer, which deadlocks under the old code and now delivers audio normally. Closes #6 --- .gitignore | 5 + services/media.py | 159 +++++++++++++++++++------ tests/test_audio_stream.py | 230 ++++++++++++++++++++++++++++++++++++ tests/test_media_helpers.py | 74 +----------- utils/player.py | 25 ++-- 5 files changed, 376 insertions(+), 117 deletions(-) create mode 100644 tests/test_audio_stream.py diff --git a/.gitignore b/.gitignore index 2eeaf9f..a01eb44 100644 --- a/.gitignore +++ b/.gitignore @@ -134,6 +134,11 @@ celerybeat.pid # SageMath parsed files *.sage.py +# yt-dlp fragment artifacts: streaming to stdout can drop these in the CWD. +--Frag* +*.part +*.ytdl + # Project-specific: AI assistant notes & sensitive runtime files CLAUDE.md .claude/ diff --git a/services/media.py b/services/media.py index 5296292..c315835 100644 --- a/services/media.py +++ b/services/media.py @@ -21,6 +21,7 @@ import os import sys import subprocess +import tempfile import asyncio import logging from typing import Optional @@ -194,13 +195,127 @@ def _stream_target(track: dict) -> str: return target -def spawn_stream(track: dict) -> subprocess.Popen: +# How much of yt-dlp's stderr to inspect when a stream produced no audio. +# The real error is written last, and a throttled download can emit megabytes of +# progress noise ahead of it, so only the tail is worth reading. +_STDERR_TAIL_BYTES = 16 * 1024 + +# Phrases YouTube uses when it wants a signed-in session (i.e. fresh cookies). +_BOT_CHECK_MARKERS = ("not a bot", "sign in to confirm") + +# Longest we wait for a SIGKILLed yt-dlp to actually disappear. +_REAP_TIMEOUT = 5.0 + + +def _close_quietly(handle) -> None: + """Close a pipe/file, ignoring anything that goes wrong during teardown.""" + if handle is None: + return + try: + handle.close() + except Exception: + pass + + +class AudioStream: """ - Start a yt-dlp subprocess that writes the track's best audio to stdout. + A running ``yt-dlp`` process writing one track's audio to a pipe. + + ``stdout`` feeds :func:`make_pipe_source`. The caller owns the stream and + MUST :meth:`close` it when playback ends, is skipped, or the player is + destroyed — otherwise the child is never reaped. - The returned process' ``stdout`` is meant to feed ``make_pipe_source``. - Caller owns the process and MUST ``kill_stream`` it when playback ends. + Two deliberate choices about the child's streams: + + * **stdout is a pipe.** That is what bounds memory on a small host: yt-dlp + blocks as soon as FFmpeg stops reading, instead of buffering a whole + track in RAM. + * **stderr is a temporary file, never a pipe.** Nothing reads stderr while + the track plays, so a pipe whose ~64 KB kernel buffer filled up would + block yt-dlp mid-write, starve FFmpeg and stall playback silently. + Writing to a file never blocks. + + :meth:`close` and :meth:`classify_error` may both block briefly (waiting on + the child, reading the file), so call them from an executor, never straight + from the event loop. """ + + __slots__ = ("_proc", "_errfile", "_error", "_closed") + + def __init__(self, proc: subprocess.Popen, errfile) -> None: + self._proc = proc + self._errfile = errfile + self._error: Optional[str] = None + self._closed = False + + @classmethod + def launch(cls, cmd: list[str]) -> "AudioStream": + """Spawn ``cmd`` with the stdout/stderr wiring described above.""" + errfile = tempfile.TemporaryFile() + try: + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=errfile, + bufsize=64 * 1024, # large buffer smooths delivery to FFmpeg + ) + except Exception: + _close_quietly(errfile) + raise + return cls(proc, errfile) + + @property + def stdout(self): + """The audio pipe, or ``None`` once the stream has been closed.""" + return None if self._closed else self._proc.stdout + + def close(self) -> None: + """Kill the process, reap it and release both handles. Idempotent.""" + if self._closed: + return + self._closed = True + proc = self._proc + try: + if proc.poll() is None: + try: + proc.kill() + except OSError: + pass # already gone between poll and kill + try: + proc.wait(timeout=_REAP_TIMEOUT) + except subprocess.TimeoutExpired: + log.warning("yt-dlp (pid %s) survived SIGKILL; not reaped", proc.pid) + # Read the error while the file is still open — the player asks for + # the reason only after the stream has been torn down. + self._error = self._read_error() + finally: + _close_quietly(proc.stdout) + _close_quietly(self._errfile) + + def classify_error(self) -> str: + """ + Why this stream produced no audio. + + ``"blocked"`` means YouTube demanded a signed-in session and the cookies + need refreshing; ``"unavailable"`` covers everything else. Safe to call + before or after :meth:`close`, and repeatable. + """ + err = self._error if self._error is not None else self._read_error() + return "blocked" if any(m in err for m in _BOT_CHECK_MARKERS) else "unavailable" + + def _read_error(self) -> str: + """The tail of the child's stderr, lowercased. Never raises.""" + try: + self._errfile.seek(0, os.SEEK_END) + start = max(0, self._errfile.tell() - _STDERR_TAIL_BYTES) + self._errfile.seek(start) + return self._errfile.read().decode("utf-8", "ignore").lower() + except Exception: + return "" + + +def spawn_stream(track: dict) -> AudioStream: + """Start streaming a track's best audio through yt-dlp.""" cmd = [ sys.executable, "-m", "yt_dlp", "-f", "bestaudio/best", @@ -213,11 +328,7 @@ def spawn_stream(track: dict) -> subprocess.Popen: if cookies: cmd += ["--cookies", cookies] cmd.append(_stream_target(track)) - # Large stdout buffer smooths delivery to FFmpeg; stderr captured for errors. - return subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - bufsize=64 * 1024, - ) + return AudioStream.launch(cmd) def make_pipe_source(stdin, *, volume: float = 0.5, ffmpeg_filter: str = ""): @@ -226,33 +337,3 @@ def make_pipe_source(stdin, *, volume: float = 0.5, ffmpeg_filter: str = ""): options = f"-vn -af {ffmpeg_filter}" if ffmpeg_filter else "-vn" source = discord.FFmpegPCMAudio(stdin, pipe=True, options=options) return discord.PCMVolumeTransformer(source, volume=volume) - - -def kill_stream(proc: Optional[subprocess.Popen]) -> None: - """Terminate a yt-dlp streaming process and reap it (idempotent).""" - if proc is None or proc.poll() is not None: - return - try: - proc.kill() - proc.wait(timeout=2) - except subprocess.TimeoutExpired: - pass - except Exception: - pass - - -def classify_stream_error(proc: Optional[subprocess.Popen]) -> str: - """ - Read a finished yt-dlp process' stderr and classify why it produced no audio. - - Returns "blocked" (YouTube bot-check — needs fresh cookies), or "unavailable". - """ - if proc is None: - return "unavailable" - try: - err = (proc.stderr.read() or b"").decode("utf-8", "ignore").lower() if proc.stderr else "" - except Exception: - err = "" - if "not a bot" in err or "sign in to confirm" in err: - return "blocked" - return "unavailable" diff --git a/tests/test_audio_stream.py b/tests/test_audio_stream.py new file mode 100644 index 0000000..7a93593 --- /dev/null +++ b/tests/test_audio_stream.py @@ -0,0 +1,230 @@ +""" +``services.media.AudioStream`` - the yt-dlp process that feeds FFmpeg. + +These tests spawn **real** subprocesses (a short Python one-liner, never +yt-dlp) because the bugs being guarded against are about operating-system pipe +behaviour and process reaping, which mocks cannot reproduce. +""" + +import subprocess +import sys + +import pytest + +from services import media +from services.media import AudioStream + + +def script(body: str) -> list: + return [sys.executable, "-c", body] + + +@pytest.fixture +def spawned(): + """Tracks streams so a failing test cannot leave a process behind.""" + created = [] + + def _launch(body: str) -> AudioStream: + stream = AudioStream.launch(script(body)) + created.append(stream) + return stream + + yield _launch + for stream in created: + stream.close() + + +# -- the deadlock this class exists to prevent ------------------------- + +def test_stderr_is_never_a_pipe(monkeypatch): + """ + Regression guard for the stall bug. + + Nothing reads stderr while a track plays, so if stderr were a pipe, a + chatty yt-dlp would fill the ~64 KB kernel buffer, block on write and stop + producing audio. It must go somewhere that never blocks. + """ + captured = {} + + class FakePopen: + def __init__(self, cmd, **kwargs): + captured.update(kwargs) + self.stdout = None + self.stderr = None + + def poll(self): + return None + + monkeypatch.setattr(subprocess, "Popen", FakePopen) + AudioStream.launch(["true"]) + + assert captured["stderr"] is not subprocess.PIPE + assert captured["stderr"] is not None, "stderr must be captured, not discarded" + + +def test_a_process_flooding_stderr_still_delivers_audio(spawned): + """ + 1 MB of stderr is ~16x the pipe buffer. With the old piped stderr this + deadlocks and no audio ever arrives. + """ + stream = spawned( + "import sys;" + "sys.stderr.write('x' * 1024 * 1024);" + "sys.stderr.flush();" + "sys.stdout.buffer.write(b'AUDIO' * 200);" + "sys.stdout.buffer.flush()" + ) + data = stream.stdout.read(1000) + assert data.startswith(b"AUDIO"), "the process stalled writing to stderr" + + +def test_stdout_still_applies_backpressure(spawned): + """ + stdout must stay a pipe. That is what bounds memory: yt-dlp blocks when + FFmpeg is not consuming, instead of buffering a whole track in RAM. + """ + stream = spawned("import sys; sys.stdout.buffer.write(b'A' * 64); sys.stdout.buffer.flush()") + assert stream.stdout is not None + assert stream.stdout.read(64) == b"A" * 64 + + +# -- reaping ----------------------------------------------------------- + +def test_close_reaps_the_process_leaving_no_zombie(spawned): + """ + The live deployment accumulated a yt-dlp zombie that survived 19 days + because the child was killed but never waited on. + """ + stream = spawned("import time; time.sleep(60)") + assert stream._proc.poll() is None, "the process should still be running" + + stream.close() + + assert stream._proc.poll() is not None, "close() must reap the child, not just kill it" + assert stream._proc.returncode is not None + + +def test_close_is_idempotent(spawned): + stream = spawned("import time; time.sleep(60)") + stream.close() + stream.close() + stream.close() + assert stream._proc.poll() is not None + + +def test_close_on_an_already_finished_process(spawned): + stream = spawned("pass") + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + stream.close() # must not raise + assert stream._proc.poll() is not None + + +def test_close_releases_the_pipes(spawned): + """Leaked pipe file descriptors accumulate over a long uptime.""" + stream = spawned("import time; time.sleep(60)") + stdout = stream.stdout + stream.close() + assert stdout.closed, "the stdout pipe must be closed explicitly, not left to the GC" + + +def test_stdout_is_none_after_close(spawned): + stream = spawned("import time; time.sleep(60)") + stream.close() + assert stream.stdout is None + + +# -- error classification ---------------------------------------------- + +def test_classify_detects_the_youtube_bot_check(spawned): + stream = spawned( + "import sys; sys.stderr.write(\"ERROR: Sign in to confirm you are not a bot\")" + ) + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + stream.close() + assert stream.classify_error() == "blocked" + + +def test_classify_reports_other_failures_as_unavailable(spawned): + stream = spawned("import sys; sys.stderr.write('ERROR: Video unavailable')") + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + stream.close() + assert stream.classify_error() == "unavailable" + + +def test_classify_works_after_close(spawned): + """ + The player kills the stream before asking why it failed, so classification + has to survive teardown. + """ + stream = spawned("import sys; sys.stderr.write('please sign in to confirm')") + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + stream.close() + assert stream.classify_error() == "blocked" + assert stream.classify_error() == "blocked", "classification must be repeatable" + + +def test_classify_only_reads_the_tail_of_a_huge_stderr(spawned): + """A megabyte of throttling noise must not be loaded whole to find a phrase.""" + stream = spawned( + "import sys;" + "sys.stderr.write('noise ' * 200000);" + "sys.stderr.write('ERROR: Sign in to confirm you are not a bot')" + ) + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + stream.close() + assert stream.classify_error() == "blocked" + + +def test_classify_survives_undecodable_stderr(spawned): + stream = spawned("import sys; sys.stderr.buffer.write(b'\\xff\\xfe bad bytes')") + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + stream.close() + assert stream.classify_error() == "unavailable" + + +def test_classify_before_close_still_works(spawned): + stream = spawned("import sys; sys.stderr.write('ERROR: Private video')") + stream._proc.wait(timeout=30) # deterministic: the child has flushed stderr + assert stream.classify_error() == "unavailable" + + +# -- command construction ---------------------------------------------- + +def test_spawn_stream_targets_the_track_url(monkeypatch): + captured = {} + monkeypatch.setattr(AudioStream, "launch", classmethod( + lambda cls, cmd: captured.setdefault("cmd", cmd) + )) + media.spawn_stream({"url": "https://example.invalid/watch", "title": "T"}) + assert captured["cmd"][-1] == "https://example.invalid/watch" + + +def test_spawn_stream_falls_back_to_a_search_when_there_is_no_url(monkeypatch): + captured = {} + monkeypatch.setattr(AudioStream, "launch", classmethod( + lambda cls, cmd: captured.setdefault("cmd", cmd) + )) + media.spawn_stream({"url": None, "query": "bohemian rhapsody", "title": "T"}) + assert captured["cmd"][-1] == "ytsearch1:bohemian rhapsody" + + +def test_spawn_stream_passes_cookies_when_configured(monkeypatch): + captured = {} + monkeypatch.setattr(AudioStream, "launch", classmethod( + lambda cls, cmd: captured.setdefault("cmd", cmd) + )) + monkeypatch.setitem(media.YTDL_OPTIONS, "cookiefile", "/tmp/cookies.txt") + media.spawn_stream({"url": "https://example.invalid/x", "title": "T"}) + cmd = captured["cmd"] + assert "--cookies" in cmd + assert cmd[cmd.index("--cookies") + 1] == "/tmp/cookies.txt" + + +def test_spawn_stream_omits_cookies_when_not_configured(monkeypatch): + captured = {} + monkeypatch.setattr(AudioStream, "launch", classmethod( + lambda cls, cmd: captured.setdefault("cmd", cmd) + )) + monkeypatch.delitem(media.YTDL_OPTIONS, "cookiefile", raising=False) + media.spawn_stream({"url": "https://example.invalid/x", "title": "T"}) + assert "--cookies" not in captured["cmd"] diff --git a/tests/test_media_helpers.py b/tests/test_media_helpers.py index 6aec248..38723b9 100644 --- a/tests/test_media_helpers.py +++ b/tests/test_media_helpers.py @@ -1,15 +1,11 @@ """ Pure helpers in ``services.media``. -Nothing here touches the network or spawns a real yt-dlp: these cover the -parsing and process-handling code that decides what gets searched and how a -failure is reported back to the user. +Nothing here touches the network or spawns a process: these cover the parsing +that decides what gets searched and how a yt-dlp result becomes a track. +Process handling lives in ``tests/test_audio_stream.py``. """ -import io -import subprocess -from unittest.mock import MagicMock - import pytest from services import media @@ -114,64 +110,6 @@ def test_first_entry_passes_through_a_single_result(): assert media._first_entry({"title": "Solo"})["title"] == "Solo" -# ── failure classification ──────────────────────────────────────────── - -def _finished_proc(stderr: bytes) -> MagicMock: - proc = MagicMock(spec=subprocess.Popen) - proc.stderr = io.BytesIO(stderr) - return proc - - -@pytest.mark.parametrize("stderr", [ - b"ERROR: Sign in to confirm you're not a bot", - b"ERROR: SIGN IN TO CONFIRM you are not a bot", - b"please confirm you're not a bot to continue", -]) -def test_bot_checks_are_classified_as_blocked(stderr): - assert media.classify_stream_error(_finished_proc(stderr)) == "blocked" - - -@pytest.mark.parametrize("stderr", [ - b"ERROR: Video unavailable", - b"ERROR: Private video", - b"", -]) -def test_other_failures_are_classified_as_unavailable(stderr): - assert media.classify_stream_error(_finished_proc(stderr)) == "unavailable" - - -def test_classify_handles_a_missing_process(): - assert media.classify_stream_error(None) == "unavailable" - - -def test_classify_survives_undecodable_stderr(): - """yt-dlp can emit non-UTF-8 bytes; classification must not raise.""" - assert media.classify_stream_error(_finished_proc(b"\xff\xfe invalid")) == "unavailable" - - -# ── process teardown ────────────────────────────────────────────────── - -def test_kill_stream_accepts_none(): - media.kill_stream(None) # must not raise - - -def test_kill_stream_leaves_an_already_finished_process_alone(): - proc = MagicMock(spec=subprocess.Popen) - proc.poll.return_value = 0 - media.kill_stream(proc) - proc.kill.assert_not_called() - - -def test_kill_stream_kills_a_running_process(): - proc = MagicMock(spec=subprocess.Popen) - proc.poll.return_value = None - media.kill_stream(proc) - proc.kill.assert_called_once() - - -def test_kill_stream_is_idempotent(): - proc = MagicMock(spec=subprocess.Popen) - proc.poll.side_effect = [None, 0] # running, then dead - media.kill_stream(proc) - media.kill_stream(proc) - assert proc.kill.call_count == 1 +# Process teardown and failure classification moved to AudioStream when the +# stderr-deadlock fix landed; they are covered against real subprocesses in +# tests/test_audio_stream.py. diff --git a/utils/player.py b/utils/player.py index be11635..9cb252b 100644 --- a/utils/player.py +++ b/utils/player.py @@ -53,7 +53,7 @@ def __init__(self, bot: discord.Client, guild: discord.Guild, self.effect_filter: str = "" self._start_ts: float = 0.0 # monotonic clock when current started - self._proc = None # active yt-dlp streaming subprocess + self._stream: Optional[media.AudioStream] = None # active yt-dlp stream # Signalling between commands and the playback loop. self._next = asyncio.Event() # set when the current source finishes @@ -180,12 +180,13 @@ async def _player_loop(self) -> None: return self.destroy() # Stream the audio through yt-dlp → FFmpeg (see services.media). - proc = await self.bot.loop.run_in_executor(None, media.spawn_stream, track) - self._proc = proc + stream = await self.bot.loop.run_in_executor( + None, media.spawn_stream, track) + self._stream = stream was_replay = self._replay self._replay = False source = media.make_pipe_source( - proc.stdout, volume=self.volume, ffmpeg_filter=self.effect_filter, + stream.stdout, volume=self.volume, ffmpeg_filter=self.effect_filter, ) vc.play(source, after=self._after_play) self._start_ts = time.monotonic() @@ -199,15 +200,16 @@ async def _player_loop(self) -> None: await self._next.wait() played = time.monotonic() - self._start_ts source.cleanup() # stop FFmpeg - media.kill_stream(proc) # stop yt-dlp - self._proc = None + # close() waits on the child and reads its stderr, so keep it + # off the event loop. + await self.bot.loop.run_in_executor(None, stream.close) + self._stream = None # A near-instant end that wasn't a user action means the source # failed to load (bot-check, unavailable). Tell the user. if (not self._destroyed and not self._skip and not was_replay and played < 2.0): - track["error"] = await self.bot.loop.run_in_executor( - None, media.classify_stream_error, proc) + track["error"] = stream.classify_error() # cached by close() await self._safe_send(self._load_error_embed(track)) self.current = None except asyncio.CancelledError: @@ -290,8 +292,11 @@ def destroy(self) -> None: self._destroyed = True self.queue.clear() self.current = None - media.kill_stream(self._proc) - self._proc = None + if self._stream is not None: + # Reaping blocks briefly; hand it to a thread so teardown from a + # command never stalls the event loop. + self.bot.loop.run_in_executor(None, self._stream.close) + self._stream = None vc = self.voice if vc and vc.is_connected(): asyncio.ensure_future(vc.disconnect(force=True))