From b9881a1bc5e01551f861a7aa505f4b291a35eeaa Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 26 Jun 2026 13:33:18 +0200 Subject: [PATCH 1/6] fix(indexer): honor preset parsing_strategy, pymupdf markdown, parse timeout - #569 honor a preset's parsing_strategy: ParserDispatcher.for_pdf_strategy() + a parser_factory wired into the pipeline, so pymupdf/docling are reachable per preset instead of always using the global default backend. - #570 build pymupdf in markdown mode with embed_images=False: structured text for the chunker, no base64 bloat / Milvus gRPC overflow. - #571 bound the parse stage with loader.parse_timeout (PARSE_TIMEOUT, default 3600s) so a wedged parse fails that file instead of hanging indexing. Tests cover strategy dispatch, pymupdf markdown mode and the parse-timeout error. --- conf/config.yaml | 7 ++++ openrag/core/config/indexation.py | 6 +++ openrag/core/config/loader.py | 1 + openrag/core/indexing/parsers/pdf/pymupdf.py | 29 ++++++------- openrag/services/workers/indexer_pool.py | 36 ++++++++++++++++ .../workers/parsers/parser_dispatcher.py | 41 +++++++++++++++++++ openrag/services/workers/stages/parse.py | 8 +++- .../workers/parsers/test_parser_dispatcher.py | 35 ++++++++++++++++ .../services/workers/stages/test_parse.py | 25 +++++++++++ .../services/workers/test_indexer_pool.py | 36 +++++++++++++++- 10 files changed, 205 insertions(+), 19 deletions(-) diff --git a/conf/config.yaml b/conf/config.yaml index 5567f4e27..040592eee 100644 --- a/conf/config.yaml +++ b/conf/config.yaml @@ -228,6 +228,13 @@ loader: docling_max_task_retry: 3 docling_retry_base_delay: 2.0 + # Env: PARSE_TIMEOUT + # Outer wall-clock bound (seconds) for a single file's parse stage. marker and + # docling self-limit at marker_timeout/docling_timeout; pymupdf has no internal + # timeout, so this prevents a wedged pymupdf parse from stalling indexing — the + # bad file fails and is reported instead. + parse_timeout: 3600 + # Env: TRANSCRIBER_BASE_URL, TRANSCRIBER_API_KEY, TRANSCRIBER_MODEL, # TRANSCRIBER_TIMEOUT, TRANSCRIBER_MAX_CONCURRENT_CHUNKS, USE_WHISPER_LANG_DETECTOR, # TRANSCRIBER_DIRECT_UPLOAD_SUFFIXES (pipe-separated, e.g. ".mp3|.mp4|.wav") diff --git a/openrag/core/config/indexation.py b/openrag/core/config/indexation.py index 0469ca5ea..e280b6e63 100644 --- a/openrag/core/config/indexation.py +++ b/openrag/core/config/indexation.py @@ -178,6 +178,12 @@ class LoaderConfig(ConfigMixin): docling_timeout: int = 3600 docling_max_task_retry: int = 3 docling_retry_base_delay: float = 2.0 + # Outer wall-clock bound (seconds) for a single file's parse stage. Unlike + # marker/docling — which self-limit at marker_timeout/docling_timeout — the + # pymupdf backend has no internal timeout, so without this a wedged pymupdf + # parse would stall indexing indefinitely. Bounds any backend (and any future + # unbounded one); a slow/wedged parse fails *that* file instead of hanging. + parse_timeout: int = 3600 transcriber: TranscriberConfig = Field(default_factory=TranscriberConfig) openai: OpenAILoaderConfig = Field(default_factory=OpenAILoaderConfig) # Max depth of nested .eml-in-.eml attachments the EmlLoader will descend diff --git a/openrag/core/config/loader.py b/openrag/core/config/loader.py index c5d8beb61..a800afae2 100644 --- a/openrag/core/config/loader.py +++ b/openrag/core/config/loader.py @@ -106,6 +106,7 @@ ("DOCLING_NUM_GPUS", "loader.docling_num_gpus", float), ("DOCLING_POOL_SIZE", "loader.docling_pool_size", int), ("DOCLING_MAX_TASKS_PER_WORKER", "loader.docling_max_tasks_per_worker", int), + ("PARSE_TIMEOUT", "loader.parse_timeout", int), ("WHISPER_MODEL", "loader.local_whisper.model", str), ("WHISPER_N_WORKERS", "loader.local_whisper.whisper_n_workers", int), ("WHISPER_NUM_GPUS", "loader.local_whisper.whisper_num_gpus", float), diff --git a/openrag/core/indexing/parsers/pdf/pymupdf.py b/openrag/core/indexing/parsers/pdf/pymupdf.py index f42b33755..503518aba 100644 --- a/openrag/core/indexing/parsers/pdf/pymupdf.py +++ b/openrag/core/indexing/parsers/pdf/pymupdf.py @@ -31,7 +31,6 @@ import pymupdf4llm from ....models.document import Document, DocumentType, ImageBlock, ProcessedDocument, TextBlock -from ...image_preprocessor import extract_data_uri_image_blocks from ..document_parser import DocumentParser from ..registry import parser_registry @@ -48,29 +47,25 @@ def _extract_text(raw: bytes) -> tuple[list[str], list[ImageBlock]]: def _extract_markdown(raw: bytes) -> tuple[list[str], list[ImageBlock]]: - """Return Markdown per page + ``ImageBlock``s built from embedded data URIs. - - ``embed_images=True`` makes ``pymupdf4llm`` write images as base64 - data URIs in-line. We decode each ref into an ``ImageBlock`` and - leave the ref in the page text untouched so the caption stage can - substitute later via ``ImageBlock.metadata['markdown_ref']``. + """Return structured Markdown per page (no images). + + pymupdf is the lightweight, no-VLM backend. ``pymupdf4llm`` preserves + document structure (headings, lists, tables) — which the markdown-aware + chunker needs to cut on real boundaries instead of mid-sentence — while + ``embed_images=False`` keeps base64 image data out of the text. That keeps + chunks small (no Milvus gRPC overflow) and skips image rendering entirely + (fast). Image-aware parsing is marker/docling's job, so no ``ImageBlock``s + are produced here. """ with pymupdf.open(stream=raw, filetype="pdf") as doc: chunks = pymupdf4llm.to_markdown( doc, page_chunks=True, - embed_images=True, + embed_images=False, write_images=False, - dpi=300, ) - pages: list[str] = [] - images: list[ImageBlock] = [] - for i, chunk in enumerate(chunks, start=1): - text = (chunk.get("text") or "").strip() - pages.append(text) - if text: - images.extend(extract_data_uri_image_blocks(text, page_number=i)) - return pages, images + pages = [(chunk.get("text") or "").strip() for chunk in chunks] + return pages, [] @parser_registry.register("pymupdf") diff --git a/openrag/services/workers/indexer_pool.py b/openrag/services/workers/indexer_pool.py index f82643612..2361bef8c 100644 --- a/openrag/services/workers/indexer_pool.py +++ b/openrag/services/workers/indexer_pool.py @@ -35,6 +35,7 @@ def __init__(self) -> None: cfg = load_config() parser = build_parser_dispatcher(cfg) + parser_factory = _build_parser_factory(parser) vlm = build_caption_vlm(cfg) chunker = _build_chunker(cfg) embedder_factory = _build_embedder_factory(cfg) @@ -62,7 +63,9 @@ def __init__(self) -> None: vector_store=self._vector_store, vlm=vlm, image_captioning=cfg.loader.image_captioning, + timeouts=_build_pipeline_timeouts(cfg), chunker_factory=_build_chunker_from_config, + parser_factory=parser_factory, embedder_factory=embedder_factory, vlm_factory=vlm_factory, contextualizer_factory=contextualizer_factory, @@ -332,6 +335,39 @@ def _build_chunker_from_config(chunker_config: Any) -> Any: return _build_chunker(SimpleNamespace(chunker=chunker_config)) +def _build_parser_factory(parser: Any) -> Any: + """Factory honoring a preset's ``parsing_strategy`` for PDFs. + + Without this, the pipeline falls back to the single global-config dispatcher + and every PDF uses the global default loader — silently ignoring a preset's + pymupdf/docling choice. Each per-strategy wrapper reuses ``parser``'s shared + backend cache, so selecting a strategy never builds a duplicate + marker/docling Ray pool. + """ + cache: dict[str, Any] = {} + + def factory(strategy: str = "marker") -> Any: + wrapper = cache.get(strategy) + if wrapper is None: + wrapper = parser.for_pdf_strategy(strategy) + cache[strategy] = wrapper + return wrapper + + return factory + + +def _build_pipeline_timeouts(cfg: Settings) -> Any: + """Per-stage timeouts for the indexing pipeline. + + Bounds the parse stage at ``loader.parse_timeout`` so a wedged parse (notably + pymupdf, which has no internal timeout) fails that file instead of stalling + indexing. Other stages stay unbounded here (their backends self-limit). See #571. + """ + from services.workers.pipeline_builder import PipelineTimeouts + + return PipelineTimeouts(parse=cfg.loader.parse_timeout) + + def _build_embedder_factory(cfg: Settings) -> Any: from core.embeddings import embedder_registry diff --git a/openrag/services/workers/parsers/parser_dispatcher.py b/openrag/services/workers/parsers/parser_dispatcher.py index 713c4bd63..92e194d93 100644 --- a/openrag/services/workers/parsers/parser_dispatcher.py +++ b/openrag/services/workers/parsers/parser_dispatcher.py @@ -38,6 +38,10 @@ "OpenAIAudioLoader": "audio_client", } +# Backend names a preset ``parsing_strategy`` may select for PDFs (the values of +# _PDF_BACKENDS). Used to validate a strategy before routing a PDF to it. +_PDF_BACKEND_NAMES: frozenset[str] = frozenset(_PDF_BACKENDS.values()) + # Attachment ext (lowercased, no dot) → DocumentType, used to wire the EML # parser's per-attachment sub-parsers. _MAX_EML_ATTACHMENT_DEPTH = 3 @@ -84,6 +88,21 @@ async def parse(self, document: Document) -> ProcessedDocument: parser = self._get(backend) return await parser.parse(document) + def for_pdf_strategy(self, strategy: str) -> DocumentParser: + """Return a parser that forces ``strategy`` (a PDF backend name such as + ``"pymupdf"`` or ``"docling"``) for PDF documents while dispatching every + other content type exactly as this dispatcher does. + + Backends come from this dispatcher's shared cache, so honoring a + per-preset ``parsing_strategy`` never spins up a duplicate marker/docling + pool. + """ + if strategy not in _PDF_BACKEND_NAMES: + raise ValueError( + f"Unsupported PDF parsing strategy {strategy!r}; expected one of {sorted(_PDF_BACKEND_NAMES)}" + ) + return _PdfStrategyParser(self, strategy) + # ----- backend resolution ----- def _resolve_backend(self, content_type: DocumentType, ext: str) -> str: @@ -202,10 +221,32 @@ async def language_detector(path): # noqa: E731 - small adapter to the (path) - return _create("core.indexing.parsers.audio.client_based", "audio_client", client=client) +class _PdfStrategyParser(DocumentParser): + """Force a specific PDF backend (a preset's ``parsing_strategy``) for PDF + documents, delegating every other content type to the shared dispatcher so + the per-preset choice never duplicates a backend or its pool.""" + + def __init__(self, dispatcher: ParserDispatcher, pdf_backend: str) -> None: + self._dispatcher = dispatcher + self._pdf_backend = pdf_backend + + def supported_types(self) -> list[str]: + return self._dispatcher.supported_types() + + async def parse(self, document: Document) -> ProcessedDocument: + if document.content_type is DocumentType.PDF: + return await self._dispatcher._get(self._pdf_backend).parse(document) + return await self._dispatcher.parse(document) + + # Only backends that need a non-conventional module path (``pymupdf`` lives # under ``pdf/``) or injected dependencies (pooled/client/eml). Everything else # is built by convention in ``ParserDispatcher._build``. _BUILDERS: dict[str, Any] = { + # pymupdf is the lightweight, fast PDF backend. It builds in markdown mode + # (the default): pymupdf4llm preserves structure (headings/tables) for the + # markdown-aware chunker, with embed_images=False so no base64 bloats chunks + # and no image rendering happens. Images/captioning are marker/docling's job. "pymupdf": lambda d: _create("core.indexing.parsers.pdf.pymupdf", "pymupdf"), "eml": lambda d: d._build_eml(), "marker": lambda d: d._build_marker(), diff --git a/openrag/services/workers/stages/parse.py b/openrag/services/workers/stages/parse.py index 29f225aed..51fbfb8ca 100644 --- a/openrag/services/workers/stages/parse.py +++ b/openrag/services/workers/stages/parse.py @@ -39,4 +39,10 @@ async def _parse_with_timeout( document: Document, timeout: float | None, ) -> ProcessedDocument: - return await run_with_optional_timeout(lambda: parser.parse(document), timeout) + try: + return await run_with_optional_timeout(lambda: parser.parse(document), timeout) + except TimeoutError as exc: + # asyncio.wait_for raises a bare TimeoutError whose str() is empty; give + # the failed-file report (row["error"]) a message that names the file and + # the bound that was exceeded. + raise TimeoutError(f"parse timed out after {timeout:g}s for {document.filename!r}") from exc diff --git a/tests/unit/services/workers/parsers/test_parser_dispatcher.py b/tests/unit/services/workers/parsers/test_parser_dispatcher.py index 802bcfcf0..9923486c1 100644 --- a/tests/unit/services/workers/parsers/test_parser_dispatcher.py +++ b/tests/unit/services/workers/parsers/test_parser_dispatcher.py @@ -115,6 +115,41 @@ async def test_parse_dispatches_to_cached_backend() -> None: assert result.text_blocks[0].text == "ok" +@pytest.mark.asyncio +async def test_for_pdf_strategy_overrides_pdf_backend_and_shares_cache() -> None: + """A preset's ``parsing_strategy`` must override the global PDF backend for + PDFs while non-PDF types still dispatch normally — reusing the dispatcher's + cached backends (no duplicate pools).""" + disp = ParserDispatcher(_config(pdf="MarkerLoader")) # global default = marker + marker, pymupdf, text = _FakeParser(), _FakeParser(), _FakeParser() + disp._by_name.update({"marker": marker, "pymupdf": pymupdf, "text": text}) + + pdf_parser = disp.for_pdf_strategy("pymupdf") + + pdf = Document(filename="a.pdf", content_type=DocumentType.PDF, raw_bytes=b"%PDF-1.4") + await pdf_parser.parse(pdf) + assert pymupdf.seen is pdf # routed to the preset strategy, not the global marker + assert marker.seen is None + + txt = Document(filename="a.txt", content_type=DocumentType.TEXT, raw_bytes=b"hi") + await pdf_parser.parse(txt) + assert text.seen is txt # non-PDF content still dispatches by content type + + +def test_for_pdf_strategy_rejects_unknown_strategy() -> None: + with pytest.raises(ValueError, match="Unsupported PDF parsing strategy"): + ParserDispatcher(_config()).for_pdf_strategy("nope") + + +def test_pymupdf_backend_builds_in_markdown_mode_without_images() -> None: + """The lightweight pymupdf backend builds in markdown mode — structured text + for the markdown-aware chunker — but with embed_images=False so it never + inlines base64 images into chunk text (which bloats chunks / breaks Milvus + inserts). Images are marker/docling's job.""" + parser = ParserDispatcher(_config(pdf="PyMuPDFLoader"))._get("pymupdf") + assert getattr(parser, "_mode", None) == "markdown" + + def test_build_caption_vlm_requires_endpoint() -> None: # No VLM endpoint configured -> unavailable, regardless of the captioning flag. assert build_caption_vlm(_config(image_captioning=True, vlm_base_url="")) is None diff --git a/tests/unit/services/workers/stages/test_parse.py b/tests/unit/services/workers/stages/test_parse.py index 85b77e000..9d89891c6 100644 --- a/tests/unit/services/workers/stages/test_parse.py +++ b/tests/unit/services/workers/stages/test_parse.py @@ -61,6 +61,31 @@ async def test_parse_stage_marks_error_and_scrubs_credentials_when_parser_fails( assert "api_key" not in row +@pytest.mark.asyncio +async def test_parse_stage_times_out_a_wedged_parse_and_reports_the_file(): + """A slow/wedged parse must fail *that* file (with a descriptive, non-empty + error naming the file) instead of hanging the pipeline indefinitely (#571).""" + import asyncio + + class _WedgedParser(DocumentParser): + async def parse(self, document: Document) -> ProcessedDocument: + await asyncio.sleep(10) # longer than the stage timeout + raise AssertionError("should have timed out") + + def supported_types(self) -> list[str]: + return [DocumentType.TEXT.value] + + document = Document(id="doc-1", filename="wedged.pdf", content_type=DocumentType.TEXT, text="x") + row = {"document": document, "token": "secret"} + + with pytest.raises(TimeoutError, match=r"parse timed out after 0.05s for 'wedged.pdf'"): + await parse_stage(row, _WedgedParser(), timeout=0.05) + + assert row["stage"] == "parse_failed" + assert row["error"] == "parse timed out after 0.05s for 'wedged.pdf'" + assert "token" not in row + + @pytest.mark.asyncio async def test_parse_stage_requires_document_in_row(): row = {"api_key": "secret"} diff --git a/tests/unit/services/workers/test_indexer_pool.py b/tests/unit/services/workers/test_indexer_pool.py index d888108d8..c99ec872c 100644 --- a/tests/unit/services/workers/test_indexer_pool.py +++ b/tests/unit/services/workers/test_indexer_pool.py @@ -20,6 +20,18 @@ class _NonCallableChunker: chunk = None +def test_build_pipeline_timeouts_bounds_parse_from_config() -> None: + """The pipeline must bound the parse stage at loader.parse_timeout so a wedged + parse fails that file instead of stalling indexing (#571).""" + from services.workers.indexer_pool import _build_pipeline_timeouts + + cfg = SimpleNamespace(loader=SimpleNamespace(parse_timeout=42)) + + timeouts = _build_pipeline_timeouts(cfg) + + assert timeouts.parse == 42 + + def test_build_chunker_returns_native_chunker(monkeypatch: pytest.MonkeyPatch) -> None: import core.chunking.factory as factory from services.workers.indexer_pool import _build_chunker @@ -156,6 +168,28 @@ def test_build_contextualizer_factory_returns_factory_for_later_hydration(tmp_pa factory("default") +def test_build_parser_factory_delegates_to_strategy_and_caches() -> None: + # The parser factory must route a preset's parsing_strategy through the + # dispatcher's for_pdf_strategy (so pymupdf/docling are honored, not the + # global default) and cache per strategy so no backend/pool is duplicated. + from services.workers.indexer_pool import _build_parser_factory + + calls: list[str] = [] + + class _FakeDispatcher: + def for_pdf_strategy(self, strategy: str): + calls.append(strategy) + return SimpleNamespace(strategy=strategy) + + factory = _build_parser_factory(_FakeDispatcher()) + + first = factory("pymupdf") + assert first.strategy == "pymupdf" + assert factory("pymupdf") is first # cached: built once per strategy + assert factory("docling").strategy == "docling" + assert calls == ["pymupdf", "docling"] # no rebuild for the repeated strategy + + def test_contextualizer_factory_reads_live_registry(tmp_path) -> None: # The factory holds a live reference to cfg.models.llm: a name added to the # registry AFTER the factory is built (mimicking the indexer's lazy DB @@ -757,7 +791,7 @@ def model_copy(self, *, update): batch_size=32, embed_concurrency=2, ), - loader=SimpleNamespace(image_captioning=True), + loader=SimpleNamespace(image_captioning=True, parse_timeout=3600), vectordb=SimpleNamespace(collection_name="vdb_test"), rdb=RDBConfig(), ) From 8fdf4a64e4040e1f5663ee9eaf1e3daa3d5b64bc Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 26 Jun 2026 13:33:18 +0200 Subject: [PATCH 2/6] fix(parsers): create marker/docling Ray pools lazily for per-preset backends (#575) A per-preset parsing_strategy can select a backend that isn't the global default, which bootstrap never pre-warmed, so the loader's get-only ray.get_actor failed with 'Failed to look up actor'. Create the pool on first use via get_or_create_actor, with get_if_exists=True for race-safe concurrent creation. Covers DoclingPool and MarkerPool. --- openrag/services/workers/bootstrap.py | 5 ++- .../workers/parsers/docling_workers.py | 7 ++- .../workers/parsers/marker_workers.py | 7 ++- .../workers/parsers/test_pool_loaders.py | 44 +++++++++++++++++++ 4 files changed, 60 insertions(+), 3 deletions(-) create mode 100644 tests/unit/services/workers/parsers/test_pool_loaders.py diff --git a/openrag/services/workers/bootstrap.py b/openrag/services/workers/bootstrap.py index a6e7278bf..e55c62921 100644 --- a/openrag/services/workers/bootstrap.py +++ b/openrag/services/workers/bootstrap.py @@ -53,7 +53,10 @@ def get_or_create_actor(name, cls, namespace="openrag", remote_args=(), **option try: return ray.get_actor(name, namespace=namespace) except ValueError: - return cls.options(name=name, namespace=namespace, **options).remote(*remote_args) + # get_if_exists makes the create idempotent: if another caller (e.g. a + # second indexer actor selecting the same backend) wins the race, Ray + # returns the existing actor instead of raising "actor already exists". + return cls.options(name=name, namespace=namespace, get_if_exists=True, **options).remote(*remote_args) except Exception: raise diff --git a/openrag/services/workers/parsers/docling_workers.py b/openrag/services/workers/parsers/docling_workers.py index 1716516b4..9edddff93 100644 --- a/openrag/services/workers/parsers/docling_workers.py +++ b/openrag/services/workers/parsers/docling_workers.py @@ -120,7 +120,12 @@ class DoclingLoader(BasePooledParser): def __init__(self) -> None: self.config = load_config() - self.worker: DoclingPool = ray.get_actor("DoclingPool", namespace="openrag") + # Create the pool lazily if bootstrap didn't: bootstrap only pre-warms the + # globally-configured PDF backend, but a per-preset parsing_strategy can + # select docling even when the global default is marker (see #569/#575). + from services.workers.bootstrap import get_or_create_actor + + self.worker: DoclingPool = get_or_create_actor("DoclingPool", DoclingPool, lifetime="detached") def supported_types(self) -> list[str]: return [DocumentType.PDF.value] diff --git a/openrag/services/workers/parsers/marker_workers.py b/openrag/services/workers/parsers/marker_workers.py index b08ca8a48..085800450 100644 --- a/openrag/services/workers/parsers/marker_workers.py +++ b/openrag/services/workers/parsers/marker_workers.py @@ -361,7 +361,12 @@ class MarkerLoader(BasePooledParser): def __init__(self) -> None: self.config = load_config() - self.worker = ray.get_actor("MarkerPool", namespace="openrag") + # Lazily create the pool if bootstrap didn't (it only pre-warms the + # globally-configured PDF backend; a preset can select marker even when + # the global default is docling — see #569/#575). + from services.workers.bootstrap import get_or_create_actor + + self.worker = get_or_create_actor("MarkerPool", MarkerPool, lifetime="detached") def supported_types(self) -> list[str]: return [DocumentType.PDF.value] diff --git a/tests/unit/services/workers/parsers/test_pool_loaders.py b/tests/unit/services/workers/parsers/test_pool_loaders.py new file mode 100644 index 000000000..c5dc42817 --- /dev/null +++ b/tests/unit/services/workers/parsers/test_pool_loaders.py @@ -0,0 +1,44 @@ +"""The pooled PDF loaders must create their Ray pool lazily via +``get_or_create_actor`` rather than assuming bootstrap already created it. + +Bootstrap only pre-warms the *globally* configured PDF backend, but a per-preset +``parsing_strategy`` can select any backend at runtime (#569). A get-only +``ray.get_actor`` then fails with "Failed to look up actor 'DoclingPool'" +(#575); lazy creation makes whichever backend a preset picks work on first use. +""" + +from __future__ import annotations + +import pytest + + +@pytest.mark.parametrize( + ("module_name", "loader_name", "pool_name", "pool_attr"), + [ + ("services.workers.parsers.docling_workers", "DoclingLoader", "DoclingPool", "DoclingPool"), + ("services.workers.parsers.marker_workers", "MarkerLoader", "MarkerPool", "MarkerPool"), + ], +) +def test_pool_loader_lazily_creates_its_pool(monkeypatch, module_name, loader_name, pool_name, pool_attr): + import importlib + + import services.workers.bootstrap as bootstrap + + module = importlib.import_module(module_name) + Loader = getattr(module, loader_name) + PoolCls = getattr(module, pool_attr) + + calls: list[tuple] = [] + + def fake_get_or_create_actor(name, cls, **options): + calls.append((name, cls, options)) + return "pool-handle" + + monkeypatch.setattr(bootstrap, "get_or_create_actor", fake_get_or_create_actor) + # A get-only ray.get_actor must NOT be used anymore — fail loudly if it is. + monkeypatch.setattr(module.ray, "get_actor", lambda *a, **k: pytest.fail("loader used get-only ray.get_actor")) + + loader = Loader() + + assert loader.worker == "pool-handle" + assert calls == [(pool_name, PoolCls, {"lifetime": "detached"})] From 7fcdc30851d335b2da84cc3d8fbd6dcd8c5cc6d3 Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 26 Jun 2026 14:13:41 +0200 Subject: [PATCH 3/6] fix(config): validate parse_timeout > 0 It feeds asyncio.wait_for, so 0/negative would fail every parse immediately instead of disabling the bound. Reject at config load (CodeRabbit #582). --- openrag/core/config/indexation.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/openrag/core/config/indexation.py b/openrag/core/config/indexation.py index e280b6e63..152145f14 100644 --- a/openrag/core/config/indexation.py +++ b/openrag/core/config/indexation.py @@ -183,7 +183,9 @@ class LoaderConfig(ConfigMixin): # pymupdf backend has no internal timeout, so without this a wedged pymupdf # parse would stall indexing indefinitely. Bounds any backend (and any future # unbounded one); a slow/wedged parse fails *that* file instead of hanging. - parse_timeout: int = 3600 + # Must be > 0: it feeds asyncio.wait_for, so 0/negative would fail every + # parse immediately rather than disable the bound. + parse_timeout: int = Field(default=3600, gt=0) transcriber: TranscriberConfig = Field(default_factory=TranscriberConfig) openai: OpenAILoaderConfig = Field(default_factory=OpenAILoaderConfig) # Max depth of nested .eml-in-.eml attachments the EmlLoader will descend From 9eee08c71cdf7a30af49bb988d5f7ecaae70142c Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 26 Jun 2026 14:15:36 +0200 Subject: [PATCH 4/6] fix(parse): don't relabel internal TimeoutError when no outer bound is set When timeout is None the parse stage applies no asyncio.wait_for, so a TimeoutError can only be internal to the parser. Re-raise it as-is instead of relabeling it 'parse timed out after {timeout}s' (which also crashed formatting {timeout:g} on None). Test covers the timeout=None path. Addresses CodeRabbit #582. --- openrag/services/workers/stages/parse.py | 4 ++++ tests/unit/services/workers/stages/test_parse.py | 14 ++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/openrag/services/workers/stages/parse.py b/openrag/services/workers/stages/parse.py index 51fbfb8ca..22a43979a 100644 --- a/openrag/services/workers/stages/parse.py +++ b/openrag/services/workers/stages/parse.py @@ -42,6 +42,10 @@ async def _parse_with_timeout( try: return await run_with_optional_timeout(lambda: parser.parse(document), timeout) except TimeoutError as exc: + if timeout is None: + # No outer bound was applied, so this TimeoutError is internal to the + # parser — surface it as-is (and avoid formatting {timeout:g} on None). + raise # asyncio.wait_for raises a bare TimeoutError whose str() is empty; give # the failed-file report (row["error"]) a message that names the file and # the bound that was exceeded. diff --git a/tests/unit/services/workers/stages/test_parse.py b/tests/unit/services/workers/stages/test_parse.py index 9d89891c6..48c607804 100644 --- a/tests/unit/services/workers/stages/test_parse.py +++ b/tests/unit/services/workers/stages/test_parse.py @@ -96,3 +96,17 @@ async def test_parse_stage_requires_document_in_row(): assert row["stage"] == "parse_failed" assert row["error"] == "parse_stage row must contain a Document under 'document'" assert "api_key" not in row + + +@pytest.mark.asyncio +async def test_parse_stage_surfaces_internal_timeout_when_no_outer_bound(): + """With timeout=None, a TimeoutError raised inside the parser is internal — + it must surface as-is (not be relabeled or crash formatting None).""" + document = Document(id="doc-1", filename="x.pdf", content_type=DocumentType.TEXT, text="hi") + row = {"document": document} + + with pytest.raises(TimeoutError, match="internal parser timeout"): + await parse_stage(row, FakeParser(error=TimeoutError("internal parser timeout")), timeout=None) + + assert row["stage"] == "parse_failed" + assert row["error"] == "internal parser timeout" From a393336ae6a660dd17e8816bd8e774a8fe330ca7 Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 26 Jun 2026 14:17:58 +0200 Subject: [PATCH 5/6] test(parsers): assert pymupdf markdown mode emits no images/base64 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Strengthen the pymupdf test beyond _mode: build a PDF that contains an image and assert _extract_markdown returns no ImageBlocks and inlines no base64 data URI — catches a regression that re-enables embed_images. (CodeRabbit #582) --- .../workers/parsers/test_parser_dispatcher.py | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/unit/services/workers/parsers/test_parser_dispatcher.py b/tests/unit/services/workers/parsers/test_parser_dispatcher.py index 9923486c1..acc405ace 100644 --- a/tests/unit/services/workers/parsers/test_parser_dispatcher.py +++ b/tests/unit/services/workers/parsers/test_parser_dispatcher.py @@ -149,6 +149,28 @@ def test_pymupdf_backend_builds_in_markdown_mode_without_images() -> None: parser = ParserDispatcher(_config(pdf="PyMuPDFLoader"))._get("pymupdf") assert getattr(parser, "_mode", None) == "markdown" + # The "without images" contract: build a PDF that actually contains an image + # and confirm the markdown extractor produces NO ImageBlocks and inlines no + # base64 data URIs. Catches a regression that re-enables embed_images. + import io + + import pymupdf + from core.indexing.parsers.pdf.pymupdf import _extract_markdown + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (8, 8), "red").save(buf, format="PNG") + doc = pymupdf.open() + page = doc.new_page() + page.insert_text((72, 72), "Hello world.") + page.insert_image(pymupdf.Rect(0, 0, 8, 8), stream=buf.getvalue()) + raw = doc.tobytes() + doc.close() + + pages, images = _extract_markdown(raw) + assert images == [] # pymupdf must not extract/inline images + assert not any("data:image" in p for p in pages) + def test_build_caption_vlm_requires_endpoint() -> None: # No VLM endpoint configured -> unavailable, regardless of the captioning flag. From 8d5fdf0b500185aa16c8cd9e7a3683eefe0d8533 Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 26 Jun 2026 23:56:21 +0200 Subject: [PATCH 6/6] fix(presets): default preset inherits global PDFLoader instead of forcing marker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default indexation preset hardcoded parsing_strategy="marker", and the new per-preset parser_factory routed every PDF through that strategy — overriding the operator's global file_loaders.pdf (PDFLoader) choice. On a pymupdf-configured deployment this forced marker, lazily spinning up the Marker Ray pool and loading models inside the indexer actor; on a GPU-less/CPU runner the parse stage never completes and indexing hangs. Make parsing_strategy optional: None now means "inherit the global PDFLoader". The default preset omits it (so it follows the deployment's configured backend); named presets (legal/finance) keep their explicit marker opt-in. _select_parser defers to the global dispatcher when no strategy is set. --- openrag/api/routers/admin/presets.py | 12 ++++++------ openrag/core/config/indexation_pipeline.py | 12 ++++++++++-- .../services/orchestrators/preset_service.py | 18 ++++++++++++------ openrag/services/workers/pipeline_builder.py | 6 +++++- 4 files changed, 33 insertions(+), 15 deletions(-) diff --git a/openrag/api/routers/admin/presets.py b/openrag/api/routers/admin/presets.py index b279147cb..5893536d1 100644 --- a/openrag/api/routers/admin/presets.py +++ b/openrag/api/routers/admin/presets.py @@ -1,7 +1,5 @@ """Admin routes for the Phase 14 pipeline preset registry.""" -from typing import get_args - from api.dependencies.auth import require_admin from api.schemas.admin.preset_schemas import ( CreatePresetRequest, @@ -11,7 +9,7 @@ UpdatePresetRequest, ) from core.chunking import chunking_registry -from core.config.indexation_pipeline import IndexationPipelineConfig +from core.config.indexation_pipeline import PARSING_STRATEGIES from core.rerankers.registry import reranker_registry from core.retrieval import retriever_registry from di.providers import get_preset_service @@ -21,9 +19,11 @@ _DEFAULT_RERANKER_PROVIDERS = ["infinity", "openai"] -# Derived from the validated Literal so the exposed options can never drift from -# what IndexationPipelineConfig actually accepts. -_PARSING_STRATEGIES = list(get_args(IndexationPipelineConfig.model_fields["parsing_strategy"].annotation)) +# The selectable PDF backends. Shares the IndexationPipelineConfig constant so +# the exposed options can never drift from what the model accepts. ``None`` +# (inherit the global PDFLoader) is the field default, not an explicit choice, +# so it is not surfaced here. +_PARSING_STRATEGIES = list(PARSING_STRATEGIES) def _registered_or_default(registered: list[str], defaults: list[str]) -> list[str]: diff --git a/openrag/core/config/indexation_pipeline.py b/openrag/core/config/indexation_pipeline.py index 793137c5c..ad0e98079 100644 --- a/openrag/core/config/indexation_pipeline.py +++ b/openrag/core/config/indexation_pipeline.py @@ -12,6 +12,13 @@ from core.config.chunking import ChunkerConfig from pydantic import BaseModel, ConfigDict, Field +# PDF parsing backends a preset may explicitly select. ``None`` (the default) +# means "inherit the deployment's global PDFLoader" — a preset that doesn't care +# about PDF parsing follows the operator's global ``file_loaders.pdf`` choice +# instead of silently forcing one backend (and, with it, lazily spinning up that +# backend's Ray pool). See pipeline_builder._select_parser. +PARSING_STRATEGIES: tuple[str, ...] = ("pymupdf", "marker", "docling") + class IndexationPipelineConfig(BaseModel): """Indexation pipeline settings for one partition preset.""" @@ -19,7 +26,8 @@ class IndexationPipelineConfig(BaseModel): model_config = ConfigDict(extra="ignore") chunking: ChunkerConfig = Field(default_factory=ChunkerConfig) - parsing_strategy: Literal["pymupdf", "marker", "docling"] = "marker" + # None => inherit the global PDFLoader (see PARSING_STRATEGIES above). + parsing_strategy: Literal["pymupdf", "marker", "docling"] | None = None # VLM / image captioning vlm: str | None = None # endpoint name; None = use global default @@ -50,4 +58,4 @@ class IndexationPipelineConfig(BaseModel): topic_tagging_llm: str | None = None -__all__ = ["IndexationPipelineConfig"] +__all__ = ["IndexationPipelineConfig", "PARSING_STRATEGIES"] diff --git a/openrag/services/orchestrators/preset_service.py b/openrag/services/orchestrators/preset_service.py index 095f1cc17..f33825e53 100644 --- a/openrag/services/orchestrators/preset_service.py +++ b/openrag/services/orchestrators/preset_service.py @@ -28,15 +28,21 @@ _VALID_PRESET_TYPES = frozenset({"indexation", "retrieval"}) _DEFAULT_SEEDS: dict[str, dict[str, dict[str, Any]]] = { - # ``enable_contextualization`` is intentionally omitted from the ``default`` - # indexation preset below — it is injected at seed time from the global - # ``chunker.contextual_retrieval`` (``CONTEXTUAL_RETRIEVAL``) toggle in - # _finalize_seed, so the env flag still drives contextualization on fresh - # deployments. Named presets (legal/finance) keep their explicit choice. + # ``enable_contextualization`` and ``parsing_strategy`` are intentionally + # omitted from the ``default`` indexation preset below. The former is injected + # at seed time from the global ``chunker.contextual_retrieval`` + # (``CONTEXTUAL_RETRIEVAL``) toggle in _finalize_seed; the latter stays unset + # so the default preset inherits the global ``file_loaders.pdf`` (``PDFLoader``) + # backend at parse time. Named presets (legal/finance) keep their explicit choice. "indexation": { "default": { "chunking": {"name": "recursive_splitter", "chunk_size": 512, "chunk_overlap_rate": 0.2}, - "parsing_strategy": "marker", + # ``parsing_strategy`` is intentionally omitted so the default preset + # inherits the deployment's global PDFLoader (``file_loaders.pdf``) + # rather than forcing one PDF backend on every partition. A hardcoded + # value here would override the operator's global choice and lazily + # spin up that backend's Ray pool — e.g. forcing marker on a + # pymupdf-configured deployment. Named presets below opt in explicitly. "enable_image_captioning": True, "enable_entity_extraction": True, "enable_topic_tagging": True, diff --git a/openrag/services/workers/pipeline_builder.py b/openrag/services/workers/pipeline_builder.py index 817b3b5e5..b90c4c1f9 100644 --- a/openrag/services/workers/pipeline_builder.py +++ b/openrag/services/workers/pipeline_builder.py @@ -127,7 +127,11 @@ def _effective_indexation_config(self, row: MutableMapping[str, Any]) -> Indexat raise TypeError("indexation_config must be an IndexationPipelineConfig or dict") def _select_parser(self, config: IndexationPipelineConfig | None) -> DocumentParser: - if config is not None and self.parser_factory is not None: + # parsing_strategy is None => the preset doesn't override PDF parsing, so + # defer to the global dispatcher (self.parser), which routes PDFs to the + # deployment's configured file_loaders.pdf. Only an explicit strategy + # goes through the factory (and lazily builds that backend's pool). + if config is not None and self.parser_factory is not None and config.parsing_strategy is not None: return self.parser_factory(config.parsing_strategy) return self.parser