diff --git a/app/utils/http.py b/app/utils/http.py index 12662d5e7..5c7e68f1b 100644 --- a/app/utils/http.py +++ b/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, diff --git a/tests/test_async_request_utils.py b/tests/test_async_request_utils.py new file mode 100644 index 000000000..b2b77e243 --- /dev/null +++ b/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]