diff --git a/CHANGELOG.md b/CHANGELOG.md index 7dece556..fa46810d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,8 @@ ### Added -- New `haiku-ingester` service for continuous document ingestion: persistent SQLite job queue, async worker pool with retries and a dead-letter queue, FS/HTTP/S3/WebDAV source adapters with per-source circuit breakers, and a FastAPI control plane (`/health`, `/jobs`, `/sources`, `/dlq`). Configured under `ingester:` in `haiku.rag.yaml`. Shipped behind the `[ingester]` extra, with Logfire spans (`ingester.poller.sweep` → `ingester.job` → `document.{fetch,convert,chunk,embed,store}`) for traceable ingestion. See [docs/ingester.md](docs/ingester.md). +- New `haiku-ingester` service for continuous document ingestion: persistent SQLite job queue, async worker pool with retries and a dead-letter queue, FS/HTTP/S3/WebDAV source adapters with per-source circuit breakers, and a FastAPI control plane (`/health`, `/jobs`, `/sources`, `/dlq`). Configured under `ingester:` in `haiku.rag.yaml`. Shipped behind the `[ingester]` extra, with Logfire spans (`ingester.poller.sweep` → `ingester.job` → `document.{fetch,convert,chunk,embed,store}`, plus `document.convert_slice` per slice when splitting) for traceable ingestion. See [docs/ingester.md](docs/ingester.md). +- `processing.split_pages` (default `0`): split large PDFs into N-page slices, convert each independently through docling-local or docling-serve, and merge with `DoclingDocument.concatenate()`. Bounds peak working set on memory-hungry docs and lets multiple docling-serve replicas parallelize per-document. `0` disables (single-pass conversion). ### Removed diff --git a/examples/docker/docker-compose.yml b/examples/docker/docker-compose.yml index ced8d224..b616ac34 100644 --- a/examples/docker/docker-compose.yml +++ b/examples/docker/docker-compose.yml @@ -72,6 +72,10 @@ services: condition: service_healthy docling-serve-2: condition: service_healthy + # Compose default is 10s — too short for the ingester's own 60s + # shutdown_grace_s. Give it 180s so SIGTERM → drain → exit completes + # before the SIGKILL fallback fires. + stop_grace_period: 180s healthcheck: # No curl in the slim image; use Python's stdlib instead. test: @@ -115,4 +119,5 @@ services: depends_on: haiku-ingester: condition: service_healthy + stop_grace_period: 30s restart: unless-stopped diff --git a/haiku_rag_slim/haiku/rag/converters/pdf_split.py b/haiku_rag_slim/haiku/rag/converters/pdf_split.py index 4c415c0a..ece6d103 100644 --- a/haiku_rag_slim/haiku/rag/converters/pdf_split.py +++ b/haiku_rag_slim/haiku/rag/converters/pdf_split.py @@ -23,12 +23,26 @@ them. import asyncio import io import tempfile +import threading from collections.abc import Iterator from pathlib import Path from typing import TYPE_CHECKING +import logfire import pypdfium2 as pdfium +from haiku.rag.client.exceptions import UnsupportedSourceError + +# pypdfium2 wraps libpdfium, which has global C state and is not thread-safe. +# Multiple workers calling iter_pdf_slices concurrently (via asyncio.to_thread) +# race on that state and corrupt it — first error surfaces as e.g. +# "Failed to import pages", and after that every subsequent PDF load fails +# with "Failed to load document" until the process restarts. Hold this lock +# around every pdfium call so only one worker's slicing runs at a time. The +# actual heavy work (docling-serve convert) happens AFTER iter_pdf_slices +# returns and continues to parallelize across workers. +_PDFIUM_LOCK = threading.Lock() + if TYPE_CHECKING: from docling_core.types.doc.document import DoclingDocument @@ -43,21 +57,31 @@ def iter_pdf_slices(path: Path, slice_size: int) -> Iterator[tuple[int, int, byt """ if slice_size <= 0: raise ValueError(f"slice_size must be >= 1, got {slice_size}") - src = pdfium.PdfDocument(str(path)) - try: - total = len(src) - for start in range(0, total, slice_size): - end = min(start + slice_size, total) - dst = pdfium.PdfDocument.new() - try: - dst.import_pages(src, list(range(start, end))) - buf = io.BytesIO() - dst.save(buf) - yield (start + 1, end, buf.getvalue()) - finally: - dst.close() - finally: - src.close() + with _PDFIUM_LOCK: + try: + src = pdfium.PdfDocument(str(path)) + except pdfium.PdfiumError as exc: + # Malformed / unreadable PDF — retrying won't help. Raise the + # typed error so the ingester pipeline classifies it as + # PermanentError and dead-letters the job instead of cycling it + # through attempts. + raise UnsupportedSourceError( + f"pypdfium2 cannot open PDF {path}: {exc}" + ) from exc + try: + total = len(src) + for start in range(0, total, slice_size): + end = min(start + slice_size, total) + dst = pdfium.PdfDocument.new() + try: + dst.import_pages(src, list(range(start, end))) + buf = io.BytesIO() + dst.save(buf) + yield (start + 1, end, buf.getvalue()) + finally: + dst.close() + finally: + src.close() async def convert_pdf_with_splitting( @@ -86,14 +110,23 @@ async def convert_pdf_with_splitting( tmp.flush() tmp_path = Path(tmp.name) try: - try: - slice_doc = await converter.convert_file( - tmp_path, source_uri=source_uri - ) - except Exception as exc: - raise ValueError( - f"Failed to convert slice pages {start}-{end} of {path}: {exc}" - ) from exc + # One span per slice so each docling-serve round-trip (or local + # docling invocation) is visible in Logfire; the outer + # `document.convert` span aggregates the lot. + with logfire.span( + "document.convert_slice", + uri=source_uri, + start_page=start, + end_page=end, + ): + try: + slice_doc = await converter.convert_file( + tmp_path, source_uri=source_uri + ) + except Exception as exc: + raise ValueError( + f"Failed to convert slice pages {start}-{end} of {path}: {exc}" + ) from exc converted.append(slice_doc) finally: tmp_path.unlink(missing_ok=True) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/server.py b/haiku_rag_slim/haiku/rag/ingester/api/server.py index c956eabe..6a04b34f 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/server.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/server.py @@ -52,5 +52,7 @@ def build_app( app.include_router(dlq.router, dependencies=auth_dep) # Every request becomes a span when logfire is configured; no-op otherwise. - logfire.instrument_fastapi(app) + # /health is the docker healthcheck endpoint — polled every few seconds + # by Compose, would otherwise drown the trace stream in idle GETs. + logfire.instrument_fastapi(app, excluded_urls=r"^.*/health$") return app