feat: implement patch to make Windows Popen support stopping live stream

app/features/ytdlp/patches.py

@@ -0,0 +1,98 @@
+import logging
+import subprocess
+import sys
+from typing import Any
+
+LOG: logging.Logger = logging.getLogger("ytdlp.utils")
+
+
+def patch_metadataparser() -> None:
+    """
+    Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues.
+    """
+    try:
+        from yt_dlp.postprocessor.metadataparser import MetadataParserPP
+        from yt_dlp.utils import Namespace
+    except Exception as exc:
+        LOG.warning(f"Unable to import yt_dlp metadata parser for patching: {exc!s}")
+        return
+
+    if getattr(MetadataParserPP.Actions, "_ytptube_patched", False):
+        return
+
+    class _ActionNS(Namespace):
+        _ACTIONS_STR: list[str] = []
+
+        @staticmethod
+        def _get_name(func) -> str | None:
+            if not callable(func):
+                return None
+
+            target = getattr(func, "__func__", func)
+            module_name = getattr(target, "__module__", None)
+            qual_name = getattr(target, "__qualname__", getattr(target, "__name__", None))
+
+            return f"{module_name}.{qual_name}" if module_name and qual_name else None
+
+        def __contains__(self, candidate: object) -> bool:
+            if candidate in self.__dict__.values():
+                return True
+
+            if func_name := _ActionNS._get_name(candidate):
+                if len(_ActionNS._ACTIONS_STR) < 1:
+                    _ActionNS._ACTIONS_STR.extend(
+                        [value for value in (_ActionNS._get_name(value) for value in self.__dict__.values()) if value]
+                    )
+
+                return func_name in _ActionNS._ACTIONS_STR
+
+            return False
+
+    actions_dict: dict[str, Any] = dict(MetadataParserPP.Actions.items_)
+    MetadataParserPP.Actions = _ActionNS(**actions_dict)
+    MetadataParserPP.Actions._ytptube_patched = True
+    LOG.debug("MetadataParserPP action namespace patch applied successfully.")
+
+
+def patch_windows_popen_wait() -> None:
+    if sys.platform != "win32":
+        return
+
+    try:
+        from yt_dlp.utils import Popen
+    except Exception as exc:
+        LOG.warning(f"Unable to import yt_dlp Popen for patching: {exc!s}")
+        return
+
+    if getattr(Popen, "_ytptube_wait_patched", False):
+        return
+
+    original_wait = Popen.wait
+
+    # Windows subprocess waits can swallow the synthetic interrupt we use to
+    # stop live downloads, especially while yt-dlp is blocked on ffmpeg.
+    def interruptible_wait(self, timeout=None):
+        if timeout is not None:
+            return original_wait(self, timeout=timeout)
+
+        while True:
+            try:
+                return original_wait(self, timeout=0.1)
+            except subprocess.TimeoutExpired:
+                continue
+
+    Popen.wait = interruptible_wait
+    Popen._ytptube_wait_patched = True
+    LOG.debug("yt_dlp Popen.wait Windows patch applied successfully.")
+
+
+def apply_ytdlp_patches() -> None:
+    try:
+        patch_metadataparser()
+    except Exception as exc:
+        LOG.debug("Metadata parser patch failed to apply: %s", exc)
+
+    try:
+        patch_windows_popen_wait()
+    except Exception as exc:
+        LOG.debug("Windows Popen wait patch failed to apply: %s", exc)

app/features/ytdlp/tests/test_ytdlp_module.py

@@ -1,5 +1,6 @@
 from unittest.mock import MagicMock, Mock, patch
 
+from app.features.ytdlp.patches import patch_windows_popen_wait
 from app.features.ytdlp.utils import _DATA
 from app.features.ytdlp.ytdlp import YTDLP, _ArchiveProxy, ytdlp_options
 
@@ -145,6 +146,61 @@ class TestYTDLP:
         assert isinstance(ytdlp.archive, _ArchiveProxy)
         assert not ytdlp.archive
 
