fix(network): harden shared httpx client against stale connections (#1018)

RSS loop runs every 900s but reuses the same httpx.AsyncClient connection pool, well past server keep-alive expiry (60-120s). Reusing dead sockets produced ConnectTimeout errors over time.

- Configure httpx.Limits with keepalive_expiry=60s so idle connections are dropped proactively
- reset_shared_client() called on retry paths to force a fresh pool after ConnectError
- asyncio.Semaphore(5) in refresh_rss() caps concurrent RSS fetches to avoid tripping site rate limits

Closes #1008
Related #1010, #742, #701

Author: zhushanwen321

backend/src/module/network/request_url.py

@@ -12,6 +12,15 @@
 _shared_client: httpx.AsyncClient | None = None
 _shared_client_proxy_key: str | None = None
 
+# RSS 循环间隔 900s， 远超服务端 keep-alive 超时（60-120s）
+# keepalive_expiry=60 让空闲连接在过期前主动丢弃，避免复用过期连接
+# max_connections=20 足够覆盖典型订阅数量
+_CONNECTION_LIMITS = httpx.Limits(
+    max_keepalive_connections=5,
+    max_connections=20,
+    keepalive_expiry=60.0,
+)
+
 
 def _proxy_config_key() -> str:
     if settings.proxy.enable:
@@ -33,22 +42,37 @@ async def get_shared_client() -> httpx.AsyncClient:
                 proxy_url = f"http://{settings.proxy.username}:{settings.proxy.password}@{settings.proxy.host}:{settings.proxy.port}"
             else:
                 proxy_url = f"http://{settings.proxy.host}:{settings.proxy.port}"
-            _shared_client = httpx.AsyncClient(proxy=proxy_url, timeout=timeout)
+            _shared_client = httpx.AsyncClient(
+                proxy=proxy_url, timeout=timeout, limits=_CONNECTION_LIMITS
+            )
         elif settings.proxy.type == "socks5":
             if settings.proxy.username:
                 socks_url = f"socks5://{settings.proxy.username}:{settings.proxy.password}@{settings.proxy.host}:{settings.proxy.port}"
             else:
                 socks_url = f"socks5://{settings.proxy.host}:{settings.proxy.port}"
             transport = AsyncProxyTransport.from_url(socks_url, rdns=True)
-            _shared_client = httpx.AsyncClient(transport=transport, timeout=timeout)
+            _shared_client = httpx.AsyncClient(
+                transport=transport, timeout=timeout, limits=_CONNECTION_LIMITS
+            )
         else:
-            _shared_client = httpx.AsyncClient(timeout=timeout)
+            _shared_client = httpx.AsyncClient(
+                timeout=timeout, limits=_CONNECTION_LIMITS
+            )
     else:
-        _shared_client = httpx.AsyncClient(timeout=timeout)
+        _shared_client = httpx.AsyncClient(timeout=timeout, limits=_CONNECTION_LIMITS)
     _shared_client_proxy_key = current_key
     return _shared_client
 
 
+async def reset_shared_client():
+    """关闭并清除共享客户端，下次请求时自动创建新连接池。"""
+    global _shared_client, _shared_client_proxy_key
+    if _shared_client is not None:
+        await _shared_client.aclose()
+    _shared_client = None
+    _shared_client_proxy_key = None
+
+
 class RequestURL:
     # More complete User-Agent to avoid Cloudflare blocking
     DEFAULT_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
@@ -91,6 +115,9 @@ async def get_url(self, url, retry=3):
                 try_time += 1
                 if try_time >= retry:
                     break
+                # 连接错误时重建客户端以清除过期连接
+                await reset_shared_client()
+                self._client = await get_shared_client()
                 await asyncio.sleep(5)
             except Exception as e:
                 logger.warning(f"[Network] Unexpected error for {url}: {e}")
@@ -114,6 +141,9 @@ async def post_url(self, url: str, data: dict, retry=3):
                 try_time += 1
                 if try_time >= retry:
                     break
+                # 连接错误时重建客户端，清除过期连接
+                await reset_shared_client()
+                self._client = await get_shared_client()
                 await asyncio.sleep(5)
             except Exception as e:
                 logger.debug(e)

backend/src/module/rss/engine.py

@@ -149,11 +149,15 @@ async def refresh_rss(self, client: DownloadClient, rss_id: Optional[int] = None
         else:
             rss_item = self.rss.search_id(rss_id)
             rss_items = [rss_item] if rss_item else []
-        # From RSS Items, fetch all torrents concurrently
+        # From RSS Items, fetch all torrents with concurrency limit
         logger.debug("[Engine] Get %s RSS items", len(rss_items))
-        results = await asyncio.gather(
-            *[self._pull_rss_with_status(rss_item) for rss_item in rss_items]
-        )
+        semaphore = asyncio.Semaphore(5)
+
+        async def _limited_pull(item):
+            async with semaphore:
+                return await self._pull_rss_with_status(item)
+
+        results = await asyncio.gather(*[_limited_pull(item) for item in rss_items])
         now = datetime.now(timezone.utc).isoformat()
         # Process results sequentially (DB operations)
         for rss_item, (new_torrents, error) in zip(rss_items, results):

backend/src/test/test_request_url.py

@@ -0,0 +1,99 @@
+"""Tests for network request_url: shared client configuration and reset."""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from module.network.request_url import get_shared_client, reset_shared_client
+
+
+@pytest.fixture(autouse=True)
+async def _clean_shared_client():
+    """Ensure shared client is reset after each test."""
+    yield
+    import module.network.request_url as mod
+
+    if mod._shared_client is not None:
+        await mod._shared_client.aclose()
+        mod._shared_client = None
+        mod._shared_client_proxy_key = None
+
+
+class TestSharedClientLimits:
+    async def test_client_has_keepalive_expiry(self):
+        """Shared client should use a finite keepalive_expiry."""
+        client = await get_shared_client()
+        pool = client._transport._pool
+        assert pool._keepalive_expiry is not None
+        assert pool._keepalive_expiry > 0
+
+    async def test_client_has_max_connections(self):
+        """Shared client should have a connection pool limit."""
+        client = await get_shared_client()
+        pool = client._transport._pool
+        assert pool._max_connections is not None
+        assert pool._max_connections > 0
+
+
+class TestResetSharedClient:
+    async def test_reset_closes_existing_client(self):
+        """reset_shared_client should close and clear the shared client."""
+        client = await get_shared_client()
+        assert client is not None
+
+        await reset_shared_client()
+
+        import module.network.request_url as mod
+
+        assert mod._shared_client is None
+        assert mod._shared_client_proxy_key is None
+
+    async def test_reset_idempotent_when_no_client(self):
+        """reset_shared_client should be safe when no client exists."""
+        import module.network.request_url as mod
+
+        mod._shared_client = None
+        mod._shared_client_proxy_key = None
+        await reset_shared_client()
+
+    async def test_new_client_after_reset(self):
+        """After reset, get_shared_client returns a fresh client."""
+        old_client = await get_shared_client()
+        await reset_shared_client()
+        new_client = await get_shared_client()
+        assert new_client is not old_client
+
+
+class TestRetryWithReset:
+    async def test_get_url_resets_on_connect_error(self):
+        """get_url should call reset_shared_client after ConnectTimeout."""
+        import httpx
+        from module.network.request_url import RequestURL
+
+        call_count = 0
+
+        async def mock_get(**kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise httpx.ConnectTimeout("Connection timed out")
+            resp = MagicMock()
+            resp.status_code = 200
+            resp.raise_for_status = MagicMock()
+            return resp
+
+        with (
+            patch("module.network.request_url.get_shared_client") as mock_get_client,
+            patch(
+                "module.network.request_url.reset_shared_client",
+                new_callable=AsyncMock,
+            ) as mock_reset,
+        ):
+            mock_client = AsyncMock()
+            mock_client.get = mock_get
+            mock_get_client.return_value = mock_client
+
+            async with RequestURL() as req:
+                result = await req.get_url("https://example.com/test", retry=2)
+
+        mock_reset.assert_called()
+        assert call_count == 2

backend/src/test/test_rss_engine_new.py

@@ -1,5 +1,7 @@
 """Tests for RSS engine: pull_rss, match_torrent, refresh_rss, add_rss."""
 
+import asyncio
+
 import pytest
 from unittest.mock import AsyncMock, patch
 
@@ -51,7 +53,9 @@ async def test_returns_only_new_torrents(self, rss_engine):
             Torrent(name="new1", url="https://example.com/new1.torrent"),
             Torrent(name="new2", url="https://example.com/new2.torrent"),
         ]
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             mock_get.return_value = all_torrents
             result = await rss_engine.pull_rss(rss_item)
 
@@ -67,7 +71,9 @@ async def test_all_existing_returns_empty(self, rss_engine):
         existing = make_torrent(url="https://example.com/only.torrent", rss_id=1)
         rss_engine.torrent.add(existing)
 
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             mock_get.return_value = [
                 Torrent(name="only", url="https://example.com/only.torrent")
             ]
@@ -81,7 +87,9 @@ async def test_empty_feed_returns_empty(self, rss_engine):
         rss_engine.rss.add(rss_item)
         rss_item = rss_engine.rss.search_id(1)
 
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             mock_get.return_value = []
             result = await rss_engine.pull_rss(rss_item)
 
@@ -99,9 +107,7 @@ def test_matches_by_title_raw_substring(self, rss_engine):
         bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="")
         rss_engine.bangumi.add(bangumi)
 
-        torrent = make_torrent(
-            name="[Lilith-Raws] Mushoku Tensei - 11 [1080p].mkv"
-        )
+        torrent = make_torrent(name="[Lilith-Raws] Mushoku Tensei - 11 [1080p].mkv")
         result = rss_engine.match_torrent(torrent)
 
         assert result is not None
@@ -122,9 +128,7 @@ def test_filter_excludes_matching_torrent(self, rss_engine):
         bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="720")
         rss_engine.bangumi.add(bangumi)
 
