fix(jina): classify retry budget expiry reliably on Windows (#6789)

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}"

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"])

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