fix(downloads): abort blocked model fetches on cancel (#888)

* fix(downloads): abort blocked model fetches on cancel

requests.get() has no way to abort while it is still resolving,
connecting, or waiting for response headers — which is what an
unreachable download server looks like. The cancel flag was only
checked between response chunks, so Cancel sat until the request
timed out. The blocking open now runs on a helper thread the caller
polls every 200 ms; on cancel it raises at once while the abandoned
attempt stays bounded by the existing timeouts.

The settings dialog's download workers also reported "Model ready to
use" whenever the apply returned True, but reconfigure() can no-op
when the selection matches live state — success without a download.
They now pass force_reinit so a reported success means the engine
actually initialized the downloaded model.

Fixes #679

* fix(downloads): bound cancel openers, close late responses

- Track blocked request opener threads and give each attempt its own stop
  event so a cancel-and-retry cannot stack unaccounted requests and a
  response landing after cancel is closed instead of leaked
- Catch requests.exceptions.RequestException in the opener instead of a
  bare Exception
- Close the streaming response on the mid-stream cancel path
- Make the blocked-open test helper cancellable and reap tracked openers;
  check each download worker call site for force_reinit separately; add
  missing return annotations on new test helpers and methods

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>

src/vocalinux/speech_recognition/recognition_manager.py

@@ -1160,6 +1160,10 @@ class SpeechRecognitionManager:
         # dialog both download in the background, and there is only one progress
         # callback and one engine configuration between them.
         self._download_claim = threading.Lock()
+        # Opener threads _stream_model_download leaves running past a cancel;
+        # each is bounded by the network timeout but tracked so repeated
+        # cancel-and-retry cannot stack unaccounted request threads.
+        self._download_openers: set[threading.Thread] = set()
         self._defer_download = defer_download
         self._model_initialized = False
         # True while auto-pause has unloaded the model for a configured app/game
@@ -2849,6 +2853,12 @@ class SpeechRecognitionManager:
     # without a timeout when the CDN is degraded (e.g. 504 / empty body).
     _MODEL_DOWNLOAD_TIMEOUT = (15, 120)
 
+    # How often a blocked request is re-checked for a pending cancel. requests
+    # has no cancellation API, so the blocking open runs on a helper thread:
+    # on cancel the caller stops waiting instead of riding out the whole
+    # network timeout (up to two minutes when a server never answers).
+    _DOWNLOAD_CANCEL_POLL_SECONDS = 0.2
+
     def _stream_model_download(self, url: str, dest_path: str) -> None:
         """Stream a model file from ``url`` to ``dest_path`` with progress.
 
@@ -2857,12 +2867,61 @@ class SpeechRecognitionManager:
         import requests
 
         logger.info(f"Downloading from {url}")
-        response = requests.get(
-            url,
-            stream=True,
-            timeout=self._MODEL_DOWNLOAD_TIMEOUT,
-            headers={"User-Agent": f"vocalinux/{__version__}"},
-        )
+
+        # While requests.get() is stuck resolving, connecting, or waiting for
+        # response headers — exactly what an unreachable server causes — it
+        # cannot see the cancel flag, and Cancel used to wait out the timeout.
+        # The attempt itself stays bounded by _MODEL_DOWNLOAD_TIMEOUT.
+        outcome: dict = {}
+        # Per-attempt stop signal: _download_cancelled is shared state that a
+        # retry resets, so the opener needs its own marker to notice the cancel
+        # that landed while it was still blocked in requests.get().
+        stop_open = threading.Event()
+
+        def _open() -> None:
+            try:
+                response = requests.get(
+                    url,
+                    stream=True,
+                    timeout=self._MODEL_DOWNLOAD_TIMEOUT,
+                    headers={"User-Agent": f"vocalinux/{__version__}"},
+                )
+            except requests.exceptions.RequestException as e:
+                # Surfaced on the calling thread below.
+                outcome["error"] = e
+                return
+            outcome["response"] = response
+            if stop_open.is_set():
+                # Cancel raced the response landing: the caller has already
+                # given up, so take the late response back and close it rather
+                # than leak an unconsumed socket.
+                outcome.pop("response", None)
+                response.close()
+
+        self._download_openers = {t for t in self._download_openers if t.is_alive()}
+        opener = threading.Thread(target=_open, daemon=True)
+        self._download_openers.add(opener)
+        opener.start()
+        while opener.is_alive():
+            if self._download_cancelled:
+                stop_open.set()
+                # If the response slipped through just as the cancel landed,
+                # the opener's own check may not have run yet; close whatever
+                # arrived late so the socket does not leak either way.
+                late = outcome.pop("response", None)
+                if late is not None:
+                    late.close()
+                logger.info("Download cancelled by user")
+                raise RuntimeError("Download cancelled")
+            opener.join(self._DOWNLOAD_CANCEL_POLL_SECONDS)
+        self._download_openers.discard(opener)
+
+        if "error" in outcome:
+            raise outcome["error"]
+        response = outcome.get("response")
+        if response is None:
+            raise RuntimeError(f"Model download from {url} produced no response")
+
         response.raise_for_status()
 
         content_type = (response.headers.get("content-type") or "").lower()
@@ -2883,6 +2942,7 @@ class SpeechRecognitionManager:
             for data in response.iter_content(chunk_size=chunk_size):
                 if self._download_cancelled:
                     logger.info("Download cancelled by user")
+                    response.close()
                     f.close()
                     if os.path.exists(dest_path):
                         os.remove(dest_path)

src/vocalinux/ui/settings_dialog.py

@@ -7598,7 +7598,9 @@ class SettingsDialog(Gtk.Dialog):
                         cancel_check_id = GLib.timeout_add(100, check_cancelled)
 
                         try:
-                            applied = self._apply_settings_internal(settings, raise_errors=True)
+                            applied = self._apply_settings_internal(
+                                settings, raise_errors=True, force_reinit=True
+                            )
                             if applied:
                                 GLib.idle_add(download_dialog.set_complete, True, "")
                                 GLib.idle_add(self._populate_model_options)
@@ -8192,7 +8194,9 @@ For now, the engine has been reverted to VOSK."""
                     cancel_check_id = GLib.timeout_add(100, check_cancelled)
 
                     try:
-                        applied = self._apply_settings_internal(settings, raise_errors=True)
+                        applied = self._apply_settings_internal(
+                            settings, raise_errors=True, force_reinit=True
+                        )
                         if applied:
                             GLib.idle_add(download_dialog.set_complete, True, "")
                         else:
@@ -8243,7 +8247,9 @@ For now, the engine has been reverted to VOSK."""
         dialog.run()
         dialog.destroy()
 
-    def _apply_settings_internal(self, settings: dict, raise_errors: bool = False) -> bool:
+    def _apply_settings_internal(
+        self, settings: dict, raise_errors: bool = False, force_reinit: bool = False
+    ) -> bool:
         """Internal method to apply settings.
 
         Args:
@@ -8252,6 +8258,10 @@ For now, the engine has been reverted to VOSK."""
                 The download threads pass True: their own handlers report the
                 failure through the progress dialog, and building a Gtk dialog
                 off the main loop is not safe anyway.
+            force_reinit: Re-initialize the engine even when the selection
+                matches its live state. The download threads pass True: they
+                only run because the model is missing on disk, and a no-op
+                reconfigure would report success for a download that never ran.
         """
         try:
             was_running = self.speech_engine.state != RecognitionState.IDLE
@@ -8262,7 +8272,7 @@ For now, the engine has been reverted to VOSK."""
             # Persist only once the engine really runs these settings: this call
             # downloads missing models, and a config saved up front would keep
             # pointing at a model that never made it to disk.
-            self.speech_engine.reconfigure(**settings)
+            self.speech_engine.reconfigure(force_reinit=force_reinit, **settings)
             self._save_selected_settings(settings)
 
             logger.info("Settings applied successfully.")

tests/test_download_error_reporting.py

@@ -98,12 +98,56 @@ def test_success_still_returns_true():
     )
 
 
-def test_both_download_threads_ask_for_the_failure():
+def test_both_download_threads_ask_for_the_failure() -> None:
     """Source guard: every _apply_settings_internal call made off the main loop
     passes raise_errors=True, so a regression cannot silently reintroduce the
-    swallowed-error path on a worker thread."""
+    swallowed-error path on a worker thread. The two download workers must
+    also force the engine re-init — their apply exists because a model is
+    missing, and a no-op reconfigure would report the green "Model ready to
+    use" for a download that never ran. Each call site is checked on its own
+    worker: a bare count would still pass if the flag moved between call
+    sites, and the already-downloaded worker must never force it."""
+    import ast
     import inspect
 
-    source = inspect.getsource(settings_dialog)
-    thread_calls = source.count("_apply_settings_internal(settings, raise_errors=True)")
-    assert thread_calls == 3
+    def worker_kwargs(worker: ast.FunctionDef) -> dict:
+        calls = [
+            node
+            for node in ast.walk(worker)
+            if isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and node.func.attr == "_apply_settings_internal"
+        ]
+        assert len(calls) == 1, "each worker applies settings exactly once"
+        return {
+            kw.arg: kw.value.value if isinstance(kw.value, ast.Constant) else None
+            for kw in calls[0].keywords
+            if kw.arg is not None
+        }
+
+    tree = ast.parse(inspect.getsource(settings_dialog))
+    functions = [node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)]
+
+    download_workers = [node for node in functions if node.name == "download_and_apply"]
+    assert len(download_workers) == 2
+    for worker in download_workers:
+        kwargs = worker_kwargs(worker)
+        assert kwargs.get("raise_errors") is True
+        assert kwargs.get("force_reinit") is True
+
+    already_downloaded = [node for node in functions if node.name == "apply_already_downloaded"]
+    assert len(already_downloaded) == 1
+    kwargs = worker_kwargs(already_downloaded[0])
+    assert kwargs.get("raise_errors") is True
+    assert "force_reinit" not in kwargs
+
+
+def test_force_reinit_reaches_reconfigure() -> None:
+    """The flag exists so a needed download cannot no-op its way to green."""
+    dialog = _dialog_stub()
+
+    assert (
+        SettingsDialog._apply_settings_internal(dialog, {"engine": "vosk"}, force_reinit=True)
+        is True
+    )
+    dialog.speech_engine.reconfigure.assert_called_once_with(force_reinit=True, engine="vosk")

