From a5e71751d7a9f8f4a71ca458da3984fc1284d098 Mon Sep 17 00:00:00 2001 From: changjing9527 <2169325545@qq.com> Date: Tue, 13 Oct 2026 22:26:28 +0800 Subject: [PATCH] fix(jina): classify retry budget expiry reliably on Windows (#6789) --- .../deerflow/community/jina_ai/jina_client.py | 32 ++++++++++++---- backend/tests/test_jina_response_limit.py | 6 ++- backend/tests/test_jina_retries.py | 37 +++++++++++++++++++ 3 files changed, 66 insertions(+), 9 deletions(-) diff --git a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py b/backend/packages/harness/deerflow/community/jina_ai/jina_client.py index 57951823d0..d402ae6d74 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py +++ b/backend/packages/harness/deerflow/community/jina_ai/jina_client.py @@ -63,6 +63,24 @@ def _retry_after(value: str | None) -> float | None: return None +class _RetryBudgetExhausted(TimeoutError): + """Distinguish the shared retry deadline from transport timeouts.""" + + +@asynccontextmanager +async def _retry_budget_timeout_at(deadline: float | None): + timeout = asyncio.timeout_at(deadline) + try: + async with timeout: + yield + except TimeoutError as exc: + # asyncio's timeout object records its own expiry. Rechecking loop.time() + # here is unreliable on hosts whose monotonic clock has coarse resolution. + if timeout.expired(): + raise _RetryBudgetExhausted from exc + raise + + class _CleanupStream(httpx.AsyncByteStream): """Keep the lease until HTTPX's underlying close completes, even on cancel. @@ -158,7 +176,7 @@ class JinaClient: # deadline covers requests, waits, and cleanup. With admission, # cleanup may outlast the retry budget; cancellation propagates after close. deadline = asyncio.get_running_loop().time() + retry_budget_seconds if max_retries else None - async with asyncio.timeout_at(None if admission else deadline): + async with _retry_budget_timeout_at(None if admission else deadline): client_kwargs: dict[str, object] = {"trust_env": trust_env, "follow_redirects": True} if proxy: client_kwargs["proxy"] = proxy @@ -167,18 +185,18 @@ class JinaClient: for attempt in range(max_retries + 1): remaining = deadline - asyncio.get_running_loop().time() if deadline is not None else None if remaining is not None and remaining <= 0: - raise TimeoutError + raise _RetryBudgetExhausted server_floor = None try: async with admission.attempt(remaining) if admission else nullcontext() as lease: # Admission consumes the original logical budget. Start the # network timer only after acquisition so queue expiry stays local. - async with asyncio.timeout_at(deadline if admission else None): + async with _retry_budget_timeout_at(deadline if admission else None): if lease is not None: lease.check_deadline() remaining = deadline - asyncio.get_running_loop().time() if deadline is not None else None if remaining is not None and remaining <= 0: - raise TimeoutError + raise _RetryBudgetExhausted request_timeout = min(timeout, remaining) if remaining is not None else timeout if max_response_bytes is None and admission is None: response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout) @@ -216,13 +234,13 @@ class JinaClient: if server_floor is not None and server_floor >= remaining: return f"Error: Jina API returned status {response.status_code}: {response_text}" if remaining <= 0: - raise TimeoutError + raise _RetryBudgetExhausted wait = min(delay, remaining) * random.uniform(0.5, 1.0) if server_floor is not None: wait = max(wait, server_floor) if wait >= remaining: return f"Error: Jina API returned status {response.status_code}: {response_text}" - async with asyncio.timeout_at(deadline if admission else None): + async with _retry_budget_timeout_at(deadline if admission else None): await asyncio.sleep(wait) delay = min(delay * 2, 4.0) except JinaAdmissionError as e: @@ -230,7 +248,7 @@ class JinaClient: logger.warning(error_message) return f"Error: {error_message}" except Exception as e: - if isinstance(e, TimeoutError) and max_retries and deadline is not None and asyncio.get_running_loop().time() >= deadline: + if isinstance(e, _RetryBudgetExhausted): error_message = "Request to Jina API failed: retry time budget exhausted" else: error_message = f"Request to Jina API failed: {type(e).__name__}: {e}" diff --git a/backend/tests/test_jina_response_limit.py b/backend/tests/test_jina_response_limit.py index 8f208fe2bb..ff1ba707bd 100644 --- a/backend/tests/test_jina_response_limit.py +++ b/backend/tests/test_jina_response_limit.py @@ -150,7 +150,7 @@ async def test_retry_resets_counter(transport, monkeypatch): assert len(transport[1]) == 2 -@pytest.mark.parametrize("failure", [httpx.ReadError("offline"), httpx.ReadTimeout("offline")]) +@pytest.mark.parametrize("failure", [httpx.ReadError("offline"), httpx.ReadTimeout("offline"), TimeoutError("transport timeout")]) async def test_read_failure_closes_without_retry(transport, failure): stream = respond(transport, [b"a"], failure=failure) result = await JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=1) @@ -180,7 +180,9 @@ async def test_shared_deadline_covers_retry_stream(transport, monkeypatch): assert "retry time budget exhausted" in result assert first.closed and second.closed and second.entered.is_set() timeouts = [request.extensions["timeout"]["read"] for request in transport[1]] - assert len(timeouts) == 2 and 0 < timeouts[1] < timeouts[0] <= 0.05 + # Windows' monotonic clock can return the same sample for both attempts. + # Equality still proves the second request did not receive a fresh budget. + assert len(timeouts) == 2 and 0 < timeouts[1] <= timeouts[0] <= 0.05 * (1 + 2**-24) @pytest.mark.parametrize("limit", [4, "4"]) diff --git a/backend/tests/test_jina_retries.py b/backend/tests/test_jina_retries.py index 9fcdeb18de..0b43344a1e 100644 --- a/backend/tests/test_jina_retries.py +++ b/backend/tests/test_jina_retries.py @@ -8,6 +8,7 @@ from unittest.mock import AsyncMock, Mock import httpx import pytest +from deerflow.community.jina_ai import jina_client from deerflow.community.jina_ai.jina_client import JinaClient pytestmark = pytest.mark.anyio @@ -128,6 +129,42 @@ async def test_shared_deadline(requests, monkeypatch, during_backoff): assert requests.await_count == 1 +async def test_shared_deadline_classifies_expiry_without_clock_recheck(requests, monkeypatch): + class _FrozenClockLoop: + def __init__(self, loop, frozen_at): + self._loop = loop + self._frozen_at = frozen_at + + def time(self): + return self._frozen_at + + def __getattr__(self, name): + return getattr(self._loop, name) + + class _AsyncioWithFrozenLoop: + def __init__(self, module, loop): + self._module = module + self._loop = loop + + def get_running_loop(self): + return self._loop + + def __getattr__(self, name): + return getattr(self._module, name) + + real_loop = asyncio.get_running_loop() + frozen_loop = _FrozenClockLoop(real_loop, real_loop.time()) + monkeypatch.setattr(jina_client, "asyncio", _AsyncioWithFrozenLoop(asyncio, frozen_loop)) + + async def post(*args, **kwargs): + await asyncio.Event().wait() + + requests.side_effect = post + result = await asyncio.wait_for(JinaClient().crawl("https://example.com", max_retries=1, retry_budget_seconds=0.02), timeout=1) + + assert result == "Error: Request to Jina API failed: retry time budget exhausted" + + async def test_budget_is_shared_across_attempts(requests, monkeypatch): timeouts = [] real_sleep = asyncio.sleep