diff --git a/services/data/src/inalpha_data/connectors/cn_market.py b/services/data/src/inalpha_data/connectors/cn_market.py index 0545f256..590da073 100644 --- a/services/data/src/inalpha_data/connectors/cn_market.py +++ b/services/data/src/inalpha_data/connectors/cn_market.py @@ -61,6 +61,10 @@ ) +_TRANSIENT_RETRY_DELAY_S = 0.5 +_TRANSIENT_RETRY_TIMEOUT_S = 10.0 + + class CnMarketError(RuntimeError): """市场级数据源失败(网络 / 反爬 / 改版)。API 层转 502,不静默。""" @@ -303,12 +307,47 @@ async def _get_json( wait = self._min_interval - elapsed if wait > 0: await asyncio.sleep(wait + random.uniform(0.1, 0.5)) - try: - resp = await self._client.get(url, params=params, headers=headers) - except httpx.HTTPError as exc: - raise CnMarketError(f"{host_key} request failed: {exc}") from exc - finally: - self._host_last[host_key] = time.monotonic() + # 首轮 10s + 重试完整 timeout,保证整体仍低于 orchestration 30s 调用预算。 + # `_host_last` 记录每次尝试结束;重试循环不会再次走上方限速计算,显式 + # sleep 使用两者较大值,既快速恢复,也不突破源站最小请求间隔。 + resp: httpx.Response | None = None + for attempt in range(2): + try: + resp = await self._client.get( + url, + params=params, + headers=headers, + timeout=( + min(self._timeout, _TRANSIENT_RETRY_TIMEOUT_S) + if attempt == 0 + else self._timeout + ), + ) + break + except httpx.TransportError as exc: + if attempt == 0: + _logger.warning( + "cn_market_transient_retry", + host=host_key, + error_type=type(exc).__name__, + error=str(exc), + ) + await asyncio.sleep( + max(_TRANSIENT_RETRY_DELAY_S, self._min_interval) + ) + continue + raise CnMarketError( + f"{host_key} request failed after retry: " + f"{type(exc).__name__}: {exc}" + ) from exc + except httpx.HTTPError as exc: + raise CnMarketError( + f"{host_key} request failed: {type(exc).__name__}: {exc}" + ) from exc + finally: + self._host_last[host_key] = time.monotonic() + if resp is None: # pragma: no cover - 循环必返回或抛错,供类型收窄 + raise CnMarketError(f"{host_key} request failed without response") if resp.status_code in (403, 429): raise CnMarketError(f"{host_key} rate-limited/blocked: HTTP {resp.status_code}") if resp.status_code >= 400: diff --git a/services/data/tests/test_market.py b/services/data/tests/test_market.py index 8c7571c5..c37f01af 100644 --- a/services/data/tests/test_market.py +++ b/services/data/tests/test_market.py @@ -8,6 +8,7 @@ import asyncio from typing import Any +import httpx import pytest from fastapi.testclient import TestClient @@ -287,6 +288,42 @@ async def test_connector_caches_success() -> None: await conn.close() +async def test_get_json_retries_one_transient_connect_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """TLS 建连瞬时超时应重试一次,避免东财边缘节点抖动直接变成 502。""" + conn = CnMarketConnector() + conn._min_interval = 1.0 + conn._timeout = 15.0 + + class FakeResp: + status_code = 200 + + @staticmethod + def json() -> dict[str, Any]: + return {"data": {"fastNewsList": []}} + + attempts = 0 + + seen_timeouts: list[float] = [] + + async def fake_get(url: str, params=None, headers=None, **kwargs: Any) -> FakeResp: + nonlocal attempts + attempts += 1 + seen_timeouts.append(kwargs["timeout"]) + if attempts == 1: + raise httpx.ConnectTimeout("TLS handshake timed out") + return FakeResp() + + monkeypatch.setattr(conn._client, "get", fake_get) + result = await conn._get_json("h", "https://example.com", params=None, headers=None) + + assert result == {"data": {"fastNewsList": []}} + assert attempts == 2 + assert seen_timeouts == [10.0, 15.0] + await conn.close() + + async def test_get_json_serializes_per_host(monkeypatch: pytest.MonkeyPatch) -> None: """同 host 两次请求间隔 ≥ min_interval(防封铁律)。""" conn = CnMarketConnector() @@ -307,7 +344,7 @@ async def fake_sleep(d: float) -> None: sleeps.append(d) await real_sleep(0) - async def fake_get(url: str, params=None, headers=None) -> FakeResp: + async def fake_get(url: str, params=None, headers=None, **kwargs: Any) -> FakeResp: return FakeResp() monkeypatch.setattr(conn._client, "get", fake_get)