tests/test_recognition_manager_downloads.py

@@ -11,6 +11,7 @@ import base64
 import json
 import os
 import sys
+import threading
 import time
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -334,6 +335,101 @@ class TestDownloadWhispercppModel:
 
         assert not os.path.exists(dest)
 
+    def test_stream_model_download_cancelled_while_connecting(self, tmp_path) -> None:
+        """Cancel must not wait out a blocked request open.
+
+        requests offers no way to abort a request stuck resolving, connecting,
+        or waiting on headers — exactly what an unreachable download server
+        does. The flag is only read once chunks flow, so Cancel used to ride
+        out the whole connect/read timeout (issue #679).
+        """
+        manager = _make_manager(engine="whisper_cpp")
+        dest = str(tmp_path / "never.bin")
+
+        mock_requests = MagicMock()
+        release_opener = threading.Event()
+
+        def blocked_get(*args, **kwargs) -> MagicMock:
+            release_opener.wait(60)  # a server that never answers
+            return MagicMock()
+
+        mock_requests.get.side_effect = blocked_get
+
+        def cancel_soon() -> None:
+            time.sleep(0.3)
+            manager._download_cancelled = True
+
+        with patch.dict("sys.modules", {"requests": mock_requests}):
+            threading.Thread(target=cancel_soon, daemon=True).start()
+            started = time.monotonic()
+            with pytest.raises(RuntimeError, match="cancelled"):
+                manager._stream_model_download("https://example.com/model.bin", dest)
+            assert time.monotonic() - started < 10
+
+        # The opener the cancel abandoned must be released and reaped, not left
+        # sleeping inside later tests.
+        release_opener.set()
+        for opener in manager._download_openers:
+            opener.join(timeout=5)
+        assert not any(t.is_alive() for t in manager._download_openers)
+
+        assert not os.path.exists(dest)
+
+    def test_stream_model_download_closes_late_response(self, tmp_path) -> None:
+        """A response landing after the cancel is closed, not left streaming."""
+        manager = _make_manager(engine="whisper_cpp")
+        dest = str(tmp_path / "late.bin")
+
+        mock_requests = MagicMock()
+        mock_response = MagicMock()
+        release_opener = threading.Event()
+
+        def slow_get(*args, **kwargs) -> MagicMock:
+            release_opener.wait(60)
+            return mock_response
+
+        mock_requests.get.side_effect = slow_get
+        manager._download_cancelled = True
+
+        with patch.dict("sys.modules", {"requests": mock_requests}):
+            with pytest.raises(RuntimeError, match="cancelled"):
+                manager._stream_model_download("https://example.com/model.bin", dest)
+            release_opener.set()
+            for opener in manager._download_openers:
+                opener.join(timeout=5)
+
+        mock_response.close.assert_called_once()
+        assert not os.path.exists(dest)
+
+    def test_stream_model_download_propagates_request_errors(self, tmp_path) -> None:
+        """A failed open re-raises the original error for callers to classify."""
+        manager = _make_manager(engine="whisper_cpp")
+        dest = str(tmp_path / "err.bin")
+
+        mock_requests = MagicMock()
+        mock_requests.get.side_effect = FakeRequestError("Connection refused")
+        mock_requests.exceptions.RequestException = FakeRequestError
+
+        with patch.dict("sys.modules", {"requests": mock_requests}):
+            with pytest.raises(FakeRequestError, match="Connection refused"):
+                manager._stream_model_download("https://example.com/model.bin", dest)
+
+        assert not os.path.exists(dest)
+
+    def test_stream_model_download_no_response(self, tmp_path) -> None:
+        """A helper that produced neither a response nor an error still fails."""
+        manager = _make_manager(engine="whisper_cpp")
+        dest = str(tmp_path / "none.bin")
+
+        mock_requests = MagicMock()
+        mock_requests.get.return_value = None
+
+        with patch.dict("sys.modules", {"requests": mock_requests}):
+            with pytest.raises(RuntimeError, match="no response"):
+                manager._stream_model_download("https://example.com/model.bin", dest)
+
+        assert not os.path.exists(dest)
+
     def test_stream_model_download_eta_minutes_and_empty_chunks(self, tmp_path):
         """Progress path covers multi-minute ETA and skips empty chunks."""
         manager = _make_manager(engine="whisper_cpp")