Merge pull request #852 from cbcoutinho/fix/pdf-parse-oom-isolation
fix: isolate PDF parse in a subprocess so a pathological file can't OOM the pod
This commit is contained in:
@@ -132,6 +132,10 @@ _DEFAULTS: dict[str, Any] = {
|
|||||||
# Document chunking
|
# Document chunking
|
||||||
"document_chunk_size": 2048,
|
"document_chunk_size": 2048,
|
||||||
"document_chunk_overlap": 200,
|
"document_chunk_overlap": 200,
|
||||||
|
# PDF parse isolation (OOM guard)
|
||||||
|
"document_pdf_graphics_limit": 5000,
|
||||||
|
"document_parse_timeout_seconds": 120.0,
|
||||||
|
"document_parse_mem_limit_mb": 1536,
|
||||||
# Observability
|
# Observability
|
||||||
"metrics_enabled": True,
|
"metrics_enabled": True,
|
||||||
"metrics_port": 9090,
|
"metrics_port": 9090,
|
||||||
@@ -275,6 +279,11 @@ _dynaconf = Dynaconf(
|
|||||||
Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1),
|
Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1),
|
||||||
Validator("VERIFICATION_CONCURRENCY", gte=1),
|
Validator("VERIFICATION_CONCURRENCY", gte=1),
|
||||||
Validator("DOCUMENT_CHUNK_SIZE", gte=1),
|
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
|
# Non-negative
|
||||||
Validator("DOCUMENT_CHUNK_OVERLAP", gte=0),
|
Validator("DOCUMENT_CHUNK_OVERLAP", gte=0),
|
||||||
# Non-empty strings
|
# Non-empty strings
|
||||||
@@ -701,6 +710,19 @@ class Settings:
|
|||||||
document_chunk_size: int = 2048 # Characters per chunk
|
document_chunk_size: int = 2048 # Characters per chunk
|
||||||
document_chunk_overlap: int = 200 # Overlapping characters between chunks
|
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.
|
||||||
|
# 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.
|
||||||
|
# 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
|
||||||
|
|
||||||
# Observability settings
|
# Observability settings
|
||||||
metrics_enabled: bool = True
|
metrics_enabled: bool = True
|
||||||
metrics_port: int = 9090
|
metrics_port: int = 9090
|
||||||
@@ -1309,6 +1331,9 @@ def get_settings() -> Settings:
|
|||||||
# Document chunking settings
|
# Document chunking settings
|
||||||
"document_chunk_size": "DOCUMENT_CHUNK_SIZE",
|
"document_chunk_size": "DOCUMENT_CHUNK_SIZE",
|
||||||
"document_chunk_overlap": "DOCUMENT_CHUNK_OVERLAP",
|
"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
|
# Observability settings
|
||||||
"metrics_enabled": "METRICS_ENABLED",
|
"metrics_enabled": "METRICS_ENABLED",
|
||||||
"metrics_port": "METRICS_PORT",
|
"metrics_port": "METRICS_PORT",
|
||||||
|
|||||||
@@ -0,0 +1,131 @@
|
|||||||
|
"""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:
|
||||||
|
# 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
|
||||||
|
# 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")
|
||||||
@@ -8,13 +8,17 @@ from typing import Any, Optional
|
|||||||
|
|
||||||
import anyio
|
import anyio
|
||||||
|
|
||||||
# NOTE: Do NOT call pymupdf.layout.activate() here!
|
# pymupdf is used here only for the cheap metadata open. The heavy
|
||||||
# It changes the behavior of pymupdf4llm.to_markdown() when page_chunks=True,
|
# pymupdf4llm.to_markdown extraction runs in an isolated worker subprocess
|
||||||
# causing it to return a string instead of a list[dict].
|
# (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
|
# See: https://github.com/pymupdf/pymupdf4llm/issues/323
|
||||||
import pymupdf
|
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
|
from .base import DocumentProcessor, ProcessingResult, ProcessorError
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -103,14 +107,19 @@ class PyMuPDFProcessor(DocumentProcessor):
|
|||||||
if progress_callback:
|
if progress_callback:
|
||||||
await progress_callback(0, 100, "Opening PDF document")
|
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]
|
doc = await anyio.to_thread.run_sync( # type: ignore[attr-defined]
|
||||||
lambda: pymupdf.open("pdf", content)
|
lambda: pymupdf.open("pdf", content)
|
||||||
)
|
)
|
||||||
|
try:
|
||||||
metadata = self._extract_metadata(doc, filename)
|
metadata = self._extract_metadata(doc, filename)
|
||||||
metadata["file_size"] = len(content)
|
metadata["file_size"] = len(content)
|
||||||
page_count = doc.page_count
|
page_count = doc.page_count
|
||||||
|
finally:
|
||||||
|
doc.close()
|
||||||
|
|
||||||
if progress_callback:
|
if progress_callback:
|
||||||
await progress_callback(10, 100, f"Extracting {page_count} pages")
|
await progress_callback(10, 100, f"Extracting {page_count} pages")
|
||||||
@@ -122,19 +131,43 @@ class PyMuPDFProcessor(DocumentProcessor):
|
|||||||
pdf_image_dir = self.image_dir / pdf_id
|
pdf_image_dir = self.image_dir / pdf_id
|
||||||
pdf_image_dir.mkdir(exist_ok=True, parents=True)
|
pdf_image_dir.mkdir(exist_ok=True, parents=True)
|
||||||
|
|
||||||
# Extract all pages in a single call with page_chunks=True
|
# Extract all pages (page_chunks=True) in an isolated worker
|
||||||
def do_extract() -> list[dict[str, Any]]:
|
# subprocess with a memory rlimit + wall-clock timeout, so a
|
||||||
# When page_chunks=True, to_markdown returns list[dict] not str
|
# pathological file (e.g. a page with ~1M vector paths that drives
|
||||||
return pymupdf4llm.to_markdown( # type: ignore[return-value]
|
# table detection past the pod memory limit) fails THIS document
|
||||||
doc,
|
# 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,
|
write_images=self.extract_images,
|
||||||
image_path=pdf_image_dir if self.extract_images else None,
|
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:
|
||||||
|
logger.warning(
|
||||||
|
"Isolated PDF parse failed for %s (reason=%s): %s",
|
||||||
|
filename or "<bytes>",
|
||||||
|
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:
|
if progress_callback:
|
||||||
await progress_callback(90, 100, "Building result")
|
await progress_callback(90, 100, "Building result")
|
||||||
@@ -169,9 +202,6 @@ class PyMuPDFProcessor(DocumentProcessor):
|
|||||||
metadata["image_paths"] = image_paths
|
metadata["image_paths"] = image_paths
|
||||||
metadata["page_boundaries"] = page_boundaries
|
metadata["page_boundaries"] = page_boundaries
|
||||||
|
|
||||||
# Close document
|
|
||||||
doc.close()
|
|
||||||
|
|
||||||
if progress_callback:
|
if progress_callback:
|
||||||
await progress_callback(100, 100, "Processing complete")
|
await progress_callback(100, 100, "Processing complete")
|
||||||
|
|
||||||
|
|||||||
@@ -261,6 +261,17 @@ document_escalation_total = Counter(
|
|||||||
["from_tier", "to_tier", "reason"],
|
["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 stages ---------------------------------------------------------
|
||||||
|
|
||||||
embedding_duration_seconds = Histogram(
|
embedding_duration_seconds = Histogram(
|
||||||
@@ -617,6 +628,15 @@ def record_document_escalation(from_tier: str, to_tier: str, reason: str) -> Non
|
|||||||
).inc()
|
).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(
|
def record_embedding(
|
||||||
kind: str,
|
kind: str,
|
||||||
provider: str,
|
provider: str,
|
||||||
|
|||||||
@@ -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.models.deck import DeckCard
|
||||||
from nextcloud_mcp_server.observability.metrics import (
|
from nextcloud_mcp_server.observability.metrics import (
|
||||||
record_document_chunks,
|
record_document_chunks,
|
||||||
|
record_document_parse_failed,
|
||||||
record_embedding,
|
record_embedding,
|
||||||
record_qdrant_operation,
|
record_qdrant_operation,
|
||||||
record_vector_sync_processing,
|
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 import payload_keys
|
||||||
from nextcloud_mcp_server.vector.document_chunker import DocumentChunker
|
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.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.qdrant_client import get_qdrant_client
|
||||||
from nextcloud_mcp_server.vector.scanner import DocumentTask
|
from nextcloud_mcp_server.vector.scanner import DocumentTask
|
||||||
from nextcloud_mcp_server.vector.sharing_state import (
|
from nextcloud_mcp_server.vector.sharing_state import (
|
||||||
@@ -233,7 +237,18 @@ async def process_document(
|
|||||||
|
|
||||||
for attempt in range(max_retries):
|
for attempt in range(max_retries):
|
||||||
try:
|
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.
|
||||||
|
# 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
|
||||||
|
|
||||||
# Record successful processing metrics
|
# Record successful processing metrics
|
||||||
duration = time.time() - start_time
|
duration = time.time() - start_time
|
||||||
@@ -299,10 +314,13 @@ async def process_document(
|
|||||||
|
|
||||||
async def _index_document(
|
async def _index_document(
|
||||||
doc_task: DocumentTask, nc_client: NextcloudClient, qdrant_client
|
doc_task: DocumentTask, nc_client: NextcloudClient, qdrant_client
|
||||||
):
|
) -> bool | None:
|
||||||
"""
|
"""
|
||||||
Index a single document (called by process_document with retry).
|
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:
|
Args:
|
||||||
doc_task: Document task to index
|
doc_task: Document task to index
|
||||||
nc_client: Authenticated Nextcloud client
|
nc_client: Authenticated Nextcloud client
|
||||||
@@ -525,6 +543,40 @@ async def _index_document(
|
|||||||
content_type=content_type,
|
content_type=content_type,
|
||||||
filename=file_path,
|
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 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)
|
||||||
|
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 False
|
||||||
|
|
||||||
content = result.text
|
content = result.text
|
||||||
file_metadata = result.metadata
|
file_metadata = result.metadata
|
||||||
title = file_metadata.get("title") or file_path.split("/")[-1]
|
title = file_metadata.get("title") or file_path.split("/")[-1]
|
||||||
|
|||||||
@@ -526,13 +526,24 @@ async def scan_user_documents(
|
|||||||
# File modified since last indexing
|
# File modified since last indexing
|
||||||
needs_indexing = True
|
needs_indexing = True
|
||||||
elif existing_metadata.get("is_placeholder", False):
|
elif existing_metadata.get("is_placeholder", False):
|
||||||
# Placeholder exists - check if it's stale (processing may have failed)
|
# Placeholder exists - check its status / staleness.
|
||||||
# Only requeue if placeholder is older than 5x scan interval
|
|
||||||
# (Large PDFs can take 3-4 minutes to process)
|
|
||||||
queued_at = existing_metadata.get("queued_at", 0)
|
queued_at = existing_metadata.get("queued_at", 0)
|
||||||
placeholder_age = time.time() - queued_at
|
placeholder_age = time.time() - queued_at
|
||||||
stale_threshold = get_settings().vector_sync_scan_interval * 5
|
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(
|
logger.debug(
|
||||||
"Found stale placeholder for file %s (ID: %s) (age=%ss), requeuing",
|
"Found stale placeholder for file %s (ID: %s) (age=%ss), requeuing",
|
||||||
file_path,
|
file_path,
|
||||||
|
|||||||
@@ -0,0 +1,203 @@
|
|||||||
|
"""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_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)
|
||||||
|
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"
|
||||||
Reference in New Issue
Block a user