From bd1699115ab3767d327c7ff84c5cbbb1832e5c0a Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Tue, 14 Jul 2026 15:21:05 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L --- CHANGELOG.md | 1 + src/clients/content_extractor.py | 55 +++++++++++++++------ tests/test_content_extractor.py | 84 ++++++++++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c7b25a2..13c9c61 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 ` (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) diff --git a/src/clients/content_extractor.py b/src/clients/content_extractor.py index 02bde89..73d6336 100644 --- a/src/clients/content_extractor.py +++ b/src/clients/content_extractor.py @@ -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.""" diff --git a/tests/test_content_extractor.py b/tests/test_content_extractor.py index bdb713a..f300b60 100644 --- a/tests/test_content_extractor.py +++ b/tests/test_content_extractor.py @@ -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."""