From 3c482e0f58a4a90cd883cebf1cbe145b39a2559d Mon Sep 17 00:00:00 2001 From: creed <115765389+97Three@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:10:26 +0800 Subject: [PATCH] feat(jina): add bounded crawl completion diagnostics (#6504) * feat(jina): add bounded crawl completion diagnostics Signed-off-by: 97three <2212371308@qq.com> * test(jina): cover diagnostics with request admission Signed-off-by: 97three <2212371308@qq.com> --------- Signed-off-by: 97three <2212371308@qq.com> --- README.md | 2 + backend/docs/CONFIGURATION.md | 25 ++ .../deerflow/community/jina_ai/AGENTS.md | 6 + .../deerflow/community/jina_ai/jina_client.py | 255 ++++++++------ backend/tests/test_jina_client.py | 2 +- backend/tests/test_jina_diagnostics.py | 329 ++++++++++++++++++ 6 files changed, 518 insertions(+), 101 deletions(-) create mode 100644 backend/tests/test_jina_diagnostics.py diff --git a/README.md b/README.md index 5efcc25f69..c2ec4dcc99 100644 --- a/README.md +++ b/README.md @@ -213,6 +213,8 @@ fences, including unclosed fences, in complete and streamed replies. Jina fetches support opt-in bounded retries via `max_retries` (default `0`) and `retry_budget_seconds` (default `30`) in the tool configuration. Valid `Retry-After` hints set a minimum wait for HTTP 429/503; 429 without a valid hint stays terminal. Hints that cannot fit the remaining budget stop retries. Local backoff remains randomized. Retries may increase upstream requests and cost; see [Jina fetch retries](backend/docs/CONFIGURATION.md#jina-fetch-retries). + At DEBUG level, Jina emits one bounded crawl completion summary with attempts, elapsed/backoff seconds, outcome and stop reason. See [diagnostics](backend/docs/CONFIGURATION.md#jina-crawl-diagnostics) for configuration and correlation. + Jina also accepts an opt-in `max_response_bytes` tool setting (positive integer; omitted/null disables it). It stops oversized decoded responses before extraction, with no partial success or retry. This leaves the 4096-character output cap unchanged and does not bound HTTPX decompressor allocations or wire bandwidth; see [response budget](backend/docs/CONFIGURATION.md#jina-response-byte-budget). Jina can also bound concurrent HTTP attempts and FIFO waiting with the opt-in `request_admission` tool extra. One immutable budget is shared across clients and agent threads in each process; replicas have independent budgets. See [Jina request admission](backend/docs/CONFIGURATION.md#jina-request-admission). diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 975491abfd..2a4bdd5344 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -1016,6 +1016,31 @@ retry budget when retries are enabled; the earlier of that deadline and only queue wait is bounded by admission, and the existing HTTP timeout semantics still apply. +#### Jina crawl diagnostics + +Set `log_level: debug` in root `config.yaml` and restart the Gateway to enable +`deerflow.community.jina_ai.jina_client` completion summaries. INFO emits no new +summary. Each logical crawl emits one `Jina crawl completed` message containing: + +- `outcome`: `success`, `error`, or `cancelled`. +- `reason`: `success`, `empty_response`, `response_limit`, `invalid_configuration`, + `nonretryable_http`, `attempt_exhaustion`, `retry_after_unfit`, + `budget_exhaustion`, `request_failure`, or `cancellation`. +- `attempts`: calls started at HTTPX's POST/stream API, including failed connections; + zero for rejected settings. This does not prove remote delivery. +- `elapsed_seconds`: monotonic duration through completion/cleanup. +- `backoff_seconds`: measured time awaiting retry sleep, including a partially + cancelled wait, rather than the planned delay. Neither duration estimates + provider latency or cost. + +Fields remain in the message in plain and enhanced JSON logging. Existing +`logging.enhance` trace filtering supplies the inherited request trace ID when +enabled; no new ID is created. Concurrent crawls keep independent counters but +may share a request trace. Summaries contain no URL, credentials, response body, +headers, or exception text. Existing error logs and return values are unchanged. +Emission is best effort if a logging handler fails. These provider-local records +explain internal retry decisions; they do not replace tracing or tool results. + #### Jina response byte budget On the same Jina `web_fetch` tool entry, optionally set `max_response_bytes: 1048576` diff --git a/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md b/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md index 8d73abe8e9..cc38a2785d 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md +++ b/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md @@ -17,3 +17,9 @@ leases cover response stream cleanup, never retry backoff, idle client-pool clos stream-close ordering; re-verify on upgrades. Drain stream/client close on cancellation, with cancellation taking precedence over cleanup errors. Null bypasses only before first enablement; afterward all callers reuse the frozen policy until restart. Config: `backend/docs/CONFIGURATION.md`. Tests: `tests/test_jina_{client,retries,retry_after,response_limit,request_admission}.py`. + +DEBUG completion summaries keep counters local, count HTTPX API attempts, and +measure monotonic elapsed/backoff time (including cancelled waits). Classify +branches, never error text; emit only enums/numbers in the message for existing +formatters/trace filters. Logging failure must preserve results/cancellation. +Offline coverage: `tests/test_jina_diagnostics.py`. 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 d402ae6d74..c4fefc85ab 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py +++ b/backend/packages/harness/deerflow/community/jina_ai/jina_client.py @@ -149,108 +149,163 @@ class JinaClient: request_admission: dict | None = None, ) -> str: """Fetch with optional bounded retries; cancellation always propagates.""" - global _api_key_warned - headers = { - "Content-Type": "application/json", - "X-Return-Format": return_format, - "X-Timeout": str(timeout), - } - if os.getenv("JINA_API_KEY"): - headers["Authorization"] = f"Bearer {os.getenv('JINA_API_KEY')}" - elif not _api_key_warned: - _api_key_warned = True - logger.warning("Jina API key is not set. Provide your own key to access a higher rate limit. See https://jina.ai/reader for more information.") - data = {"url": url} - deadline = None + diagnostic = logger.isEnabledFor(logging.DEBUG) + started = time.monotonic() if diagnostic else 0.0 + attempts = 0 + backoff_seconds = 0.0 + outcome, reason = "error", "request_failure" + pending_return = False try: - if max_response_bytes is not None and (isinstance(max_response_bytes, bool) or not isinstance(max_response_bytes, int) or max_response_bytes <= 0): - raise ValueError("max_response_bytes must be a positive integer or null") - if isinstance(max_retries, bool) or not isinstance(max_retries, int) or max_retries < 0: - raise ValueError("max_retries must be a non-negative integer") - if isinstance(retry_budget_seconds, bool) or not isinstance(retry_budget_seconds, (int, float)) or not math.isfinite(retry_budget_seconds) or retry_budget_seconds <= 0: - raise ValueError("retry_budget_seconds must be a finite positive number") + global _api_key_warned + headers = { + "Content-Type": "application/json", + "X-Return-Format": return_format, + "X-Timeout": str(timeout), + } + if os.getenv("JINA_API_KEY"): + headers["Authorization"] = f"Bearer {os.getenv('JINA_API_KEY')}" + elif not _api_key_warned: + _api_key_warned = True + logger.warning("Jina API key is not set. Provide your own key to access a higher rate limit. See https://jina.ai/reader for more information.") + data = {"url": url} + deadline = None + try: + reason = "invalid_configuration" + if max_response_bytes is not None and (isinstance(max_response_bytes, bool) or not isinstance(max_response_bytes, int) or max_response_bytes <= 0): + raise ValueError("max_response_bytes must be a positive integer or null") + if isinstance(max_retries, bool) or not isinstance(max_retries, int) or max_retries < 0: + raise ValueError("max_retries must be a non-negative integer") + if isinstance(retry_budget_seconds, bool) or not isinstance(retry_budget_seconds, (int, float)) or not math.isfinite(retry_budget_seconds) or retry_budget_seconds <= 0: + raise ValueError("retry_budget_seconds must be a finite positive number") - admission = get_admission(request_admission) + admission = get_admission(request_admission) + reason = "request_failure" - # HTTPX timeouts are per network phase. Without admission, the outer - # 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 _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 - async with _admission_client(client_kwargs) if admission else httpx.AsyncClient(**client_kwargs) as client: - delay = 0.5 - 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 _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 _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 _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) - response_text = response.text - else: - async with client.stream("POST", "https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout) as response: - content = bytearray() - # Count decoded bytes before retaining a chunk; HTTPX - # decompressor allocations remain outside the cap. - async for chunk in response.aiter_bytes(): - if max_response_bytes is not None and len(content) + len(chunk) > max_response_bytes: - return f"Error: Jina API response exceeds max_response_bytes ({max_response_bytes})" - content.extend(chunk) - response_text = content.decode(response.encoding or "utf-8", errors="replace") - except (httpx.ConnectError, httpx.ConnectTimeout): - if attempt == max_retries: - raise - else: - if response.status_code == 200: - if response_text and response_text.strip(): - return response_text - error_message = "Jina API returned empty response" - logger.error(error_message) - return f"Error: {error_message}" - if response.status_code in {429, 503} and attempt < max_retries: - server_floor = _retry_after(response.headers.get("Retry-After")) - retryable = response.status_code in {502, 503, 504} or (response.status_code == 429 and server_floor is not None) - if not retryable or attempt == max_retries: - error_message = f"Jina API returned status {response.status_code}: {response_text}" - logger.error(error_message) - return f"Error: {error_message}" + # HTTPX timeouts are per network phase. Without admission, the outer + # 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 _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 + async with _admission_client(client_kwargs) if admission else httpx.AsyncClient(**client_kwargs) as client: + delay = 0.5 + 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 _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 _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 _RetryBudgetExhausted + request_timeout = min(timeout, remaining) if remaining is not None else timeout + if max_response_bytes is None and admission is None: + attempts += 1 + response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout) + response_text = response.text + else: + attempts += 1 + async with client.stream("POST", "https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout) as response: + content = bytearray() + # Count decoded bytes before retaining a chunk; HTTPX + # decompressor allocations remain outside the cap. + async for chunk in response.aiter_bytes(): + if max_response_bytes is not None and len(content) + len(chunk) > max_response_bytes: + reason = "response_limit" + pending_return = True + return f"Error: Jina API response exceeds max_response_bytes ({max_response_bytes})" + content.extend(chunk) + response_text = content.decode(response.encoding or "utf-8", errors="replace") + except (httpx.ConnectError, httpx.ConnectTimeout): + if attempt == max_retries: + reason = "attempt_exhaustion" + raise + else: + if response.status_code == 200: + if response_text and response_text.strip(): + outcome, reason = "success", "success" + pending_return = True + return response_text + reason = "empty_response" + error_message = "Jina API returned empty response" + logger.error(error_message) + pending_return = True + return f"Error: {error_message}" + if response.status_code in {429, 503} and attempt < max_retries: + server_floor = _retry_after(response.headers.get("Retry-After")) + retryable = response.status_code in {502, 503, 504} or (response.status_code == 429 and server_floor is not None) + if not retryable or attempt == max_retries: + reason = "attempt_exhaustion" if retryable or (response.status_code == 429 and attempt == max_retries and attempt > 0) else "nonretryable_http" + error_message = f"Jina API returned status {response.status_code}: {response_text}" + logger.error(error_message) + pending_return = True + return f"Error: {error_message}" - # Only allowlisted failures reach the non-blocking wait. - remaining = deadline - asyncio.get_running_loop().time() - 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 _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: + # Only allowlisted failures reach the non-blocking wait. + remaining = deadline - asyncio.get_running_loop().time() + if server_floor is not None and server_floor >= remaining: + reason = "retry_after_unfit" + pending_return = True return f"Error: Jina API returned status {response.status_code}: {response_text}" - 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: - error_message = f"Local Jina request admission rejected: {e}" - logger.warning(error_message) - return f"Error: {error_message}" - except Exception as e: - 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}" - logger.warning(error_message) - return f"Error: {error_message}" + if remaining <= 0: + 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: + reason = "retry_after_unfit" + pending_return = True + return f"Error: Jina API returned status {response.status_code}: {response_text}" + async with _retry_budget_timeout_at(deadline if admission else None): + wait_started = time.monotonic() if diagnostic else 0.0 + try: + await asyncio.sleep(wait) + finally: + if diagnostic: + backoff_seconds += time.monotonic() - wait_started + delay = min(delay * 2, 4.0) + except JinaAdmissionError as e: + if pending_return: + outcome, reason = "error", "request_failure" + error_message = f"Local Jina request admission rejected: {e}" + logger.warning(error_message) + return f"Error: {error_message}" + except Exception as e: + # Cleanup can override a pending success or Error return. + if pending_return: + outcome, reason = "error", "request_failure" + if isinstance(e, _RetryBudgetExhausted): + reason = "budget_exhaustion" + error_message = "Request to Jina API failed: retry time budget exhausted" + else: + error_message = f"Request to Jina API failed: {type(e).__name__}: {e}" + logger.warning(error_message) + return f"Error: {error_message}" + except asyncio.CancelledError: + outcome, reason = "cancelled", "cancellation" + raise + finally: + if diagnostic: + # Keep fields in the message: both plain and enhanced JSON + # formatters retain it. Existing filters supply ambient trace context. + try: + logger.debug( + "Jina crawl completed outcome=%s reason=%s attempts=%d elapsed_seconds=%.6f backoff_seconds=%.6f", + outcome, + reason, + attempts, + time.monotonic() - started, + backoff_seconds, + ) + except (Exception, asyncio.CancelledError): + # Only diagnostic emission is best effort; never mask a result + # or the original cancellation with a broken logging handler. + pass diff --git a/backend/tests/test_jina_client.py b/backend/tests/test_jina_client.py index e386ad9e22..61c7cfecb1 100644 --- a/backend/tests/test_jina_client.py +++ b/backend/tests/test_jina_client.py @@ -152,7 +152,7 @@ async def test_crawl_transient_failure_logs_without_traceback(jina_client, monke with caplog.at_level(logging.DEBUG, logger="deerflow.community.jina_ai.jina_client"): result = await jina_client.crawl("https://example.com") - jina_records = [r for r in caplog.records if r.name == "deerflow.community.jina_ai.jina_client"] + jina_records = [r for r in caplog.records if r.name == "deerflow.community.jina_ai.jina_client" and r.levelno >= logging.WARNING] assert len(jina_records) == 1, f"expected exactly one log record, got {len(jina_records)}" record = jina_records[0] assert record.levelno == logging.WARNING, f"expected WARNING, got {record.levelname}" diff --git a/backend/tests/test_jina_diagnostics.py b/backend/tests/test_jina_diagnostics.py new file mode 100644 index 0000000000..1cfa75c00d --- /dev/null +++ b/backend/tests/test_jina_diagnostics.py @@ -0,0 +1,329 @@ +"""Offline completion diagnostics through real HTTPX request paths.""" + +import asyncio +import logging + +import httpx +import pytest + +from deerflow.community.jina_ai import jina_client as jina + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.fixture +def transport(monkeypatch): + monkeypatch.setenv("JINA_API_KEY", "sentinel-key") + original = httpx.AsyncClient + + def install(handler): + monkeypatch.setattr(httpx, "AsyncClient", lambda **kwargs: original(**kwargs, transport=httpx.MockTransport(handler))) + + return install + + +def summaries(caplog): + return [record.getMessage() for record in caplog.records if record.name == jina.__name__ and record.getMessage().startswith("Jina crawl completed ")] + + +async def test_first_attempt_summary(transport, caplog): + transport(lambda request: httpx.Response(200, text="content")) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + assert await jina.JinaClient().crawl("https://example.invalid") == "content" + records = summaries(caplog) + assert len(records) == 1 + assert "outcome=success reason=success attempts=1" in records[0] + assert "backoff_seconds=0.000000" in records[0] + + +@pytest.mark.parametrize("limit", [None, 100]) +@pytest.mark.parametrize( + "responses, reason, attempts", + [ + ([httpx.Response(200, text="ok")], "success", 1), + ([httpx.Response(503), httpx.Response(200, text="ok")], "success", 2), + ([httpx.Response(429, headers={"Retry-After": "0"}), httpx.Response(200, text="ok")], "success", 2), + ([httpx.ConnectError("sentinel-exception"), httpx.Response(200, text="ok")], "success", 2), + ([httpx.Response(503), httpx.Response(503)], "attempt_exhaustion", 2), + ([httpx.ConnectError("sentinel-exception")] * 2, "attempt_exhaustion", 2), + ([httpx.Response(429)], "nonretryable_http", 1), + ([httpx.Response(429, headers={"Retry-After": "sentinel-header"})], "nonretryable_http", 1), + ([httpx.Response(503, headers={"Retry-After": "60"})], "retry_after_unfit", 1), + ([httpx.Response(401, text="sentinel-body")], "nonretryable_http", 1), + ([httpx.ReadError("sentinel-exception")], "request_failure", 1), + ([httpx.Response(200, text=" ")], "empty_response", 1), + ], +) +async def test_branches(transport, caplog, responses, reason, attempts, limit): + pending = iter(responses) + seen = [] + + def handle(request): + seen.append(request) + item = next(pending) + if isinstance(item, Exception): + raise item + return item + + transport(handle) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + result = await jina.JinaClient().crawl("https://sentinel-url.invalid", max_retries=1, retry_budget_seconds=10, max_response_bytes=limit) + assert result == "ok" if reason == "success" else result.startswith("Error:") + records = summaries(caplog) + assert len(records) == 1 + assert f"reason={reason} attempts={attempts}" in records[0] + assert len(seen) == attempts + assert "sentinel" not in records[0] + + +@pytest.mark.parametrize( + "options,reason,attempts", + [({"max_retries": -1}, "invalid_configuration", 0), ({"retry_budget_seconds": 0}, "invalid_configuration", 0), ({"max_response_bytes": False}, "invalid_configuration", 0), ({"max_response_bytes": 1}, "response_limit", 1)], +) +async def test_validation_and_limit(transport, caplog, options, reason, attempts): + transport(lambda request: httpx.Response(200, text="sentinel-body")) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + assert (await jina.JinaClient().crawl("https://example.invalid", **options)).startswith("Error:") + assert len(summaries(caplog)) == 1 + assert f"reason={reason} attempts={attempts}" in summaries(caplog)[0] + + +async def test_info_quiet(transport, caplog): + transport(lambda request: httpx.Response(200, text="ok")) + caplog.set_level(logging.INFO, logger=jina.__name__) + assert await jina.JinaClient().crawl("https://example.invalid") == "ok" + assert summaries(caplog) == [] + + +@pytest.mark.parametrize("backoff", [False, True]) +@pytest.mark.parametrize("budget", [False, True]) +async def test_cancellation_and_budget(transport, caplog, monkeypatch, backoff, budget): + from types import SimpleNamespace + + entered = asyncio.Event() + clock = [10.0] + # Replace only the provider's observation clock, never asyncio's clock. + monkeypatch.setattr(jina, "time", SimpleNamespace(monotonic=lambda: clock[0])) + loop = asyncio.get_running_loop() + + async def block(): + entered.set() + clock[0] = 12.0 + await asyncio.Event().wait() + + async def handle(request): + if not backoff: + await block() + return httpx.Response(503) + + async def sleep(delay): + await block() + + if backoff: + monkeypatch.setattr(asyncio, "sleep", sleep) + transport(handle) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + task = asyncio.create_task(jina.JinaClient().crawl("https://example.invalid", max_retries=1)) + try: + await entered.wait() + if budget: + # Production classifies budget expiry against its original deadline. + # Preserve that test by expiring the loop clock too, after admission. + original_time = loop.time + monkeypatch.setattr(loop, "time", lambda: original_time() + 31) + assert "budget exhausted" in await task + else: + task.cancel("original-cancellation") + with pytest.raises(asyncio.CancelledError, match="original-cancellation"): + await task + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + (record,) = summaries(caplog) + assert f"reason={'budget_exhaustion' if budget else 'cancellation'} attempts=1" in record + assert "elapsed_seconds=2.000000" in record + assert f"backoff_seconds={'2.000000' if backoff else '0.000000'}" in record + + +@pytest.mark.parametrize("mode", ["success", "error", "cancel"]) +@pytest.mark.parametrize("failure", [RuntimeError, asyncio.CancelledError]) +async def test_logging_failure(transport, caplog, mode, failure): + def handle(request): + if mode == "cancel": + raise asyncio.CancelledError("original") + return httpx.Response(200 if mode == "success" else 403, text="body") + + class BrokenHandler(logging.Handler): + def emit(self, record): + if record.levelno == logging.DEBUG: + raise failure("sentinel-handler") + + transport(handle) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + handler = BrokenHandler() + jina.logger.addHandler(handler) + try: + if mode == "cancel": + with pytest.raises(asyncio.CancelledError, match="original"): + await jina.JinaClient().crawl("https://example.invalid") + else: + result = await jina.JinaClient().crawl("https://example.invalid") + assert result == "body" if mode == "success" else result == "Error: Jina API returned status 403: body" + finally: + jina.logger.removeHandler(handler) + + +async def test_concurrent_trace_and_formatters(transport, caplog): + import json + + from deerflow.logging_config import DEFAULT_LOG_FORMAT, TRACE_TEXT_LOG_FORMAT, JsonTraceFormatter, TraceContextFilter, TraceTextFormatter + from deerflow.trace_context import request_trace_context + + entered = asyncio.Event() + release = asyncio.Event() + count = 0 + + async def handle(request): + nonlocal count + count += 1 + if count == 1: + entered.set() + await release.wait() + return httpx.Response(503, headers={"Retry-After": "0"}) + return httpx.Response(200, text="ok") + + transport(handle) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + trace_filter = TraceContextFilter() + caplog.handler.addFilter(trace_filter) + + async def crawl(trace): + with request_trace_context(trace): + return await jina.JinaClient().crawl("https://example.invalid", max_retries=1) + + task = asyncio.create_task(crawl("first")) + try: + await entered.wait() + assert await crawl("second") == "ok" + release.set() + assert await task == "ok" + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + caplog.handler.removeFilter(trace_filter) + records = [r for r in caplog.records if r.getMessage().startswith("Jina crawl completed ")] + assert len(records) == 2 + assert [(r.trace_id, "attempts=2" in r.getMessage()) for r in records] == [("second", False), ("first", True)] + for record in records: + assert record.getMessage() in logging.Formatter(DEFAULT_LOG_FORMAT).format(record) + assert record.getMessage() in TraceTextFormatter(TRACE_TEXT_LOG_FORMAT).format(record) + payload = json.loads(JsonTraceFormatter().format(record)) + assert payload["message"] == record.getMessage() + assert payload["trace_id"] == record.trace_id + + +async def test_measured_recovery_wait(transport, caplog, monkeypatch): + from types import SimpleNamespace + + clock = [10.0] + monkeypatch.setattr(jina, "time", SimpleNamespace(monotonic=lambda: clock[0])) + responses = iter([httpx.Response(503), httpx.Response(200, text="ok")]) + transport(lambda request: next(responses)) + real_sleep = asyncio.sleep + + async def measured_sleep(delay): + await real_sleep(0) # Preserve a cancellation point; no deadline is advanced. + clock[0] += 0.125 + + monkeypatch.setattr(asyncio, "sleep", measured_sleep) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + assert await jina.JinaClient().crawl("https://example.invalid", max_retries=1) == "ok" + (record,) = summaries(caplog) + assert "elapsed_seconds=0.125000 backoff_seconds=0.125000" in record + + +async def test_client_creation_failure_has_zero_attempts(caplog, monkeypatch): + monkeypatch.setenv("JINA_API_KEY", "sentinel-key") + + def fail(**kwargs): + raise ValueError("sentinel-proxy-credentials") + + monkeypatch.setattr(httpx, "AsyncClient", fail) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + assert (await jina.JinaClient().crawl("https://sentinel-url.invalid")).startswith("Error:") + (record,) = summaries(caplog) + assert "reason=request_failure attempts=0" in record + assert "sentinel" not in record + + +@pytest.mark.parametrize( + ("status", "body", "options"), + [ + (200, "ok", {}), + (200, "", {}), + (200, "large body", {"max_response_bytes": 3}), + (403, "denied", {}), + (503, "unavailable", {}), + (503, "unavailable", {"max_retries": 1, "retry_budget_seconds": 1}), + ], +) +async def test_cleanup_failure_overrides_pending_summary(monkeypatch, caplog, status, body, options): + monkeypatch.setenv("JINA_API_KEY", "sentinel-key") + original = httpx.AsyncClient + + class BrokenClose(httpx.MockTransport): + async def aclose(self): + raise RuntimeError("sentinel-close-failed") + + mock_transport = BrokenClose(lambda request: httpx.Response(status, text=body, headers={"Retry-After": "60"})) + monkeypatch.setattr(httpx, "AsyncClient", lambda **kwargs: original(**kwargs, transport=mock_transport)) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + result = await jina.JinaClient().crawl("https://example.invalid", **options) + assert result == "Error: Request to Jina API failed: RuntimeError: sentinel-close-failed" + (record,) = summaries(caplog) + assert "outcome=error reason=request_failure attempts=1" in record + assert "sentinel" not in record + + +async def test_admission_rejection_has_zero_http_attempts(transport, caplog, monkeypatch): + from deerflow.community.jina_ai import request_admission + + monkeypatch.setattr(request_admission, "_admission", None) + entered, release = asyncio.Event(), asyncio.Event() + seen = [] + + class HeldResponse(httpx.AsyncByteStream): + async def __aiter__(self): + entered.set() + await release.wait() + yield b"ok" + + async def aclose(self): + pass + + def handle(request): + seen.append(request) + return httpx.Response(200, stream=HeldResponse()) + + transport(handle) + caplog.set_level(logging.DEBUG, logger=jina.__name__) + policy = {"max_concurrent_requests": 1, "max_queue_size": 0, "max_wait_seconds": 5} + first = asyncio.create_task(jina.JinaClient().crawl("https://example.invalid/first", request_admission=policy)) + try: + await asyncio.wait_for(entered.wait(), 3) + result = await jina.JinaClient().crawl("https://example.invalid/second", request_admission=policy) + assert result == "Error: Local Jina request admission rejected: queue full" + assert len(seen) == 1 + assert len(summaries(caplog)) == 1 + assert "outcome=error reason=request_failure attempts=0" in summaries(caplog)[0] + finally: + release.set() + assert await first == "ok" + assert len(summaries(caplog)) == 2 + assert "outcome=success reason=success attempts=1" in summaries(caplog)[1] + assert request_admission._admission._active == 0