fix: stream content fetches so the 5MB cap aborts the download
ContentExtractor._fetch used client.get(), buffering the whole body in memory before the MAX_RESPONSE_BYTES check truncated it - the cap protected Trafilatura but not memory/bandwidth (a multi-hundred-MB URL was still fully downloaded, on up to max_urls_per_batch concurrent fetches, bounded only by the read timeout). Fetches now stream via client.stream + aiter_bytes and close the connection as soon as the cap is reached; charset still comes from the Content-Type header, available before the body is read. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
This commit is contained in:
Claude Fable 5
@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
- **Runtime prefetch registration is executable** - `SchedulerClient.register_volatile_fetch` (used by consolidation's `_register_prefetch`) registered tasks that were dead on arrival three ways: the JSON body sat under the ignored `body` key (`rest_api_executor` only reads `config["payload"]`), there was no `auth` block (the scheduled POST would 401 against library-desk's `verify_api_key`), and `user` was in the body while every `/volatile/fetch` endpoint requires it as a QUERY parameter (would 422). The config now puts `user` in the URL query string (URL-encoded), an empty `payload`, and `auth: {type: bearer, token: "${LIBRARY_API_KEY}"}` (substituted from the Scheduler's environment; never stored raw). `SchedulerClient` itself also sent no Authorization to the Scheduler API, so registration failed silently at consolidation time — it now sends `Authorization: Bearer <SCHEDULER_API_KEY>` (new `scheduler_api_key` setting; a client without a key logs a warning).
|
||||
- **Document sync uses delete-last reindex order** - `DocumentSyncService._index_vectors` deleted the document's existing chunks (`delete_by_filter`) BEFORE embedding, so a failed embedding pass (e.g. Ollama down) left the Paperless document with zero vectors until the next successful sync — the exact hazard already fixed for wiki pages in `VectorService.update_from_page`. Chunk point ids are now deterministic (uuid5 of `document_{id}_chunk_{i}`, replacing random uuid4) so re-upserting overwrites in place; new points are upserted first and stale points (including legacy uuid4 ones) are pruned afterwards, only after a successful upsert.
|
||||
- **5MB response cap now aborts the download** - `ContentExtractor._fetch` buffered the entire response body in memory before truncating to `MAX_RESPONSE_BYTES`, so the cap protected Trafilatura but not memory/bandwidth (a multi-hundred-MB URL was still fully downloaded, on up to `max_urls_per_batch` concurrent fetches). Fetches are now streamed (`client.stream` + `aiter_bytes`) and the connection is closed as soon as the cap is reached.
|
||||
- **Registrar authenticates to the Scheduler API** - `scripts/register_scheduler_tasks.py --execute` sent no `Authorization` header while the Scheduler's task-management endpoints are Bearer-guarded (`verify_api_key`: 401 on missing key), so the existence probes 401'd (misread as "task absent") and every registration failed; only the public `/health` gate passed. `--execute` now requires `SCHEDULER_API_KEY` in the environment (refuses to run without it, key is never stored) and sends `Authorization: Bearer $SCHEDULER_API_KEY` on all of its own HTTP calls. Deploy note updated alongside the existing `LIBRARY_API_KEY` requirement.
|
||||
|
||||
### Fixed (hazards batch)
|
||||
|
||||
@@ -38,8 +38,9 @@ DEFAULT_HEADERS = {
|
||||
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) LibraryDesk-ContentExtractor"
|
||||
}
|
||||
|
||||
# Responses larger than this are truncated before parsing (Trafilatura on a
|
||||
# multi-hundred-MB response would pin a worker thread and exhaust memory).
|
||||
# Downloads are streamed and ABORTED past this many bytes (protects memory
|
||||
# and bandwidth during the fetch, and keeps Trafilatura from pinning a
|
||||
# worker thread on a multi-hundred-MB response).
|
||||
MAX_RESPONSE_BYTES = 5 * 1024 * 1024
|
||||
|
||||
|
||||
@@ -103,25 +104,49 @@ class ContentExtractor:
|
||||
"""
|
||||
Fetch a URL asynchronously under real connect/read timeouts.
|
||||
|
||||
The body is STREAMED and the download is aborted as soon as
|
||||
MAX_RESPONSE_BYTES have been received, so the cap bounds memory and
|
||||
bandwidth during the download itself (the old implementation
|
||||
buffered the entire response before truncating, so a
|
||||
multi-hundred-MB URL was still fully downloaded).
|
||||
|
||||
Returns:
|
||||
Response text (truncated to MAX_RESPONSE_BYTES) or None when the
|
||||
Response text (capped at MAX_RESPONSE_BYTES) or None when the
|
||||
response is empty / not OK.
|
||||
|
||||
Raises:
|
||||
httpx.HTTPError subclasses on timeout/network errors.
|
||||
"""
|
||||
response = await self._http.get(url)
|
||||
if response.status_code != 200 or not response.content:
|
||||
logger.debug(f"Fetch returned status {response.status_code} for {url}")
|
||||
return None
|
||||
if len(response.content) > MAX_RESPONSE_BYTES:
|
||||
logger.warning(
|
||||
f"Response for {url} exceeds {MAX_RESPONSE_BYTES} bytes; truncating"
|
||||
)
|
||||
return response.content[:MAX_RESPONSE_BYTES].decode(
|
||||
response.encoding or "utf-8", errors="replace"
|
||||
)
|
||||
return response.text
|
||||
async with self._http.stream("GET", url) as response:
|
||||
if response.status_code != 200:
|
||||
logger.debug(
|
||||
f"Fetch returned status {response.status_code} for {url}"
|
||||
)
|
||||
return None
|
||||
|
||||
chunks: List[bytes] = []
|
||||
received = 0
|
||||
truncated = False
|
||||
async for chunk in response.aiter_bytes():
|
||||
if received + len(chunk) >= MAX_RESPONSE_BYTES:
|
||||
chunks.append(chunk[: MAX_RESPONSE_BYTES - received])
|
||||
truncated = True
|
||||
break
|
||||
chunks.append(chunk)
|
||||
received += len(chunk)
|
||||
|
||||
body = b"".join(chunks)
|
||||
if not body:
|
||||
return None
|
||||
if truncated:
|
||||
logger.warning(
|
||||
f"Response for {url} exceeds {MAX_RESPONSE_BYTES} bytes; "
|
||||
"download aborted and body truncated"
|
||||
)
|
||||
# charset comes from the Content-Type header, available before
|
||||
# the body is read.
|
||||
encoding = response.charset_encoding or "utf-8"
|
||||
return body.decode(encoding, errors="replace")
|
||||
|
||||
def _log_queue_depth(self, context: str) -> None:
|
||||
"""Log thread-pool queue depth so extraction backpressure is visible."""
|
||||
|
||||
@@ -243,6 +243,90 @@ class TestContentExtractor:
|
||||
assert result.content == "Article content here."
|
||||
|
||||
|
||||
class TestFetchStreamingCap:
|
||||
"""The 5MB cap must abort the DOWNLOAD, not just truncate after it."""
|
||||
|
||||
def _extractor_with_transport(self, handler):
|
||||
extractor = ContentExtractor(timeout=5, max_length=2000)
|
||||
extractor._http = httpx.AsyncClient(
|
||||
transport=httpx.MockTransport(handler)
|
||||
)
|
||||
return extractor
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_aborts_past_cap(self):
|
||||
from src.clients.content_extractor import MAX_RESPONSE_BYTES
|
||||
|
||||
chunk = b"x" * (1024 * 1024) # 1MB per chunk
|
||||
chunks_produced = []
|
||||
|
||||
async def body():
|
||||
for i in range(100): # 100MB on offer
|
||||
chunks_produced.append(i)
|
||||
yield chunk
|
||||
|
||||
def handler(request):
|
||||
return httpx.Response(
|
||||
200,
|
||||
content=body(),
|
||||
headers={"Content-Type": "text/html; charset=utf-8"},
|
||||
)
|
||||
|
||||
extractor = self._extractor_with_transport(handler)
|
||||
try:
|
||||
text = await extractor._fetch("https://example.com/huge")
|
||||
finally:
|
||||
await extractor.close()
|
||||
|
||||
assert text is not None
|
||||
assert len(text.encode()) == MAX_RESPONSE_BYTES
|
||||
# Streaming stopped at the cap instead of consuming all 100 chunks
|
||||
assert len(chunks_produced) <= (MAX_RESPONSE_BYTES // len(chunk)) + 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_small_response_returned_whole(self):
|
||||
def handler(request):
|
||||
return httpx.Response(
|
||||
200,
|
||||
content=TEST_HTML.encode(),
|
||||
headers={"Content-Type": "text/html; charset=utf-8"},
|
||||
)
|
||||
|
||||
extractor = self._extractor_with_transport(handler)
|
||||
try:
|
||||
text = await extractor._fetch("https://example.com/small")
|
||||
finally:
|
||||
await extractor.close()
|
||||
|
||||
assert text == TEST_HTML
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_200_returns_none(self):
|
||||
def handler(request):
|
||||
return httpx.Response(404, content=b"not found")
|
||||
|
||||
extractor = self._extractor_with_transport(handler)
|
||||
try:
|
||||
text = await extractor._fetch("https://example.com/missing")
|
||||
finally:
|
||||
await extractor.close()
|
||||
|
||||
assert text is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_body_returns_none(self):
|
||||
def handler(request):
|
||||
return httpx.Response(200, content=b"")
|
||||
|
||||
extractor = self._extractor_with_transport(handler)
|
||||
try:
|
||||
text = await extractor._fetch("https://example.com/empty")
|
||||
finally:
|
||||
await extractor.close()
|
||||
|
||||
assert text is None
|
||||
|
||||
|
||||
class TestContentExtractionResult:
|
||||
"""Tests for ContentExtractionResult model."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user