+    @patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__")
+    def test_init_patches_windows_popen_wait_once(self, mock_super_init) -> None:
+        mock_super_init.return_value = None
+
+        class FakePopen:
+            def wait(self, timeout=None):
+                return timeout
+
+        with patch("app.features.ytdlp.patches.sys.platform", "win32"):
+            with patch("yt_dlp.utils.Popen", FakePopen):
+                YTDLP(params={})
+
+        assert getattr(FakePopen, "_ytptube_wait_patched", False) is True
+
+    def test_windows_wait_patch_uses_polling_for_blocking_wait(self) -> None:
+        calls: list[float | None] = []
+
+        class FakePopen:
+            _ytptube_wait_patched = False
+
+            def wait(self, timeout=None):
+                calls.append(timeout)
+                if len(calls) < 3:
+                    raise TimeoutError
+                return 0
+
+        with patch("app.features.ytdlp.patches.sys.platform", "win32"):
+            with (
+                patch("yt_dlp.utils.Popen", FakePopen),
+                patch("app.features.ytdlp.patches.subprocess.TimeoutExpired", TimeoutError),
+            ):
+                patch_windows_popen_wait()
+                result = FakePopen().wait()
+
+        assert result == 0
+        assert calls == [0.1, 0.1, 0.1]
+
+    def test_windows_wait_patch_preserves_explicit_timeout(self) -> None:
+        calls: list[float | None] = []
+
+        class FakePopen:
+            _ytptube_wait_patched = False
+
+            def wait(self, timeout=None):
+                calls.append(timeout)
+                return 0
+
+        with patch("app.features.ytdlp.patches.sys.platform", "win32"):
+            with patch("yt_dlp.utils.Popen", FakePopen):
+                patch_windows_popen_wait()
+                result = FakePopen().wait(timeout=5)
+
+        assert result == 0
+        assert calls == [5]
+
     @patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL._delete_downloaded_files")
     def test_delete_downloaded_files_skips_when_interrupted(self, mock_super_delete) -> None:
         """Test _delete_downloaded_files skips cleanup when _interrupted is True."""

app/features/ytdlp/utils.py

@@ -9,6 +9,7 @@ from email.utils import formatdate
 from pathlib import Path
 from typing import Any
 
+from app.features.ytdlp.patches import apply_ytdlp_patches
 from app.features.ytdlp.ytdlp import YTDLP
 from app.library.Utils import merge_dict, timed_lru_cache
 
@@ -143,52 +144,6 @@ class LogWrapper:
         self._log(logging.CRITICAL, msg, *args, **kwargs)
 
 
