fix(http): AsyncRequestUtils 代理请求按 host 熔断降级 h2 隧道 (#6208)

app/utils/http.py

@@ -3,10 +3,11 @@ import collections
 import re
 import sys
 import threading
+import time
 import weakref
 from contextlib import AsyncExitStack, contextmanager, asynccontextmanager
 from pathlib import Path
-from typing import Any, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Tuple, Union
 
 import chardet
 import httpx
@@ -72,6 +73,53 @@ _DEFAULT_MAX_CONNECTIONS = 40
 _DEFAULT_KEEPALIVE_EXPIRY = 30
 # 同步 requests.Session 复用连接时，遇到对端或代理关闭 keep-alive 后允许重试的方法
 _REQUESTS_RETRY_IDEMPOTENT_METHODS = ("GET", "HEAD", "OPTIONS")
+
+# 代理走 CONNECT 隧道时，httpx 默认开启的 HTTP/2 多路复用会把并发请求叠加到极少数隧道上；
+# 隧道被代理节点切换或空闲回收打断后，复用其上的所有请求会同时失败。按 (proxy, host) 熔断：
+# 命中一次连接层失败就记录下次允许再尝试 h2 的时间戳（time.monotonic 基准），冷却期内该
+# (proxy, host) 的请求直接退化为 http1.1；冷却期结束后自动恢复尝试 h2。
+_H2_PROXY_BREAKER_COOLDOWN = 1800  # 30 分钟
+_h2_proxy_breaker_lock = threading.Lock()
+_h2_proxy_retry_at: Dict[Tuple[str, str], float] = {}
+# 只有这些错误是"h2 隧道被打断"的特征（对应实测日志里的 SEND_HEADERS in CLOSED、
+# EndOfStream 等）；超时、连接失败、代理不可达等错误换 h1 一样会发生，
+# 不应触发熔断，也不值得付出一次注定同样失败的 h1 重试
+_H2_TUNNEL_BREAK_ERRORS = (
+    httpx.RemoteProtocolError,
+    httpx.LocalProtocolError,
+    httpx.ReadError,
+    httpx.WriteError,
+    httpx.CloseError,
+)
+
+
+def _h2_proxy_breaker_key(proxy: str, url: str) -> Tuple[str, str]:
+    try:
+        host = httpx.URL(url).host or ""
+    except Exception:
+        host = url
+    return proxy, host
+
+
+def _h2_proxy_allowed(proxy: Optional[str], url: str) -> bool:
+    """判断给定代理 + 目标 host 当前是否允许尝试 h2（未处于熔断冷却期）"""
+    if not proxy:
+        return True
+    with _h2_proxy_breaker_lock:
+        retry_at = _h2_proxy_retry_at.get(_h2_proxy_breaker_key(proxy, url), 0.0)
+    return time.monotonic() >= retry_at
+
+
+def _trip_h2_proxy_breaker(proxy: str, url: str) -> None:
+    """记录一次 h2 连接层失败，熔断该 (proxy, host) 冷却期内的 h2 尝试"""
+    now = time.monotonic()
+    with _h2_proxy_breaker_lock:
+        # 顺手清掉已过冷却期的条目，防止长期运行下字典无限增长
+        for key in [k for k, retry_at in _h2_proxy_retry_at.items() if now >= retry_at]:
+            del _h2_proxy_retry_at[key]
+        _h2_proxy_retry_at[_h2_proxy_breaker_key(proxy, url)] = now + _H2_PROXY_BREAKER_COOLDOWN
+
+
 # 持有 LRU 淘汰后正在异步关闭的 transport task，避免 fire-and-forget 被 GC 警告
 _pending_eviction_tasks: set[asyncio.Task] = set()
 
@@ -1087,13 +1135,48 @@ class AsyncRequestUtils:
                 self._client, method, url, raise_exception, **kwargs
             )
 
