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"