fix(indexer): honor preset parsing_strategy, lazy pools, pymupdf markdown, parse timeout

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")

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

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),

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")

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
 

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
 

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]

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]

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(),

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

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

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"})]

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