Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 45 additions & 6 deletions services/data/src/inalpha_data/connectors/cn_market.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,10 @@
)


_TRANSIENT_RETRY_DELAY_S = 0.5
_TRANSIENT_RETRY_TIMEOUT_S = 10.0


class CnMarketError(RuntimeError):
"""市场级数据源失败(网络 / 反爬 / 改版)。API 层转 502,不静默。"""

Expand Down Expand Up @@ -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:
Expand Down
39 changes: 38 additions & 1 deletion services/data/tests/test_market.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import asyncio
from typing import Any

import httpx
import pytest
from fastapi.testclient import TestClient

Expand Down Expand Up @@ -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()
Expand All @@ -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)
Expand Down
Loading