-        torrent = make_torrent(
-            name="[Sub] Mushoku Tensei - 01 [720p].mkv"
-        )
+        torrent = make_torrent(name="[Sub] Mushoku Tensei - 01 [720p].mkv")
         result = rss_engine.match_torrent(torrent)
 
         assert result is None
@@ -134,9 +138,7 @@ def test_empty_filter_allows_match(self, rss_engine):
         bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="")
         rss_engine.bangumi.add(bangumi)
 
-        torrent = make_torrent(
-            name="[Sub] Mushoku Tensei - 01 [720p].mkv"
-        )
+        torrent = make_torrent(name="[Sub] Mushoku Tensei - 01 [720p].mkv")
         result = rss_engine.match_torrent(torrent)
 
         assert result is not None
@@ -147,9 +149,7 @@ def test_filter_case_insensitive(self, rss_engine):
         rss_engine.bangumi.add(bangumi)
 
         # Torrent has "hevc" in lowercase - should still be filtered
-        torrent = make_torrent(
-            name="[Sub] Mushoku Tensei - 01 [1080p][hevc].mkv"
-        )
+        torrent = make_torrent(name="[Sub] Mushoku Tensei - 01 [1080p][hevc].mkv")
         result = rss_engine.match_torrent(torrent)
 
         assert result is None
