From 7db8d3e3013bceb2bcd02cf875b65fd9d275d5f3 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 21:44:06 +0200 Subject: [PATCH 1/4] fix: isolate PDF parse in a subprocess so a bad file can't OOM the pod The document processor crash-looped on one pathological PDF: pymupdf4llm's table/graphics detection over a page with ~1M vector path items ballooned past the 2 GiB pod limit. The parse ran in a thread, so nothing could interrupt or memory-bound it -- a single bad file OOM-killed the whole pod. Run the parse in an isolated worker subprocess (anyio.to_process, cancellable) with an RLIMIT_AS memory cap and a wall-clock timeout, so a pathological file fails THAT document instead of the pod (new document_processors/_isolation.py). Also pass graphics_limit (default 5000) to to_markdown -- validated to cut the known trigger page from 112 s to 23 s with bounded memory. On a permanent parse failure the processor returns success=False (instead of raising, which would retry 3x); vector/processor.py marks the placeholder "failed" and skips indexing, and the scanner stops re-queuing failed placeholders until the file changes -- so a doomed file no longer churns. New per-tenant (per-pod env) settings: DOCUMENT_PDF_GRAPHICS_LIMIT, DOCUMENT_PARSE_TIMEOUT_SECONDS, DOCUMENT_PARSE_MEM_LIMIT_MB. New metric astrolabe_document_parse_failed_total{reason=timeout|oom|error} surfaces hard failures that previously killed the process before any except ran. First PR of the tiered document-processor effort (Deck #199); tier 0/1/3 pipeline tracked separately. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 22 +++ .../document_processors/_isolation.py | 126 ++++++++++++ .../document_processors/pymupdf.py | 57 ++++-- nextcloud_mcp_server/observability/metrics.py | 20 ++ nextcloud_mcp_server/vector/processor.py | 40 +++- nextcloud_mcp_server/vector/scanner.py | 19 +- tests/unit/test_pdf_parse_isolation.py | 185 ++++++++++++++++++ 7 files changed, 450 insertions(+), 19 deletions(-) create mode 100644 nextcloud_mcp_server/document_processors/_isolation.py create mode 100644 tests/unit/test_pdf_parse_isolation.py diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 334508cb..bc608f7c 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -132,6 +132,10 @@ _DEFAULTS: dict[str, Any] = { # Document chunking "document_chunk_size": 2048, "document_chunk_overlap": 200, + # PDF parse isolation (OOM guard) + "document_pdf_graphics_limit": 5000, + "document_parse_timeout_seconds": 120, + "document_parse_mem_limit_mb": 1536, # Observability "metrics_enabled": True, "metrics_port": 9090, @@ -275,8 +279,11 @@ _dynaconf = Dynaconf( Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1), Validator("VERIFICATION_CONCURRENCY", gte=1), Validator("DOCUMENT_CHUNK_SIZE", gte=1), + Validator("DOCUMENT_PARSE_TIMEOUT_SECONDS", gte=1), + Validator("DOCUMENT_PARSE_MEM_LIMIT_MB", gte=128), # Non-negative Validator("DOCUMENT_CHUNK_OVERLAP", gte=0), + Validator("DOCUMENT_PDF_GRAPHICS_LIMIT", gte=0), # Non-empty strings Validator("VECTOR_SYNC_PDF_TAG", len_min=1), # Enum constraints @@ -701,6 +708,18 @@ class Settings: document_chunk_size: int = 2048 # Characters per chunk document_chunk_overlap: int = 200 # Overlapping characters between chunks + # PDF parse isolation (OOM guard). The parse runs in a subprocess so one + # pathological file fails that doc, not the pod. + document_pdf_graphics_limit: int = ( + 5000 # to_markdown graphics cap; pages above skip graphics analysis + ) + document_parse_timeout_seconds: int = ( + 120 # wall-clock cap per parse; the worker subprocess is killed on timeout + ) + document_parse_mem_limit_mb: int = ( + 1536 # RLIMIT_AS in the parse subprocess (kept below the pod memory limit) + ) + # Observability settings metrics_enabled: bool = True metrics_port: int = 9090 @@ -1309,6 +1328,9 @@ def get_settings() -> Settings: # Document chunking settings "document_chunk_size": "DOCUMENT_CHUNK_SIZE", "document_chunk_overlap": "DOCUMENT_CHUNK_OVERLAP", + "document_pdf_graphics_limit": "DOCUMENT_PDF_GRAPHICS_LIMIT", + "document_parse_timeout_seconds": "DOCUMENT_PARSE_TIMEOUT_SECONDS", + "document_parse_mem_limit_mb": "DOCUMENT_PARSE_MEM_LIMIT_MB", # Observability settings "metrics_enabled": "METRICS_ENABLED", "metrics_port": "METRICS_PORT", diff --git a/nextcloud_mcp_server/document_processors/_isolation.py b/nextcloud_mcp_server/document_processors/_isolation.py new file mode 100644 index 00000000..f452124f --- /dev/null +++ b/nextcloud_mcp_server/document_processors/_isolation.py @@ -0,0 +1,126 @@ +"""Subprocess isolation for PDF parsing. + +PDF text/markdown extraction (pymupdf + pymupdf4llm) is CPU-bound C code that can +balloon memory or hang on a pathological document -- e.g. a page with ~1M vector +path items drives pymupdf4llm's table detection past the pod memory limit and +OOM-kills the whole process. Running the parse in a worker *subprocess* (via +``anyio.to_process``) with an address-space rlimit and a wall-clock timeout means +one bad file fails *that document*, not the pod: an rlimit breach raises +``MemoryError`` in the worker, a hang is killed when the timeout cancels the call. + +The worker function is module-level (picklable) so ``anyio.to_process`` can run it +in its process pool. +""" + +import logging +import resource +from pathlib import Path +from typing import Any + +import anyio +import anyio.to_process +from anyio import BrokenWorkerProcess + +logger = logging.getLogger(__name__) + +# Guard so the address-space limit is applied once per (reused) worker process. +_MEM_LIMIT_APPLIED = False + + +class PdfParseFailed(Exception): + """A PDF parse failed in the isolated worker. + + ``reason`` is one of ``timeout`` | ``oom`` | ``error`` and maps directly to + the ``astrolabe_document_parse_failed_total{reason}`` metric label. + """ + + def __init__(self, reason: str, message: str | None = None) -> None: + self.reason = reason + super().__init__(message or f"PDF parse failed ({reason})") + + +def _apply_mem_limit(mem_limit_mb: int) -> None: + """Cap the worker's address space (RLIMIT_AS) so a bomb raises MemoryError. + + Applied once per worker process. The hard limit is left untouched (we only + lower the soft limit), and we never set a soft limit above the hard limit. + """ + global _MEM_LIMIT_APPLIED + if _MEM_LIMIT_APPLIED or mem_limit_mb <= 0: + return + target = mem_limit_mb * 1024 * 1024 + soft, hard = resource.getrlimit(resource.RLIMIT_AS) + soft_target = target if hard == resource.RLIM_INFINITY else min(target, hard) + resource.setrlimit(resource.RLIMIT_AS, (soft_target, hard)) + _MEM_LIMIT_APPLIED = True + + +def _parse_pdf_worker( + content: bytes, + write_images: bool, + image_path: str | None, + graphics_limit: int, + mem_limit_mb: int, +) -> list[dict[str, Any]]: + """Run pymupdf4llm.to_markdown in the worker subprocess (positional args only). + + Returns the ``page_chunks`` list (picklable dicts of text + metadata). Imports + pymupdf lazily so the parent process isn't forced to load them here. + """ + _apply_mem_limit(mem_limit_mb) + + # Imported inside the worker so the parent process (and any module that + # imports this one) doesn't load pymupdf4llm -- which prints a banner to + # stdout on import, the channel anyio's worker uses for IPC. + import pymupdf # noqa: PLC0415 + import pymupdf4llm # noqa: PLC0415 + + doc = pymupdf.open("pdf", content) + try: + # page_chunks=True makes to_markdown return list[dict], not str. + return pymupdf4llm.to_markdown( # type: ignore[return-value] + doc, + write_images=write_images, + image_path=image_path if write_images else None, + page_chunks=True, + graphics_limit=graphics_limit, + ) + finally: + doc.close() + + +async def run_isolated_pdf_parse( + content: bytes, + *, + write_images: bool, + image_path: Path | None, + graphics_limit: int, + timeout_seconds: float, + mem_limit_mb: int, +) -> list[dict[str, Any]]: + """Parse a PDF in an isolated worker subprocess with a memory cap and timeout. + + Raises ``PdfParseFailed`` (reason ``timeout`` | ``oom`` | ``error``) instead of + taking the pod down. On timeout the worker process is killed (``cancellable``). + """ + with anyio.move_on_after(timeout_seconds): + try: + return await anyio.to_process.run_sync( + _parse_pdf_worker, + content, + write_images, + str(image_path) if image_path is not None else None, + graphics_limit, + mem_limit_mb, + cancellable=True, + ) + except MemoryError as e: + raise PdfParseFailed("oom", str(e)) from e + except BrokenWorkerProcess as e: + # Worker died without a clean exception (e.g. SIGKILL from the OS OOM + # killer beating the rlimit). Treat as an out-of-memory failure. + raise PdfParseFailed("oom", str(e)) from e + except Exception as e: + raise PdfParseFailed("error", f"{type(e).__name__}: {e}") from e + # Reached only when move_on_after swallowed the timeout cancellation. + raise PdfParseFailed("timeout", f"parse exceeded {timeout_seconds}s") diff --git a/nextcloud_mcp_server/document_processors/pymupdf.py b/nextcloud_mcp_server/document_processors/pymupdf.py index 96e464fd..722fad40 100644 --- a/nextcloud_mcp_server/document_processors/pymupdf.py +++ b/nextcloud_mcp_server/document_processors/pymupdf.py @@ -8,13 +8,17 @@ from typing import Any, Optional import anyio -# NOTE: Do NOT call pymupdf.layout.activate() here! -# It changes the behavior of pymupdf4llm.to_markdown() when page_chunks=True, -# causing it to return a string instead of a list[dict]. +# pymupdf is used here only for the cheap metadata open. The heavy +# pymupdf4llm.to_markdown extraction runs in an isolated worker subprocess +# (see _isolation.py) so a pathological file can't OOM the pod. +# NOTE: Do NOT call pymupdf.layout.activate()! It changes the behavior of +# pymupdf4llm.to_markdown() when page_chunks=True (returns str, not list[dict]). # See: https://github.com/pymupdf/pymupdf4llm/issues/323 import pymupdf -import pymupdf4llm +from nextcloud_mcp_server.config import get_settings + +from ._isolation import PdfParseFailed, run_isolated_pdf_parse from .base import DocumentProcessor, ProcessingResult, ProcessorError logger = logging.getLogger(__name__) @@ -122,19 +126,44 @@ class PyMuPDFProcessor(DocumentProcessor): pdf_image_dir = self.image_dir / pdf_id pdf_image_dir.mkdir(exist_ok=True, parents=True) - # Extract all pages in a single call with page_chunks=True - def do_extract() -> list[dict[str, Any]]: - # When page_chunks=True, to_markdown returns list[dict] not str - return pymupdf4llm.to_markdown( # type: ignore[return-value] - doc, + # Extract all pages (page_chunks=True) in an isolated worker + # subprocess with a memory rlimit + wall-clock timeout, so a + # pathological file (e.g. a page with ~1M vector paths that drives + # table detection past the pod memory limit) fails THIS document + # instead of OOM-killing the pod. graphics_limit caps per-page + # vector-graphics analysis (the known trigger). + settings = get_settings() + try: + page_chunks: list[dict[str, Any]] = await run_isolated_pdf_parse( + content, write_images=self.extract_images, image_path=pdf_image_dir if self.extract_images else None, - page_chunks=True, + graphics_limit=settings.document_pdf_graphics_limit, + timeout_seconds=settings.document_parse_timeout_seconds, + mem_limit_mb=settings.document_parse_mem_limit_mb, + ) + except PdfParseFailed as exc: + doc.close() + logger.warning( + "Isolated PDF parse failed for %s (reason=%s): %s", + filename or "", + exc.reason, + exc, + extra={ + "processor": self.name, + "tier": self.tier, + "status": "error", + "reason": exc.reason, + }, + ) + metadata["parse_failed_reason"] = exc.reason + return ProcessingResult( + text="", + metadata=metadata, + processor=self.name, + success=False, + error=f"isolated parse failed ({exc.reason}): {exc}", ) - - page_chunks: list[dict[str, Any]] = await anyio.to_thread.run_sync( # type: ignore[attr-defined] - do_extract - ) if progress_callback: await progress_callback(90, 100, "Building result") diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 7b9221cd..7cd40e4f 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -261,6 +261,17 @@ document_escalation_total = Counter( ["from_tier", "to_tier", "reason"], ) +# Hard parse failures: the parse now runs in an isolated subprocess, so a +# timeout/OOM that kills the worker is caught here. This is distinct from +# ``document_parse_total{status="error"}`` (an in-process exception): a hard +# OOM previously killed the pod before any except ran, so it incremented +# nothing -- this counter makes those failures visible. +document_parse_failed_total = Counter( + "astrolabe_document_parse_failed_total", + "Document parses that failed in the isolated worker (process killed)", + ["reason"], # reason: timeout | oom | error +) + # --- Embedding stages --------------------------------------------------------- embedding_duration_seconds = Histogram( @@ -617,6 +628,15 @@ def record_document_escalation(from_tier: str, to_tier: str, reason: str) -> Non ).inc() +def record_document_parse_failed(reason: str) -> None: + """Record a hard parse failure from the isolated worker. + + Args: + reason: ``timeout`` | ``oom`` | ``error`` + """ + document_parse_failed_total.labels(reason=reason).inc() + + def record_embedding( kind: str, provider: str, diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 4cc207bc..cd2b329b 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -21,6 +21,7 @@ from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_servi from nextcloud_mcp_server.models.deck import DeckCard from nextcloud_mcp_server.observability.metrics import ( record_document_chunks, + record_document_parse_failed, record_embedding, record_qdrant_operation, record_vector_sync_processing, @@ -31,7 +32,10 @@ from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter from nextcloud_mcp_server.vector import payload_keys from nextcloud_mcp_server.vector.document_chunker import DocumentChunker from nextcloud_mcp_server.vector.html_processor import html_to_markdown -from nextcloud_mcp_server.vector.placeholder import delete_placeholder_point +from nextcloud_mcp_server.vector.placeholder import ( + delete_placeholder_point, + update_placeholder_status, +) from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.scanner import DocumentTask from nextcloud_mcp_server.vector.sharing_state import ( @@ -525,6 +529,40 @@ async def _index_document( content_type=content_type, filename=file_path, ) + + # A permanent parse failure (e.g. an isolated-worker OOM/timeout + # on a pathological PDF) returns success=False rather than + # raising -- there is nothing to index and retrying would just + # fail again. Mark the placeholder "failed" so the scanner stops + # re-queuing it (until the file changes) and return without + # indexing empty content. + if not result.success: + reason = result.metadata.get("parse_failed_reason", "error") + record_document_parse_failed(reason) + logger.warning( + "Permanent parse failure for %s (reason=%s); marking " + "failed and skipping index", + file_path, + reason, + ) + try: + await update_placeholder_status( + doc_id=doc_task.doc_id, + doc_type=doc_task.doc_type, + user_id=doc_task.user_id, + status="failed", + ) + except Exception: + # Best-effort: a transient Qdrant error here only means + # the placeholder isn't marked, so the scanner retries + # the (still un-indexable) file later -- not fatal. + logger.debug( + "Could not mark placeholder failed for %s", + doc_task.doc_id, + exc_info=True, + ) + return + content = result.text file_metadata = result.metadata title = file_metadata.get("title") or file_path.split("/")[-1] diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index 7108bffa..8bb489cf 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -526,13 +526,24 @@ async def scan_user_documents( # File modified since last indexing needs_indexing = True elif existing_metadata.get("is_placeholder", False): - # Placeholder exists - check if it's stale (processing may have failed) - # Only requeue if placeholder is older than 5x scan interval - # (Large PDFs can take 3-4 minutes to process) + # Placeholder exists - check its status / staleness. queued_at = existing_metadata.get("queued_at", 0) placeholder_age = time.time() - queued_at stale_threshold = get_settings().vector_sync_scan_interval * 5 - if placeholder_age > stale_threshold: + if existing_metadata.get("status") == "failed": + # A permanent parse failure (e.g. an isolated-worker + # OOM/timeout on a pathological PDF). Don't keep + # re-queuing an unchanged file that will just fail + # again -- the modified_at branch above still retries + # it once the file actually changes. + logger.debug( + "Skipping file %s (ID: %s): previous parse failed permanently", + file_path, + file_id, + ) + elif placeholder_age > stale_threshold: + # Only requeue if placeholder is older than 5x scan + # interval (large PDFs can take minutes to process). logger.debug( "Found stale placeholder for file %s (ID: %s) (age=%ss), requeuing", file_path, diff --git a/tests/unit/test_pdf_parse_isolation.py b/tests/unit/test_pdf_parse_isolation.py new file mode 100644 index 00000000..201e682b --- /dev/null +++ b/tests/unit/test_pdf_parse_isolation.py @@ -0,0 +1,185 @@ +"""Unit tests for the isolated PDF parse (OOM hotfix). + +The parse runs in a worker subprocess (``anyio.to_process``) with a memory +rlimit and a wall-clock timeout so a pathological PDF fails *that document* +instead of OOM-killing the pod. These tests pin: + * the failure classification (oom / timeout / error) of the async wrapper; + * ``_apply_mem_limit`` rlimit computation (mocked, never applied in-process); + * the PyMuPDF processor wiring: settings forwarded, success path, and a + graceful ``success=False`` result on a permanent parse failure. + +The real subprocess + rlimit enforcement is exercised by the local end-to-end +check on the sample PDFs, not here (unit tests must not spawn the heavy worker +or depend on the sample files). +""" + +import resource + +import anyio +import anyio.to_process +import pymupdf +import pytest +from anyio import BrokenWorkerProcess + +from nextcloud_mcp_server.document_processors import _isolation +from nextcloud_mcp_server.document_processors._isolation import ( + PdfParseFailed, + run_isolated_pdf_parse, +) + +pytestmark = pytest.mark.unit + + +def _tiny_pdf() -> bytes: + doc = pymupdf.open() + page = doc.new_page(width=595, height=842) + page.insert_text((50, 50), "Hello world") + data: bytes = doc.tobytes() + doc.close() + return data + + +async def _run(monkeypatch, fake_run_sync) -> list: + monkeypatch.setattr(anyio.to_process, "run_sync", fake_run_sync) + return await run_isolated_pdf_parse( + b"%PDF-1.7", + write_images=False, + image_path=None, + graphics_limit=5000, + timeout_seconds=5, + mem_limit_mb=1536, + ) + + +# --- failure classification of the async wrapper ---------------------------- + + +async def test_success_returns_worker_value(monkeypatch): + page_chunks = [{"text": "ok", "metadata": {"page": 1}}] + + async def fake(*args, **kwargs): + return page_chunks + + assert await _run(monkeypatch, fake) == page_chunks + + +async def test_memory_error_classified_as_oom(monkeypatch): + async def fake(*args, **kwargs): + raise MemoryError("rlimit hit") + + with pytest.raises(PdfParseFailed) as exc: + await _run(monkeypatch, fake) + assert exc.value.reason == "oom" + + +async def test_broken_worker_classified_as_oom(monkeypatch): + async def fake(*args, **kwargs): + raise BrokenWorkerProcess("worker died") + + with pytest.raises(PdfParseFailed) as exc: + await _run(monkeypatch, fake) + assert exc.value.reason == "oom" + + +async def test_other_exception_classified_as_error(monkeypatch): + async def fake(*args, **kwargs): + raise ValueError("not a pdf") + + with pytest.raises(PdfParseFailed) as exc: + await _run(monkeypatch, fake) + assert exc.value.reason == "error" + + +async def test_timeout_kills_and_classifies_as_timeout(monkeypatch): + async def fake(*args, **kwargs): + # Simulate a hung worker; the move_on_after timeout must win. + await anyio.sleep(30) + + monkeypatch.setattr(anyio.to_process, "run_sync", fake) + with pytest.raises(PdfParseFailed) as exc: + await run_isolated_pdf_parse( + b"%PDF-1.7", + write_images=False, + image_path=None, + graphics_limit=5000, + timeout_seconds=0.2, + mem_limit_mb=1536, + ) + assert exc.value.reason == "timeout" + + +# --- _apply_mem_limit computation (mocked; never applied to the test proc) --- + + +def test_apply_mem_limit_caps_soft_below_finite_hard(monkeypatch): + captured = {} + monkeypatch.setattr(_isolation, "_MEM_LIMIT_APPLIED", False) + monkeypatch.setattr( + _isolation.resource, + "getrlimit", + lambda _w: (resource.RLIM_INFINITY, 4 * 1024**3), + ) + monkeypatch.setattr( + _isolation.resource, "setrlimit", lambda _w, pair: captured.update(pair=pair) + ) + # target = 1536 MiB < hard (4 GiB) -> soft becomes the target, hard untouched + _isolation._apply_mem_limit(1536) + soft, hard = captured["pair"] + assert soft == 1536 * 1024 * 1024 + assert hard == 4 * 1024**3 + + +def test_apply_mem_limit_is_applied_once(monkeypatch): + calls = [] + monkeypatch.setattr(_isolation, "_MEM_LIMIT_APPLIED", False) + monkeypatch.setattr( + _isolation.resource, + "getrlimit", + lambda _w: (resource.RLIM_INFINITY, resource.RLIM_INFINITY), + ) + monkeypatch.setattr(_isolation.resource, "setrlimit", lambda *a: calls.append(a)) + _isolation._apply_mem_limit(1536) + _isolation._apply_mem_limit(1536) + assert len(calls) == 1 # second call is a no-op + + +# --- PyMuPDF processor wiring ------------------------------------------------ + + +async def test_processor_success_builds_page_boundaries(monkeypatch): + from nextcloud_mcp_server.document_processors import pymupdf as pymupdf_proc + + seen = {} + + async def fake_parse(content, **kwargs): + seen.update(kwargs) + return [{"text": "Hello world", "metadata": {"page": 1}}] + + monkeypatch.setattr(pymupdf_proc, "run_isolated_pdf_parse", fake_parse) + + proc = pymupdf_proc.PyMuPDFProcessor(extract_images=False) + result = await proc.process(_tiny_pdf(), "application/pdf", filename="t.pdf") + + assert result.success is True + assert "Hello world" in result.text + assert result.metadata["page_boundaries"][0]["page"] == 1 + # settings forwarded to the isolated parse + assert seen["graphics_limit"] == 5000 + assert seen["timeout_seconds"] == 120 + assert seen["mem_limit_mb"] == 1536 + + +async def test_processor_parse_failure_returns_success_false(monkeypatch): + from nextcloud_mcp_server.document_processors import pymupdf as pymupdf_proc + + async def fake_parse(content, **kwargs): + raise PdfParseFailed("oom", "killed") + + monkeypatch.setattr(pymupdf_proc, "run_isolated_pdf_parse", fake_parse) + + proc = pymupdf_proc.PyMuPDFProcessor(extract_images=False) + result = await proc.process(_tiny_pdf(), "application/pdf", filename="bomb.pdf") + + assert result.success is False + assert result.text == "" + assert result.metadata["parse_failed_reason"] == "oom" From 7ec116a3c7013706a8207be1e9e8cd582902f00b Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 21:57:55 +0200 Subject: [PATCH 2/4] fix(review): close doc via try/finally; don't count parse failures as indexed Address PR #852 review: - pymupdf.py: the metadata `doc` was only closed on the PdfParseFailed and success paths, so a failure in `_extract_metadata`/`mkdir`/`get_settings` leaked it. `doc` is only needed for metadata + page_count (the heavy parse works from `content` bytes in the worker), so open it, read metadata, and close it immediately under try/finally; drop the two later doc.close() calls. - processor.py: a permanent parse failure early-returned from `_index_document`, after which `process_document` still recorded record_qdrant_operation("upsert", "success") + record_vector_sync_processing(success) -- counting an OOM/timeout bomb as astrolabe_documents_indexed_total{status="success"}. `_index_document` now returns False on that path and the caller skips the success metrics (the failure is already recorded via document_parse_failed_total + the registry's document_parse_total{error}). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../document_processors/pymupdf.py | 19 ++++++++++--------- nextcloud_mcp_server/vector/processor.py | 16 ++++++++++++---- 2 files changed, 22 insertions(+), 13 deletions(-) diff --git a/nextcloud_mcp_server/document_processors/pymupdf.py b/nextcloud_mcp_server/document_processors/pymupdf.py index 722fad40..d1262b02 100644 --- a/nextcloud_mcp_server/document_processors/pymupdf.py +++ b/nextcloud_mcp_server/document_processors/pymupdf.py @@ -107,14 +107,19 @@ class PyMuPDFProcessor(DocumentProcessor): if progress_callback: await progress_callback(0, 100, "Opening PDF document") - # Open document and extract metadata in thread + # Open document only to read metadata + page count, then close it + # immediately (try/finally so a failure in _extract_metadata can't + # leak it). The heavy extraction below works from ``content`` bytes + # in the isolated worker, so ``doc`` is not needed past this point. doc = await anyio.to_thread.run_sync( # type: ignore[attr-defined] lambda: pymupdf.open("pdf", content) ) - - metadata = self._extract_metadata(doc, filename) - metadata["file_size"] = len(content) - page_count = doc.page_count + try: + metadata = self._extract_metadata(doc, filename) + metadata["file_size"] = len(content) + page_count = doc.page_count + finally: + doc.close() if progress_callback: await progress_callback(10, 100, f"Extracting {page_count} pages") @@ -143,7 +148,6 @@ class PyMuPDFProcessor(DocumentProcessor): mem_limit_mb=settings.document_parse_mem_limit_mb, ) except PdfParseFailed as exc: - doc.close() logger.warning( "Isolated PDF parse failed for %s (reason=%s): %s", filename or "", @@ -198,9 +202,6 @@ class PyMuPDFProcessor(DocumentProcessor): metadata["image_paths"] = image_paths metadata["page_boundaries"] = page_boundaries - # Close document - doc.close() - if progress_callback: await progress_callback(100, 100, "Processing complete") diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index cd2b329b..47138344 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -237,7 +237,15 @@ async def process_document( for attempt in range(max_retries): try: - await _index_document(doc_task, nc_client, qdrant_client) + indexed = await _index_document(doc_task, nc_client, qdrant_client) + + # A permanent parse failure returns False: it was already + # recorded (document_parse_failed_total + the registry's + # document_parse_total{error}) and the placeholder marked + # "failed". It is not an indexing event and not retryable, so + # don't count it as a successful upsert/indexed document. + if indexed is False: + return # Record successful processing metrics duration = time.time() - start_time @@ -534,8 +542,8 @@ async def _index_document( # on a pathological PDF) returns success=False rather than # raising -- there is nothing to index and retrying would just # fail again. Mark the placeholder "failed" so the scanner stops - # re-queuing it (until the file changes) and return without - # indexing empty content. + # re-queuing it (until the file changes) and return False so the + # caller skips the success metrics (it was not indexed). if not result.success: reason = result.metadata.get("parse_failed_reason", "error") record_document_parse_failed(reason) @@ -561,7 +569,7 @@ async def _index_document( doc_task.doc_id, exc_info=True, ) - return + return False content = result.text file_metadata = result.metadata From 6589e8e7fcb58c3a32bc17f7514840227e03bf3e Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 22:11:13 +0200 Subject: [PATCH 3/4] fix(review): require graphics_limit>=1, type _index_document, cover rlimit branch Address PR #852 round 2: - config: DOCUMENT_PDF_GRAPHICS_LIMIT validator is now gte=1 (pymupdf4llm treats 0 as "no cap", which would re-expose the OOM); documented the zero semantics and that the per-worker mem rlimit needs a pod restart to change. - processor: annotate `_index_document -> bool | None` and document the contract so the `if indexed is False` check is explicit/type-checkable. - tests: add the RLIM_INFINITY-hard branch assertion for _apply_mem_limit (soft==target, hard stays unbounded). Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 21 +++++++++++---------- nextcloud_mcp_server/vector/processor.py | 5 ++++- tests/unit/test_pdf_parse_isolation.py | 18 ++++++++++++++++++ 3 files changed, 33 insertions(+), 11 deletions(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index bc608f7c..a2bb3a27 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -281,9 +281,11 @@ _dynaconf = Dynaconf( Validator("DOCUMENT_CHUNK_SIZE", gte=1), Validator("DOCUMENT_PARSE_TIMEOUT_SECONDS", gte=1), Validator("DOCUMENT_PARSE_MEM_LIMIT_MB", gte=128), + # >=1: pymupdf4llm treats graphics_limit=0 as "no cap", which would + # re-expose the OOM this guards against. + Validator("DOCUMENT_PDF_GRAPHICS_LIMIT", gte=1), # Non-negative Validator("DOCUMENT_CHUNK_OVERLAP", gte=0), - Validator("DOCUMENT_PDF_GRAPHICS_LIMIT", gte=0), # Non-empty strings Validator("VECTOR_SYNC_PDF_TAG", len_min=1), # Enum constraints @@ -710,15 +712,14 @@ class Settings: # PDF parse isolation (OOM guard). The parse runs in a subprocess so one # pathological file fails that doc, not the pod. - document_pdf_graphics_limit: int = ( - 5000 # to_markdown graphics cap; pages above skip graphics analysis - ) - document_parse_timeout_seconds: int = ( - 120 # wall-clock cap per parse; the worker subprocess is killed on timeout - ) - document_parse_mem_limit_mb: int = ( - 1536 # RLIMIT_AS in the parse subprocess (kept below the pod memory limit) - ) + # to_markdown graphics cap; pages above it skip graphics analysis. Must be + # >=1 -- pymupdf4llm treats 0 as "no cap", which re-exposes the OOM. + document_pdf_graphics_limit: int = 5000 + # wall-clock cap per parse; the worker subprocess is killed on timeout. + document_parse_timeout_seconds: int = 120 + # RLIMIT_AS in the parse subprocess (below the pod limit). Applied once per + # worker for its lifetime, so changing it needs a pod restart. + document_parse_mem_limit_mb: int = 1536 # Observability settings metrics_enabled: bool = True diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 47138344..6a825662 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -311,10 +311,13 @@ async def process_document( async def _index_document( doc_task: DocumentTask, nc_client: NextcloudClient, qdrant_client -): +) -> bool | None: """ Index a single document (called by process_document with retry). + Returns ``False`` when a permanent parse failure means nothing was indexed + (the caller must then skip the success metrics); ``None`` otherwise. + Args: doc_task: Document task to index nc_client: Authenticated Nextcloud client diff --git a/tests/unit/test_pdf_parse_isolation.py b/tests/unit/test_pdf_parse_isolation.py index 201e682b..7daf90ce 100644 --- a/tests/unit/test_pdf_parse_isolation.py +++ b/tests/unit/test_pdf_parse_isolation.py @@ -129,6 +129,24 @@ def test_apply_mem_limit_caps_soft_below_finite_hard(monkeypatch): assert hard == 4 * 1024**3 +def test_apply_mem_limit_uses_target_when_hard_unlimited(monkeypatch): + captured = {} + monkeypatch.setattr(_isolation, "_MEM_LIMIT_APPLIED", False) + monkeypatch.setattr( + _isolation.resource, + "getrlimit", + lambda _w: (resource.RLIM_INFINITY, resource.RLIM_INFINITY), + ) + monkeypatch.setattr( + _isolation.resource, "setrlimit", lambda _w, pair: captured.update(pair=pair) + ) + # hard is unbounded -> soft is exactly the target, hard stays RLIM_INFINITY + _isolation._apply_mem_limit(1536) + soft, hard = captured["pair"] + assert soft == 1536 * 1024 * 1024 + assert hard == resource.RLIM_INFINITY + + def test_apply_mem_limit_is_applied_once(monkeypatch): calls = [] monkeypatch.setattr(_isolation, "_MEM_LIMIT_APPLIED", False) From 93f0f4f881d504293ddfa2da7a9fe694115cb3d4 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 22:20:06 +0200 Subject: [PATCH 4/4] fix(review): type timeout as float; document worker reuse + identity check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR #852 round 3 (all 🟡, no blockers): - config: DOCUMENT_PARSE_TIMEOUT_SECONDS is now float (default 120.0) so a fractional value is honoured rather than silently stored in an int field; matches anyio.move_on_after's float seconds. - _isolation: comment that a clean rlimit MemoryError leaves the worker alive in anyio's pool (vs the SIGKILL/BrokenWorkerProcess path that respawns) -- acceptable since RLIMIT_AS caps virtual address space, not RSS. - processor: note the `if indexed is False` is a deliberate identity check -- a successful index (incl. dedup hit) returns None and must not be mistaken for a parse failure. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 6 ++++-- nextcloud_mcp_server/document_processors/_isolation.py | 5 +++++ nextcloud_mcp_server/vector/processor.py | 3 +++ 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index a2bb3a27..08a79081 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -134,7 +134,7 @@ _DEFAULTS: dict[str, Any] = { "document_chunk_overlap": 200, # PDF parse isolation (OOM guard) "document_pdf_graphics_limit": 5000, - "document_parse_timeout_seconds": 120, + "document_parse_timeout_seconds": 120.0, "document_parse_mem_limit_mb": 1536, # Observability "metrics_enabled": True, @@ -716,7 +716,9 @@ class Settings: # >=1 -- pymupdf4llm treats 0 as "no cap", which re-exposes the OOM. document_pdf_graphics_limit: int = 5000 # wall-clock cap per parse; the worker subprocess is killed on timeout. - document_parse_timeout_seconds: int = 120 + # float so a fractional DOCUMENT_PARSE_TIMEOUT_SECONDS is honoured, matching + # anyio.move_on_after's float seconds. + document_parse_timeout_seconds: float = 120.0 # RLIMIT_AS in the parse subprocess (below the pod limit). Applied once per # worker for its lifetime, so changing it needs a pod restart. document_parse_mem_limit_mb: int = 1536 diff --git a/nextcloud_mcp_server/document_processors/_isolation.py b/nextcloud_mcp_server/document_processors/_isolation.py index f452124f..f030bbe4 100644 --- a/nextcloud_mcp_server/document_processors/_isolation.py +++ b/nextcloud_mcp_server/document_processors/_isolation.py @@ -115,6 +115,11 @@ async def run_isolated_pdf_parse( cancellable=True, ) except MemoryError as e: + # A clean rlimit breach: the worker raised MemoryError and stays + # ALIVE in anyio's pool (unlike the BrokenWorkerProcess/SIGKILL path, + # which spawns a fresh worker). Its heap may be slightly fragmented + # for the next document. Acceptable: RLIMIT_AS caps virtual address + # space (not RSS), so practical fragmentation risk is low. raise PdfParseFailed("oom", str(e)) from e except BrokenWorkerProcess as e: # Worker died without a clean exception (e.g. SIGKILL from the OS OOM diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 6a825662..73a8c0c2 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -244,6 +244,9 @@ async def process_document( # document_parse_total{error}) and the placeholder marked # "failed". It is not an indexing event and not retryable, so # don't count it as a successful upsert/indexed document. + # Identity check, not `if not indexed`: a successful index + # (including a dedup hit) returns None, which must NOT be + # treated as a parse failure. if indexed is False: return