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

conf/config.yaml

@@ -232,6 +232,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/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]:

openrag/core/config/indexation.py

@@ -178,6 +178,14 @@ 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.
+    # 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

openrag/core/config/indexation_pipeline.py

@@ -12,14 +12,22 @@
 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."""
 
     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"]

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

openrag/services/workers/bootstrap.py

@@ -51,10 +51,10 @@ def wrapper(name, cls, namespace="openrag", remote_args=(), **options):
 def get_or_create_actor(name, cls, namespace="openrag", remote_args=(), **options):
     # ``get_if_exists`` makes get-or-create atomic inside Ray: concurrent callers
     # (e.g. the multiple Ray Serve replicas that all run worker bootstrap in their
-    # lifespan at once) either all attach to the same actor or exactly one creates
-    # it and the rest attach. The previous get_actor()-then-create pattern had a
-    # TOCTOU race — two replicas both saw "not found" and then collided on
-    # creation, crashing a replica's startup with "Failed to look up actor".
+    # lifespan at once, or a per-preset backend selecting the same pool) either all
+    # attach to the same actor or exactly one creates it and the rest attach. The
+    # previous get_actor()-then-create pattern had a TOCTOU race — two callers both
+    # saw "not found" and collided on creation, crashing with "Failed to look up actor".
     return cls.options(name=name, namespace=namespace, get_if_exists=True, **options).remote(*remote_args)
 
 

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/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
 

openrag/services/workers/stages/parse.py

@@ -39,4 +39,14 @@ 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:
+        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.
+        raise TimeoutError(f"parse timed out after {timeout:g}s for {document.filename!r}") from exc