Fix channel worker stalling when stopping ffmpeg with unread stdout

CHANGELOG.md

@@ -20,6 +20,10 @@ All notable changes to this project are documented in this file.
 
 ### Fixed
 
+- Stopping ffmpeg (on every proactive URL refresh) no longer stalls. The worker stops reading ffmpeg's stdout when it
+  breaks out to refresh, and asyncio's `Process.wait()` does not return until every pipe reaches EOF, so `stop()` always
+  ran into the terminate timeout ("did not exit in time, killing it") and could leave a channel worker stuck for good,
+  silently ending that channel's stream. `stop()` now drains the leftover stdout while waiting for ffmpeg to exit.
 - `/stream/{channel_id}` now forces the connection closed after each response (`Connection: close`) instead of
   allowing keep-alive. Jellyfin's M3U tuner client (`SharedHttpStream`) was reusing a pooled keep-alive connection
   still mid-stream from an earlier request and reading raw MPEG-TS bytes as if they were a fresh HTTP response,

src/restreamer/ffmpeg_runner.py

@@ -74,6 +74,10 @@ class FfmpegStreamProcess:
         if self._process is None or self._process.returncode is not None:
             return
 
+        # wait() only returns once every pipe hits EOF, so unread stdout left over
+        # from the consumer breaking out of iter_chunks() must be drained, or it hangs forever.
+        stdout_drain_task = asyncio.create_task(self._drain_stdout())
+
         self._process.terminate()
         try:
             await asyncio.wait_for(self._process.wait(), timeout=FFMPEG_TERMINATE_TIMEOUT_SECONDS)
@@ -82,9 +86,16 @@ class FfmpegStreamProcess:
             self._process.kill()
             await self._process.wait()
 
+        await stdout_drain_task
         if self._stderr_task is not None:
             await self._stderr_task
 
+    async def _drain_stdout(self) -> None:
+        if self._process is None or self._process.stdout is None:
+            return
+        while await self._process.stdout.read(FFMPEG_READ_CHUNK_SIZE):
+            pass
+
     async def _drain_stderr(self) -> None:
         if self._process is None or self._process.stderr is None:
             return

tests/test_ffmpeg_runner.py

@@ -1,4 +1,10 @@
-from restreamer.ffmpeg_runner import build_ffmpeg_args
+import asyncio
+
+import pytest
+
+from restreamer import ffmpeg_runner
+from restreamer.constants import FFMPEG_TERMINATE_TIMEOUT_SECONDS
+from restreamer.ffmpeg_runner import FfmpegStreamProcess, build_ffmpeg_args
 from restreamer.resolver import ResolvedStream
 
 
@@ -42,3 +48,15 @@ def test_build_ffmpeg_args_maps_separate_video_and_audio_inputs() -> None:
     assert "-map" in args
     map_values = [args[i + 1] for i, a in enumerate(args) if a == "-map"]
     assert map_values == ["0:v:0", "1:a:0"]
+
+
+async def test_stop_returns_when_consumer_left_stdout_unread(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(ffmpeg_runner, "build_ffmpeg_args", lambda stream: ["yes"])
+    process = FfmpegStreamProcess(ResolvedStream(video_url="https://example.com/stream.m3u8"), "test")
+    await process.start()
+    async for _chunk in process.iter_chunks():
+        break
+
+    await asyncio.wait_for(process.stop(), timeout=FFMPEG_TERMINATE_TIMEOUT_SECONDS / 2)
+
+    assert await process.wait() != 0