@@ -201,7 +201,9 @@ async def test_downloads_matched_torrents(self, rss_engine, mock_qb_client):
             name="[Sub] Mushoku Tensei - 12 [1080p].mkv",
             url="https://example.com/ep12.torrent",
         )
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             mock_get.return_value = [new_torrent]
 
             # Create a mock client
@@ -227,7 +229,9 @@ async def test_unmatched_torrents_stored_not_downloaded(self, rss_engine):
             name="[Sub] Unknown Anime - 01 [1080p].mkv",
             url="https://example.com/unknown.torrent",
         )
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             mock_get.return_value = [unmatched]
             client = AsyncMock()
             await rss_engine.refresh_rss(client)
@@ -244,7 +248,9 @@ async def test_refresh_specific_rss_id(self, rss_engine):
         rss_engine.rss.add(rss1)
         rss_engine.rss.add(rss2)
 
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             mock_get.return_value = []
             client = AsyncMock()
             await rss_engine.refresh_rss(client, rss_id=2)
@@ -254,7 +260,9 @@ async def test_refresh_specific_rss_id(self, rss_engine):
 
     async def test_refresh_nonexistent_rss_id(self, rss_engine):
         """refresh_rss with non-existent rss_id does nothing."""
-        with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as mock_get:
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
             client = AsyncMock()
             await rss_engine.refresh_rss(client, rss_id=999)
 
@@ -284,9 +292,7 @@ async def test_add_with_name(self, rss_engine):
 
     async def test_add_without_name_fetches_title(self, rss_engine):
         """add_rss without name calls get_rss_title to auto-discover title."""
-        with patch(
-            "module.rss.engine.RequestContent"
-        ) as MockReq:
+        with patch("module.rss.engine.RequestContent") as MockReq:
             mock_instance = AsyncMock()
             mock_instance.get_rss_title = AsyncMock(return_value="Fetched Title")
             MockReq.return_value.__aenter__ = AsyncMock(return_value=mock_instance)
@@ -303,9 +309,7 @@ async def test_add_without_name_fetches_title(self, rss_engine):
 
     async def test_add_without_name_fetch_fails(self, rss_engine):
         """add_rss returns error when title fetch fails."""
-        with patch(
-            "module.rss.engine.RequestContent"
-        ) as MockReq:
+        with patch("module.rss.engine.RequestContent") as MockReq:
             mock_instance = AsyncMock()
             mock_instance.get_rss_title = AsyncMock(return_value=None)
             MockReq.return_value.__aenter__ = AsyncMock(return_value=mock_instance)
@@ -332,3 +336,36 @@ async def test_add_duplicate_url_fails(self, rss_engine):
 
         assert result.status is False
         assert result.status_code == 406
+
+
+class TestRefreshRssConcurrency:
+    async def test_concurrent_requests_limited(self, rss_engine):
+        """refresh_rss should limit concurrent requests via semaphore."""
+        rss_items = [
+            make_rss_item(name=f"Feed {i}", url=f"https://feed{i}.com/rss")
+            for i in range(10)
+        ]
+        for item in rss_items:
+            rss_engine.rss.add(item)
+
+        active_count = 0
+        max_active = 0
+        lock = asyncio.Lock()
+
+        async def track_concurrency(rss_item):
+            nonlocal active_count, max_active
+            async with lock:
+                active_count += 1
+                max_active = max(max_active, active_count)
+            await asyncio.sleep(0.01)
+            async with lock:
+                active_count -= 1
+            return [], None
+
+        with patch.object(
+            rss_engine, "_pull_rss_with_status", side_effect=track_concurrency
+        ):
+            client = AsyncMock()
+            await rss_engine.refresh_rss(client)
+
+        assert max_active <= 5