From ba721779c3d59a39d80642c0ddc6eb1fa0d99f62 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 29 Sep 2026 22:52:12 -0700 Subject: [PATCH] fix(downloads): abort blocked model fetches on cancel (#888) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 --- .../speech_recognition/recognition_manager.py | 72 ++++++++++++-- src/vocalinux/ui/settings_dialog.py | 18 +++- tests/test_download_error_reporting.py | 54 ++++++++++- tests/test_recognition_manager_downloads.py | 96 +++++++++++++++++++ 4 files changed, 225 insertions(+), 15 deletions(-) diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 1f868f1..54685d2 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/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) diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 81a3c4c..5b3831d 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/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.") diff --git a/tests/test_download_error_reporting.py b/tests/test_download_error_reporting.py index b202179..2d0cbac 100644 --- a/tests/test_download_error_reporting.py +++ b/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") diff --git a/tests/test_recognition_manager_downloads.py b/tests/test_recognition_manager_downloads.py index f10cf6c..25305dc 100644 --- a/tests/test_recognition_manager_downloads.py +++ b/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")