github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-27 13:40:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(jina): add bounded crawl completion diagnostics (#6504)

Browse Source 
* 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>
This commit is contained in:
creed 2026-10-27 10:10:26 +08:00 committed by GitHub
parent 89acf51caa
commit 3c482e0f58
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 518 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -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).

25
backend/docs/CONFIGURATION.md
View File

@ -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`

6
backend/packages/harness/deerflow/community/jina_ai/AGENTS.md
View File

@ -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`.

255
backend/packages/harness/deerflow/community/jina_ai/jina_client.py
View File

@ -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

2
backend/tests/test_jina_client.py
View File

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

329
backend/tests/test_jina_diagnostics.py Normal file
View File

@ -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