feat(downloader): reuse qBittorrent session across operations (#1039, #900)

Cache one concrete client per connection-settings key at module level and reuse
it across `async with DownloadClient()` blocks instead of logging in on enter and
out on exit every operation (~1/min from the rename loop). QbDownloader.auth is
now idempotent — a live, authenticated client short-circuits without re-posting
login — and data operations go through a _request wrapper that re-authenticates
once and retries on a 403 (expired session). __aexit__ no longer logs out; a new
shutdown() module function tears the cached client down (wired up by the Phase 5
composition root). MockDownloader/Aria2Downloader gain the same auth idempotency.
When connection settings change, the stale client is retired and closed lazily on
the next enter. A failed fresh login still closes the pool and drops the cache
entry before raising (leak fix, #1043).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014w1Z6Nxy6XTRgkFXqPr9Zh

backend/src/module/downloader/__init__.py

@@ -3,11 +3,12 @@ from .base import (
     DownloaderCapabilities,
     DownloaderClient,
 )
-from .download_client import DownloadClient
+from .download_client import DownloadClient, shutdown
 
 __all__ = [
     "CoreDownloaderClient",
     "DownloadClient",
     "DownloaderCapabilities",
     "DownloaderClient",
+    "shutdown",
 ]

backend/src/module/downloader/client/aria2_downloader.py

@@ -24,6 +24,7 @@ class Aria2Downloader:
         self.host = host
         self.secret = password
         self._client: httpx.AsyncClient | None = None
+        self._authed = False
         self._rpc_url = f"{host}/jsonrpc"
         self._id = 0
 
@@ -46,13 +47,17 @@ class Aria2Downloader:
         return result.get("result")
 
     async def auth(self, retry=3):
-        self._client = httpx.AsyncClient(
-            timeout=httpx.Timeout(connect=3.1, read=10.0, write=10.0, pool=10.0)
-        )
+        if self._client is not None and self._authed:
+            return True
+        if self._client is None:
+            self._client = httpx.AsyncClient(
+                timeout=httpx.Timeout(connect=3.1, read=10.0, write=10.0, pool=10.0)
+            )
         times = 0
         while times < retry:
             try:
                 await self._call("getVersion")
+                self._authed = True
                 return True
             except Exception as e:
                 logger.warning(
@@ -63,6 +68,7 @@ class Aria2Downloader:
         return False
 
     async def logout(self):
+        self._authed = False
         if self._client:
             await self._client.aclose()
             self._client = None

backend/src/module/downloader/client/mock_downloader.py

@@ -31,6 +31,7 @@ class MockDownloader:
         self._rules: dict[str, dict] = {}
         self._feeds: dict[str, dict] = {}
         self._categories: set[str] = {"Bangumi", "BangumiCollection"}
+        self._authed = False
         self._prefs = {
             "save_path": "/tmp/mock-downloads",
             "rss_auto_downloading_enabled": True,
@@ -41,10 +42,14 @@ class MockDownloader:
         logger.debug("[MockDownloader] Initialized")
 
     async def auth(self, retry=3) -> bool:
+        # No real session; idempotent by construction, kept for parity with
+        # QbDownloader so session-reuse tests stay meaningful.
+        self._authed = True
         logger.debug("[MockDownloader] Auth successful (mocked)")
         return True
 
     async def logout(self):
+        self._authed = False
         logger.debug("[MockDownloader] Logout (mocked)")
 
     async def check_host(self) -> bool:

backend/src/module/downloader/client/qb_downloader.py

@@ -29,13 +29,12 @@ class QbDownloader:
         self.password = password
         self.ssl = ssl
         self._client: httpx.AsyncClient | None = None
+        self._authed = False
 
     def _url(self, endpoint: str) -> str:
         return f"{self.host}/api/v2/{endpoint}"
 
-    async def auth(self, retry=3):
-        times = 0
-        use_https = self.host.startswith("https://")
+    def _new_client(self) -> httpx.AsyncClient:
         timeout = httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=10.0)
         # Keepalive_expiry keeps idle TCP sockets short-lived so they can't
         # outlive a proxy / NAS idle-reap, which would otherwise surface as
@@ -49,7 +48,17 @@ class QbDownloader:
         )
         # Never verify certificates - self-signed certs are the norm for
         # home-server / NAS / Docker qBittorrent setups.
-        self._client = httpx.AsyncClient(timeout=timeout, limits=limits, verify=False)
+        return httpx.AsyncClient(timeout=timeout, limits=limits, verify=False)
+
+    async def auth(self, retry=3):
+        # Session reuse: a live, already-authenticated client short-circuits so
+        # repeated operations don't re-login every cycle (#1039 / #900).
+        if self._client is not None and self._authed:
+            return True
+        times = 0
+        use_https = self.host.startswith("https://")
+        if self._client is None:
+            self._client = self._new_client()
         while times < retry:
             try:
                 resp = await self._client.post(
@@ -64,6 +73,7 @@ class QbDownloader:
                 if (
                     resp.status_code == 200 and resp.text.startswith("Ok")
                 ) or resp.status_code == 204:
+                    self._authed = True
                     return True
                 elif resp.status_code == 403:
                     logger.error("Login refused by qBittorrent Server")
@@ -98,6 +108,26 @@ class QbDownloader:
                 break
         return False
 
+    async def _request(self, method: str, endpoint: str, **kwargs) -> httpx.Response:
+        """Issue a request, re-authenticating once if the session expired.
+
+        qBittorrent answers 403 when the session cookie is no longer valid
+        (e.g. the server restarted or reaped the session). Since we now keep one
+        long-lived session, force a single re-login and retry the request once.
+        """
+        resp = await self._client.request(method, self._url(endpoint), **kwargs)
+        if resp.status_code == 403:
+            self._authed = False
+            await self.auth()
+            resp = await self._client.request(method, self._url(endpoint), **kwargs)
+        return resp
+
+    async def _get(self, endpoint: str, **kwargs) -> httpx.Response:
+        return await self._request("GET", endpoint, **kwargs)
+
+    async def _post(self, endpoint: str, **kwargs) -> httpx.Response:
+        return await self._request("POST", endpoint, **kwargs)
+
     async def logout(self):
         if self._client:
             try:
@@ -110,10 +140,11 @@ class QbDownloader:
                 logger.debug("[Downloader] Logout request failed (non-critical): %s", e)
             await self._client.aclose()
             self._client = None
+        self._authed = False
 
     async def check_host(self):
         try:
-            resp = await self._client.get(self._url("app/version"))
+            resp = await self._get("app/version")
             return resp.status_code == 200
         except (httpx.ConnectError, httpx.RequestError):
             return False
@@ -123,20 +154,20 @@ class QbDownloader:
 
     @qb_connect_failed_wait
     async def prefs_init(self, prefs):
-        resp = await self._client.post(
-            self._url("app/setPreferences"),
+        resp = await self._post(
+            "app/setPreferences",
             data={"json": json.dumps(prefs)},
         )
         return resp
 
     @qb_connect_failed_wait
     async def get_app_prefs(self):
-        resp = await self._client.get(self._url("app/preferences"))
+        resp = await self._get("app/preferences")
         return resp.json()
 
     async def add_category(self, category):
-        await self._client.post(
-            self._url("torrents/createCategory"),
+        await self._post(
+            "torrents/createCategory",
             data={"category": category, "savePath": ""},
         )
 
@@ -149,14 +180,12 @@ class QbDownloader:
             params["category"] = category
         if tag:
             params["tag"] = tag
-        resp = await self._client.get(self._url("torrents/info"), params=params)
+        resp = await self._get("torrents/info", params=params)
         return resp.json()
 
     @qb_connect_failed_wait
     async def torrents_files(self, torrent_hash: str):
-        resp = await self._client.get(
-            self._url("torrents/files"), params={"hash": torrent_hash}
-        )
+        resp = await self._get("torrents/files", params={"hash": torrent_hash})
         return resp.json()
 
     async def add_torrents(
@@ -195,8 +224,8 @@ class QbDownloader:
         max_retries = 3
         for attempt in range(max_retries):
             try:
-                resp = await self._client.post(
-                    self._url("torrents/add"),
+                resp = await self._post(
+                    "torrents/add",
                     data=data,
                     files=files if files else None,
                 )
@@ -214,15 +243,15 @@ class QbDownloader:
                     raise
 
     async def get_torrents_by_tag(self, tag: str) -> list[dict]:
-        resp = await self._client.get(self._url("torrents/info"), params={"tag": tag})
+        resp = await self._get("torrents/info", params={"tag": tag})
         return resp.json()
 
     async def torrents_delete(self, hash, delete_files: bool = True) -> bool:
         # qBittorrent expects one pipe-joined "hashes" field; a Python list would
         # be form-encoded as repeated fields and silently ignored (#1046).
         hashes = "|".join(hash) if isinstance(hash, (list, tuple)) else hash
-        resp = await self._client.post(
-            self._url("torrents/delete"),
+        resp = await self._post(
+            "torrents/delete",
             data={"hashes": hashes, "deleteFiles": str(delete_files).lower()},
         )
         if resp.status_code != 200:
@@ -235,14 +264,14 @@ class QbDownloader:
         return True
 
     async def torrents_pause(self, hashes: str):
-        await self._client.post(
-            self._url("torrents/pause"),
+        await self._post(
+            "torrents/pause",
             data={"hashes": hashes},
         )
 
     async def torrents_resume(self, hashes: str):
-        await self._client.post(
-            self._url("torrents/resume"),
+        await self._post(
+            "torrents/resume",
             data={"hashes": hashes},
         )
 
@@ -250,8 +279,8 @@ class QbDownloader:
         self, torrent_hash, old_path, new_path, verify: bool = True
     ) -> bool:
         try:
-            resp = await self._client.post(
-                self._url("torrents/renameFile"),
+            resp = await self._post(
+                "torrents/renameFile",
                 data={"hash": torrent_hash, "oldPath": old_path, "newPath": new_path},
             )
             if resp.status_code == 409:
@@ -291,75 +320,73 @@ class QbDownloader:
             return False
 
     async def rss_add_feed(self, url, item_path):
-        resp = await self._client.post(
-            self._url("rss/addFeed"),
+        resp = await self._post(
+            "rss/addFeed",
             data={"url": url, "path": item_path},
         )
         if resp.status_code == 409:
             logger.warning(f"[Downloader] RSS feed {url} already exists")
 
     async def rss_remove_item(self, item_path):
-        resp = await self._client.post(
-            self._url("rss/removeItem"),
+        resp = await self._post(
+            "rss/removeItem",
             data={"path": item_path},
         )
         if resp.status_code == 409:
             logger.warning(f"[Downloader] RSS item {item_path} does not exist")
 
     async def rss_get_feeds(self):
-        resp = await self._client.get(self._url("rss/items"))
+        resp = await self._get("rss/items")
         return resp.json()
 
     async def rss_set_rule(self, rule_name, rule_def):
-        await self._client.post(
-            self._url("rss/setRule"),
+        await self._post(
+            "rss/setRule",
             data={"ruleName": rule_name, "ruleDef": json.dumps(rule_def)},
         )
 
     async def move_torrent(self, hashes, new_location):
-        await self._client.post(
-            self._url("torrents/setLocation"),
+        await self._post(
+            "torrents/setLocation",
             data={"hashes": hashes, "location": new_location},
         )
 
     async def get_download_rule(self):
-        resp = await self._client.get(self._url("rss/rules"))
+        resp = await self._get("rss/rules")
         return resp.json()
 
     async def get_torrent_path(self, _hash):
-        resp = await self._client.get(
-            self._url("torrents/info"), params={"hashes": _hash}
-        )
+        resp = await self._get("torrents/info", params={"hashes": _hash})
         torrents = resp.json()
         if torrents:
             return torrents[0].get("save_path", "")
         return ""
 
     async def set_category(self, _hash, category):
-        resp = await self._client.post(
-            self._url("torrents/setCategory"),
+        resp = await self._post(
+            "torrents/setCategory",
             data={"hashes": _hash, "category": category},
         )
         if resp.status_code == 409:
             logger.warning(f"[Downloader] Category {category} does not exist")
             await self.add_category(category)
-            await self._client.post(
-                self._url("torrents/setCategory"),
+            await self._post(
+                "torrents/setCategory",
                 data={"hashes": _hash, "category": category},
             )
 
     async def check_connection(self):
-        resp = await self._client.get(self._url("app/version"))
+        resp = await self._get("app/version")
         return resp.text
 
     async def remove_rule(self, rule_name):
-        await self._client.post(
-            self._url("rss/removeRule"),
+        await self._post(
+            "rss/removeRule",
             data={"ruleName": rule_name},
         )
 
     async def add_tag(self, _hash, tag):
-        await self._client.post(
-            self._url("torrents/addTags"),
+        await self._post(
+            "torrents/addTags",
             data={"hashes": _hash, "tags": tag},
         )

backend/src/module/downloader/download_client.py

@@ -10,21 +10,79 @@ from .rules import build_rss_rule
 
 logger = logging.getLogger(__name__)
 
+
+# ---------------------------------------------------------------------------
+# Module-level concrete-client cache (session reuse, #1039 / #900)
+#
+# Every ``async with DownloadClient()`` used to spin up a fresh concrete client
+# and log in on enter / log out on exit -- one qB login+logout per operation,
+# roughly once a minute from the rename loop. The cache keeps a single concrete
+# client alive across operations, keyed by the connection-relevant settings.
+# When those settings change the old client is retired and re-closed lazily
+# inside the next ``__aenter__`` (so aclose runs in an async context).
+# ---------------------------------------------------------------------------
+_client_cache: tuple[tuple, object] | None = None
+_stale_client: object | None = None
+
 # Warn at most once per (client type, operation) when a backend cannot perform
 # an operation, so aria2 users are not spammed every rename cycle.
 _warned_unsupported: set[tuple[str, str]] = set()
 
 
+def _settings_key() -> tuple:
+    d = settings.downloader
+    return (d.type, d.host, d.username, d.password, d.ssl)
+
+
+def _reset_client_cache() -> None:
+    """Drop the cached/stale concrete clients (used by tests)."""
+    global _client_cache, _stale_client
+    _client_cache = None
+    _stale_client = None
+
+
+async def shutdown() -> None:
+    """Log out and close the cached concrete client.
+
+    Not called anywhere yet -- the Phase 5 composition root will own the
+    downloader lifecycle and invoke this on application shutdown.
+    """
+    global _client_cache, _stale_client
+    clients = []
+    if _stale_client is not None:
+        clients.append(_stale_client)
+    if _client_cache is not None:
+        clients.append(_client_cache[1])
+    _client_cache = None
+    _stale_client = None
+    for client in clients:
+        try:
+            await client.logout()
+        except Exception as e:  # pragma: no cover - best-effort cleanup
+            logger.debug("[Downloader] Error closing client on shutdown: %s", e)
+
+
 class DownloadClient:
     """Unified async download client.
 
     Wraps qBittorrent, Aria2, or MockDownloader behind a common interface.
     Intended to be used as an async context manager; authentication is
-    performed on ``__aenter__`` and the session is closed on ``__aexit__``.
+    performed on ``__aenter__``. The concrete client's session is reused across
+    context-manager blocks (see the module-level cache above) and only torn
+    down by :func:`shutdown`.
     """
 
     def __init__(self):
-        self.client = self.__getClient()
+        global _client_cache, _stale_client
+        key = _settings_key()
+        if _client_cache is not None and _client_cache[0] == key:
+            self.client = _client_cache[1]
+        else:
+            if _client_cache is not None:
+                # Settings changed: retire the previous client, close it later.
+                _stale_client = _client_cache[1]
+            self.client = self.__getClient()
+            _client_cache = (key, self.client)
         self.authed = False
 
     @staticmethod
@@ -70,22 +128,30 @@ class DownloadClient:
         return False
 
     async def __aenter__(self):
+        global _stale_client, _client_cache
+        if _stale_client is not None:
+            stale, _stale_client = _stale_client, None
+            try:
+                await stale.logout()
+            except Exception as e:  # pragma: no cover - best-effort cleanup
+                logger.debug("[Downloader] Error closing stale client: %s", e)
         if not self.authed:
             await self.auth()
             if not self.authed:
                 # __aexit__ never runs when we raise here, so close the
                 # concrete client's connection pool now or it leaks on every
-                # failed connect (#1043).
+                # failed connect (#1043). Also drop it from the cache so the
+                # next construction rebuilds a fresh client.
                 await self.client.logout()
+                if _client_cache is not None and _client_cache[1] is self.client:
+                    _client_cache = None
                 raise ConnectionError("Download client authentication failed")
-        else:
-            logger.error("[Downloader] Already authed.")
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        if self.authed:
-            await self.client.logout()
-            self.authed = False
+        # The concrete session is reused across operations; do NOT log out here.
+        # Teardown is deferred to shutdown() (Phase 5 composition root).
+        self.authed = False
 
     async def auth(self):
         self.authed = await self.client.auth()

backend/src/test/conftest.py

@@ -32,6 +32,21 @@ def _clear_bangumi_cache():
     _invalidate_bangumi_cache()
 
 
+@pytest.fixture(autouse=True)
+def _reset_downloader_client_cache():
+    """Clear the module-level concrete-client cache between tests.
+
+    DownloadClient reuses one concrete client across context-manager blocks
+    (session reuse, #1039 / #900); the cache would otherwise leak a mock from
+    one test into the next.
+    """
+    from module.downloader.download_client import _reset_client_cache
+
+    _reset_client_cache()
+    yield
+    _reset_client_cache()
+
+
 @pytest.fixture
 def db_engine():
     """Create an in-memory SQLite engine for testing."""

backend/src/test/test_download_client.py

@@ -365,9 +365,15 @@ class TestContextManagerAuth:
         assert result is download_client
         assert download_client.authed is True
 
-    async def test_aexit_calls_logout_when_authed(self, download_client, mock_qb_client):
-        """__aexit__ calls logout and resets authed when session was active."""
+    async def test_aexit_reuses_session_without_logout(
+        self, download_client, mock_qb_client
+    ):
+        """__aexit__ resets authed but keeps the session (no logout).
+
+        The concrete client is reused across operations (#1039 / #900);
+        teardown is deferred to shutdown().
+        """
         download_client.authed = True
         await download_client.__aexit__(None, None, None)
-        mock_qb_client.logout.assert_called_once()
+        mock_qb_client.logout.assert_not_called()
         assert download_client.authed is False

backend/src/test/test_download_session_reuse.py

@@ -0,0 +1,148 @@
+"""Regression tests for qBittorrent session reuse (#1039 / #900).
+
+The concrete client is cached across ``async with DownloadClient()`` blocks so
+we log in once, re-authenticate only when the session expires, and rebuild when
+connection settings change.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from module.downloader.client.qb_downloader import QbDownloader
+from module.downloader.download_client import DownloadClient
+
+
+def _patch_qb_settings(mock_settings, host="localhost:8080"):
+    mock_settings.downloader.type = "qbittorrent"
+    mock_settings.downloader.host = host
+    mock_settings.downloader.username = "admin"
+    mock_settings.downloader.password = "admin"
+    mock_settings.downloader.ssl = False
+
+
+class _FakeHTTP:
+    """Minimal httpx.AsyncClient stand-in counting logins and data requests."""
+
+    def __init__(self, login_calls: list, request_calls: list):
+        self._login_calls = login_calls
+        self._request_calls = request_calls
+
+    async def post(self, url, data=None):
+        # auth/login and auth/logout both arrive via post(); count logins only.
+        if url.endswith("auth/login"):
+            self._login_calls.append(url)
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.text = "Ok."
+        return resp
+
+    async def request(self, method, url, **kwargs):
+        self._request_calls.append((method, url))
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.text = "Ok."
+        resp.json = lambda: []
+        return resp
+
+    async def aclose(self):
+        pass
+
+
+class TestSessionReuse:
+    async def test_two_blocks_share_one_login(self):
+        """Two sequential ``async with DownloadClient()`` blocks log in once."""
+        login_calls: list = []
+        request_calls: list = []
+
+        def make_http(*args, **kwargs):
+            return _FakeHTTP(login_calls, request_calls)
+
+        with patch("module.downloader.download_client.settings") as mock_settings:
+            _patch_qb_settings(mock_settings)
+            with patch(
+                "module.downloader.client.qb_downloader.httpx.AsyncClient",
+                side_effect=make_http,
+            ):
+                async with DownloadClient():
+                    pass
+                async with DownloadClient():
+                    pass
+
+        assert len(login_calls) == 1
+
+    async def test_expired_session_reauths_once_and_retries(self):
+        """A 403 mid-operation forces exactly one re-login, then the op succeeds."""
+        qb = QbDownloader("localhost:8080", "admin", "admin", ssl=False)
+        login_calls: list = []
+        request_calls: list = []
+
+        class _Fake403Once:
+            async def post(self, url, data=None):
+                login_calls.append(url)
+                resp = MagicMock()
+                resp.status_code = 200
+                resp.text = "Ok."
+                return resp
+
+            async def request(self, method, url, **kwargs):
+                request_calls.append((method, url))
+                resp = MagicMock()
+                # First data request hits an expired session; retry succeeds.
+                resp.status_code = 403 if len(request_calls) == 1 else 200
+                resp.json = lambda: []
+                return resp
+
+            async def aclose(self):
+                pass
+
+        with patch(
+            "module.downloader.client.qb_downloader.httpx.AsyncClient",
+            return_value=_Fake403Once(),
+        ):
+            await qb.auth()
+            resp = await qb._get("torrents/info")
+
+        assert resp.status_code == 200
+        assert len(login_calls) == 2  # initial login + one forced re-login
+        assert len(request_calls) == 2  # original request + one retry
+
+    async def test_settings_change_creates_new_client(self):
+        """Changing the connection settings retires the cached client."""
+        with patch("module.downloader.download_client.settings") as mock_settings:
+            _patch_qb_settings(mock_settings, host="host-a:8080")
+            first = DownloadClient().client
+            reused = DownloadClient().client  # same settings -> same client
+            _patch_qb_settings(mock_settings, host="host-b:8080")
+            rebuilt = DownloadClient().client  # settings changed -> new client
+
+        assert first is reused
+        assert first is not rebuilt
+
+    async def test_no_login_idempotency_on_cheap_reauth(self):
+        """QbDownloader.auth() short-circuits when already authenticated."""
+        qb = QbDownloader("localhost:8080", "admin", "admin", ssl=False)
+        login_calls: list = []
+
+        def make_http(*args, **kwargs):
+            client = MagicMock()
+
+            async def _post(url, data=None):
+                login_calls.append(url)
+                resp = MagicMock()
+                resp.status_code = 200
+                resp.text = "Ok."
+                return resp
+
+            client.post = _post
+            client.aclose = AsyncMock()
+            return client
+
+        with patch(
+            "module.downloader.client.qb_downloader.httpx.AsyncClient",
+            side_effect=make_http,
+        ):
+            assert await qb.auth() is True
+            assert await qb.auth() is True  # cheap, no second login
+
+        assert len(login_calls) == 1

backend/src/test/test_qb_downloader.py

@@ -625,12 +625,13 @@ class TestTorrentsDelete:
         qb._client = AsyncMock()
         mock_resp = MagicMock()
         mock_resp.status_code = 200
-        qb._client.post = AsyncMock(return_value=mock_resp)
+        # Data ops now go through _request -> _client.request (session reuse).
+        qb._client.request = AsyncMock(return_value=mock_resp)
 
         result = await qb.torrents_delete(["aaa", "bbb"], delete_files=True)
 
         assert result is True
-        sent = qb._client.post.call_args.kwargs["data"]
+        sent = qb._client.request.call_args.kwargs["data"]
         assert sent["hashes"] == "aaa|bbb"
         assert sent["deleteFiles"] == "true"
 
@@ -639,8 +640,8 @@ class TestTorrentsDelete:
         qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False)
         qb._client = AsyncMock()
         mock_resp = MagicMock()
-        mock_resp.status_code = 403
-        qb._client.post = AsyncMock(return_value=mock_resp)
+        mock_resp.status_code = 500
+        qb._client.request = AsyncMock(return_value=mock_resp)
 
         result = await qb.torrents_delete("aaa", delete_files=True)