-def patch_metadataparser() -> None:
-    """
-    Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues.
-    """
-    try:
-        from yt_dlp.postprocessor.metadataparser import MetadataParserPP
-        from yt_dlp.utils import Namespace
-    except Exception as exc:
-        LOG.warning(f"Unable to import yt_dlp metadata parser for patching: {exc!s}")
-        return
-
-    if getattr(MetadataParserPP.Actions, "_ytptube_patched", False):
-        return
-
-    class _ActionNS(Namespace):
-        _ACTIONS_STR: list[str] = []
-
-        @staticmethod
-        def _get_name(func) -> str | None:
-            if not callable(func):
-                return None
-
-            target = getattr(func, "__func__", func)
-            module_name = getattr(target, "__module__", None)
-            qual_name = getattr(target, "__qualname__", getattr(target, "__name__", None))
-
-            return f"{module_name}.{qual_name}" if module_name and qual_name else None
-
-        def __contains__(self, candidate: object) -> bool:
-            if candidate in self.__dict__.values():
-                return True
-
-            if func_name := _ActionNS._get_name(candidate):
-                if len(_ActionNS._ACTIONS_STR) < 1:
-                    _ActionNS._ACTIONS_STR.extend([_ActionNS._get_name(value) for value in self.__dict__.values()])
-
-                return func_name in _ActionNS._ACTIONS_STR
-
-            return False
-
-    actions_dict: dict[str, Any] = dict(MetadataParserPP.Actions.items_)
-    MetadataParserPP.Actions = _ActionNS(**actions_dict)
-    MetadataParserPP.Actions._ytptube_patched = True
-    LOG.debug("MetadataParserPP action namespace patch applied successfully.")
-
-
 def arg_converter(
     args: str,
     level: int | bool | None = None,
@@ -222,10 +177,7 @@ def arg_converter(
         finally:
             yt_dlp.options.create_parser = create_parser
 
-    try:
+    apply_ytdlp_patches()
-        patch_metadataparser()
-    except Exception as exc:
-        LOG.debug("Metadata parser patch failed to apply: %s", exc)
 
     default_opts = _default_opts([]).ydl_opts
 

app/features/ytdlp/ytdlp.py

@@ -5,6 +5,7 @@ from typing import Any
 import yt_dlp
 from yt_dlp.utils import make_archive_id
 
+from app.features.ytdlp.patches import apply_ytdlp_patches
 from app.library.cf_solver_handler import set_cf_handler
 
 
@@ -52,13 +53,15 @@ class YTDLP(yt_dlp.YoutubeDL):
     _registered = False
 
     def __init__(self, params=None, auto_init=True):
+        apply_ytdlp_patches()
+
         # Avoid yt-dlp preloading the archive file by stripping the param first
         orig_file = None
-        patched_params = None
+        patched_params: dict[str, Any] | None = None
         if params is not None:
             try:
                 orig_file: str | None = params.get("download_archive")
-                patched_params: dict = dict(params)
+                patched_params = dict(params)
                 if "download_archive" in patched_params:
                     patched_params.pop("download_archive", None)
             except Exception:

app/library/downloads/core.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import _thread
 import asyncio
 import logging
 import os
@@ -217,7 +218,7 @@ class Download:
                     if "posix" == os.name:
                         os.kill(os.getpid(), signal.SIGINT)
                     else:
-                        signal.raise_signal(signal.SIGINT)
+                        _thread.interrupt_main()
 
                 threading.Thread(target=trigger_live_cancel, name=f"cancel-watch-{self.id}", daemon=True).start()
 

app/tests/test_download.py

@@ -371,20 +371,32 @@ class TestDownloadFlow:
             def process_ie_result(self, ie_result, download):
                 return ie_result, download
 
+            def download(self, url_list):
+                return 0
+
         signal_mock = Mock()
-        thread_instance = Mock(start=Mock())
+        thread_instances: list[Mock] = []
-        thread_mock = Mock(return_value=thread_instance)
+
+        def build_thread(*_args, **_kwargs):
+            thread = Mock(start=Mock())
+            thread_instances.append(thread)
+            return thread
+
+        thread_mock = Mock(side_effect=build_thread)
         monkeypatch.setattr("app.library.downloads.core.YTDLP", FakeYTDLP)
         monkeypatch.setattr("app.library.downloads.core.signal.signal", signal_mock)
         monkeypatch.setattr("app.library.downloads.core.threading.Thread", thread_mock)
 
         download._download()
 
-        thread_mock.assert_called_once()
-        thread_instance.start.assert_called_once()
         signal_mock.assert_any_call(signal.SIGINT, signal.default_int_handler)
 
-        target = thread_mock.call_args.kwargs["target"]
+        live_cancel_thread = next(
+            call for call in thread_mock.call_args_list if call.kwargs.get("name", "").startswith("cancel-watch-")
+        )
+        live_cancel_thread_index = thread_mock.call_args_list.index(live_cancel_thread)
+        thread_instances[live_cancel_thread_index].start.assert_called_once()
+        target = live_cancel_thread.kwargs["target"]
         ydl = created_ydl[0]
 
         if "posix" == os.name:
@@ -395,9 +407,9 @@ class TestDownloadFlow:
                 target()
                 mock_kill.assert_called_once_with(12345, signal.SIGINT)
         else:
-            with patch("app.library.downloads.core.signal.raise_signal") as mock_raise_signal:
+            with patch("app.library.downloads.core._thread.interrupt_main") as mock_interrupt_main:
                 target()
-                mock_raise_signal.assert_called_once_with(signal.SIGINT)
+                mock_interrupt_main.assert_called_once_with()
 
         assert ydl._interrupted is True
         ydl.to_screen.assert_called_once_with("[info] Interrupt received, exiting cleanly...")

pyproject.toml

@@ -194,6 +194,7 @@ testpaths = ["app/tests", "app/features"]
 addopts = "-v --tb=short"
 filterwarnings = [
   "ignore:Parsing dates involving a day of month without a year:DeprecationWarning",
+  "ignore:This process \\(pid\\=.*\\) is multi-threaded, use of fork\\(\\) may lead to deadlocks in the child\\.:DeprecationWarning",
 ]
 
 [dependency-groups]