+        # 代理走 CONNECT 隧道时 h2 多路复用容易被隧道打断放大成批量失败（见
+        # _h2_proxy_allowed 处注释）；仅对幂等方法做"h2 失败就地降级 h1 重试"，
+        # 避免非幂等请求在服务端可能已收到数据的情况下重复产生副作用
+        http2 = self._http2 and _h2_proxy_allowed(self._proxies, url)
+        if not (http2 and self._proxies and method.upper() in _REQUESTS_RETRY_IDEMPOTENT_METHODS):
+            return await self._dispatch_request(
+                http2, cookies_dict, method, url, raise_exception, **kwargs
+            )
+
+        try:
+            return await self._dispatch_request(
+                True, cookies_dict, method, url, True, **kwargs
+            )
+        except _H2_TUNNEL_BREAK_ERRORS as e:
+            logger.debug(f"h2 代理连接层失败，熔断 {url} 所在 host 并降级 h1 重试: {e!r}")
+            _trip_h2_proxy_breaker(self._proxies, url)
+            return await self._dispatch_request(
+                False, cookies_dict, method, url, raise_exception, **kwargs
+            )
+        except httpx.RequestError as e:
+            # 与 h2 隧道无关的失败（超时、连接失败等）：不熔断也不重试，
+            # 恢复调用方原本的 raise_exception 语义
+            if raise_exception:
+                raise
+            error_msg = str(e) or f"未知网络错误 (URL: {url}, Method: {method.upper()})"
+            logger.debug(f"异步请求失败: {error_msg}")
+            return None
+
+    async def _dispatch_request(
+        self, http2: bool, cookies_dict: Optional[dict], method: str, url: str,
+        raise_exception: bool, **kwargs
+    ) -> Optional[httpx.Response]:
+        """
+        按给定 http2 开关构建/复用底层连接并发起请求，供 request() 的 h2/h1 熔断切换复用
+        """
         # 共享底层 transport（连接池+TLS 复用），每次请求创建轻量 AsyncClient。
         # AsyncClient 持有的 cookie jar 仅存活于本次请求 lifecycle，
         # 既复用握手又彻底避免 jar 跨调用累积。
         transport = _get_shared_async_transport(
             proxy=self._proxies,
             verify=self._verify,
-            http2=self._http2,
+            http2=http2,
             max_keepalive_connections=self._max_keepalive_connections,
             max_connections=self._max_connections,
             keepalive_expiry=self._keepalive_expiry,
@@ -1113,7 +1196,7 @@ class AsyncRequestUtils:
 
         # 兜底：没有运行中的事件循环时，临时客户端走完即关
         async with httpx.AsyncClient(
-            http2=self._http2,
+            http2=http2,
             proxy=self._proxies,
             timeout=self._timeout,
             verify=self._verify,

tests/test_async_request_utils.py

@@ -0,0 +1,156 @@
+import asyncio
+import time
+
+import httpx
+import pytest
+
+from app.utils import http as http_module
+from app.utils.http import AsyncRequestUtils
+
+PROXY = "http://proxy.example:7890"
+URL = "https://raw.githubusercontent.com/demo/repo/main/package.json"
+
+
+@pytest.fixture(autouse=True)
+def _reset_h2_proxy_breaker():
+    """每个用例前后清空熔断状态，避免跨用例污染进程级熔断字典。"""
+    http_module._h2_proxy_retry_at.clear()
+    yield
+    http_module._h2_proxy_retry_at.clear()
+
+
+def _fake_dispatch(calls, fail_when):
+    async def fake(_self, http2, _cookies_dict, _method, _url, raise_exception, **_kwargs):
+        calls.append(http2)
+        if fail_when(http2):
+            if raise_exception:
+                raise httpx.RemoteProtocolError("tunnel closed")
+            return None
+        return "ok"
+
+    return fake
+
+
+def test_get_downgrades_to_h1_and_trips_breaker_on_h2_failure(monkeypatch):
+    """
+    走代理的幂等请求 h2 遇到连接层失败时，就地降级 h1 重试一次并触发该 (proxy, host) 的熔断。
+    """
+    calls = []
+    monkeypatch.setattr(
+        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: http2)
+    )
+
+    utils = AsyncRequestUtils(proxies={"https": PROXY})
+    result = asyncio.run(utils.request("get", URL))
+
+    assert result == "ok"
+    assert calls == [True, False]
+    assert http_module._h2_proxy_allowed(PROXY, URL) is False
+
+
+def test_get_skips_h2_while_breaker_is_tripped(monkeypatch):
+    """
+    熔断冷却期内，走代理的请求直接使用 h1，不再尝试 h2。
+    """
+    http_module._trip_h2_proxy_breaker(PROXY, URL)
+    calls = []
+    monkeypatch.setattr(
+        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: False)
+    )
+
+    utils = AsyncRequestUtils(proxies={"https": PROXY})
+    result = asyncio.run(utils.request("get", URL))
+
+    assert result == "ok"
+    assert calls == [False]
+
+
+def test_get_retries_h2_after_cooldown_expires(monkeypatch):
+    """
+    熔断冷却期结束后，代理请求恢复尝试 h2。
+    """
+    http_module._h2_proxy_retry_at[
+        http_module._h2_proxy_breaker_key(PROXY, URL)
+    ] = time.monotonic() - 1
+    calls = []
+    monkeypatch.setattr(
+        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: False)
+    )
+
+    utils = AsyncRequestUtils(proxies={"https": PROXY})
+    result = asyncio.run(utils.request("get", URL))
+
+    assert result == "ok"
+    assert calls == [True]
+
+
+def test_timeout_does_not_trip_breaker_or_retry(monkeypatch):
+    """
+    超时等与 h2 隧道无关的错误不触发熔断、不做 h1 重试，且保持 raise_exception=False 返回 None 的语义。
+    """
+    calls = []
+
+    async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs):
+        calls.append(http2)
+        raise httpx.ConnectTimeout("proxy slow")
+
+    monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake)
+
+    utils = AsyncRequestUtils(proxies={"https": PROXY})
+    result = asyncio.run(utils.request("get", URL))
+
+    assert result is None
+    assert calls == [True]
+    assert http_module._h2_proxy_allowed(PROXY, URL) is True
+
+
+def test_timeout_still_raises_when_raise_exception_enabled(monkeypatch):
+    """
+    与 h2 隧道无关的错误在 raise_exception=True 时按原语义抛出。
+    """
+    calls = []
+
+    async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs):
+        calls.append(http2)
+        raise httpx.ConnectTimeout("proxy slow")
+
+    monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake)
+
+    utils = AsyncRequestUtils(proxies={"https": PROXY})
+    with pytest.raises(httpx.ConnectTimeout):
+        asyncio.run(utils.request("get", URL, raise_exception=True))
+
+    assert calls == [True]
+    assert http_module._h2_proxy_allowed(PROXY, URL) is True
+
+
+def test_post_does_not_downgrade_on_h2_failure(monkeypatch):
+    """
+    非幂等方法 h2 失败时不做 h1 降级重试，避免服务端可能已收到数据时重复产生副作用。
+    """
+    calls = []
+    monkeypatch.setattr(
+        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: True)
+    )
+
+    utils = AsyncRequestUtils(proxies={"https": PROXY})
+    result = asyncio.run(utils.request("post", URL))
+
+    assert result is None
+    assert calls == [True]
+
+
+def test_no_proxy_configured_skips_breaker_logic(monkeypatch):
+    """
+    未配置代理时不触发熔断判断，行为与原实现一致（仅走一次 h2）。
+    """
+    calls = []
+    monkeypatch.setattr(
+        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: False)
+    )
+
+    utils = AsyncRequestUtils()
+    result = asyncio.run(utils.request("get", URL))
+
+    assert result == "ok"
+    assert calls == [True]