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