From 8c9339501e0e483d60d2895278fbcadd3b4d9b14 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:09:49 +0200 Subject: [PATCH 1/6] fix(vector): dead-letter terminally-failed documents to stop multi-user re-queue loop A pathological PDF (a 206-page ChronoScan scan with ~3400 JBIG2/JPX images) jammed a tenant's structured ingest worker in an infinite reprocess loop, re-burning a 120s pymupdf4llm parse (and occasionally OOM-racing the 2Gi pod) every few minutes. Root cause: the per-user placeholder "failed" mark could not stop the loop. The placeholder point ID is user-agnostic (uuid5("file::placeholder")) but the scanner's freshness gate, query, and status update all filter by user_id. For a file visible to several users the single shared placeholder's user_id is overwritten by whoever scanned last, so every other user's scan sees "no record" and re-queues -- an N-user ping-pong that never honours the failed status. Fix: when a parse fails terminally (no higher escalation tier available, e.g. structured with OCR off) record a durable, content-addressed, user-agnostic dead-letter marker (mirrors vector/sharing_state.py). The scanner consults it tenant-wide for every user and skips re-queuing until the content (etag) OR the escalation-tier set (tiers_sig -- e.g. OCR enabled) changes, so the document is attempted once per content-version instead of forever. - new vector/dead_letter.py: mark/is/clear, content-addressed marker carrying is_placeholder=True (inherits search exclusion) + dead_letter=True - escalation.escalation_tiers_signature(settings): retry-on-tier-change key - processor: dead-letter terminal failures, clear on successful (re-)index - scanner: user-agnostic is_dead_lettered skip beside claim_existing_index - placeholder: exempt dead_letter markers from the orphan sweep (durability) - metrics: astrolabe_document_dead_lettered_total{reason} Deck #349. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../document_processors/escalation.py | 25 ++- nextcloud_mcp_server/observability/metrics.py | 24 +++ nextcloud_mcp_server/vector/dead_letter.py | 190 ++++++++++++++++++ nextcloud_mcp_server/vector/placeholder.py | 7 + nextcloud_mcp_server/vector/processor.py | 114 +++++++++-- nextcloud_mcp_server/vector/scanner.py | 29 +++ tests/unit/test_escalation_signature.py | 42 ++++ tests/unit/vector/test_dead_letter.py | 170 ++++++++++++++++ tests/unit/vector/test_placeholder.py | 20 ++ .../unit/vector/test_processor_dead_letter.py | 141 +++++++++++++ 10 files changed, 739 insertions(+), 23 deletions(-) create mode 100644 nextcloud_mcp_server/vector/dead_letter.py create mode 100644 tests/unit/test_escalation_signature.py create mode 100644 tests/unit/vector/test_dead_letter.py create mode 100644 tests/unit/vector/test_processor_dead_letter.py diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py index d9c3126e..67af6c0a 100644 --- a/nextcloud_mcp_server/document_processors/escalation.py +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -22,13 +22,36 @@ mapping lives in the queue layer, which imports :class:`EscalateError` from here from __future__ import annotations from dataclasses import dataclass -from typing import Literal +from typing import Any, Literal # Cheapest-first. ``llm`` is reserved (see base.DocumentProcessor.tier) and not # wired yet, so it is intentionally absent from the live ladder. TIER_LADDER: tuple[str, ...] = ("fast", "structured", "ocr") +def escalation_tiers_signature(settings: Any) -> str: + """A stable string fingerprint of the runtime escalation-tier configuration. + + Used by the document dead-letter marker (``vector/dead_letter.py``) as part of + its content key: a document that fails its terminal tier is dead-lettered until + either its content (etag) OR this signature changes. The signature therefore + captures every setting that can make a *new* escalation tier become available + at runtime — flip it and previously dead-lettered documents become retryable. + + Derived purely from settings (not the live ``ProcessorRegistry``) so it is + identical across the API/scanner and worker roles: the *registered* processor + set is build-constant, so the only runtime variables are the OCR-enabled gate + (``document_ocr_enabled`` — the single tier toggle today) and the tier-1 engine + pin (``document_tier1_engine``). Enabling OCR changes the signature, so the + pathological-but-OCR-recoverable documents dead-lettered while OCR was off are + re-attempted automatically. + """ + return ( + f"ocr={int(bool(settings.document_ocr_enabled))};" + f"t1={settings.document_tier1_engine}" + ) + + @dataclass(frozen=True) class EscalationDecision: """Outcome of the post-parse quality gate (``ProcessorRegistry.evaluate_escalation``). diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 139b6c87..e7537964 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -305,6 +305,20 @@ document_parse_failed_total = Counter( ["reason"], # reason: timeout | oom | error ) +# Documents dead-lettered after a terminal parse failure: the failing tier had +# no higher escalation tier available (e.g. structured timed out with OCR off), +# so the document is recorded as permanently failed for this content-version and +# stops being re-queued (vector/dead_letter.py). Distinct from +# ``document_parse_failed_total`` (which counts every failed parse attempt, +# including the ones that will be retried) -- this fires once when a document +# is given up on, and clears implicitly when its etag or the escalation-tier set +# changes and it is re-attempted. +document_dead_lettered_total = Counter( + "astrolabe_document_dead_lettered_total", + "Documents dead-lettered after a terminal parse failure (no escalation tier)", + ["reason"], # reason: timeout | oom | error +) + # Documents dropped after exhausting in-process indexing retries (the scanner # re-picks them on a later full scan, so this is "dropped for this cycle", not # "lost forever"). Labelled by classified cause so the embed-drop rate from a @@ -787,6 +801,16 @@ def record_document_parse_failed(reason: str) -> None: document_parse_failed_total.labels(reason=reason).inc() +def record_document_dead_lettered(reason: str) -> None: + """Record a document dead-lettered after a terminal parse failure. + + Args: + reason: ``timeout`` | ``oom`` | ``error`` (the terminal parse failure + reason carried from the isolated worker). + """ + document_dead_lettered_total.labels(reason=reason).inc() + + def record_ingest_dropped(reason: str) -> None: """Record a document dropped after exhausting in-process indexing retries. diff --git a/nextcloud_mcp_server/vector/dead_letter.py b/nextcloud_mcp_server/vector/dead_letter.py new file mode 100644 index 00000000..eca389de --- /dev/null +++ b/nextcloud_mcp_server/vector/dead_letter.py @@ -0,0 +1,190 @@ +"""Content-addressed dead-letter markers for terminally-failed documents. + +A document that fails its terminal extraction tier (a hard parse failure with no +higher tier to escalate to — e.g. the ``structured`` tier timing out while OCR is +disabled) must not be retried forever. The per-user placeholder ``status="failed"`` +mark cannot stop the loop on its own: placeholder point IDs are user-agnostic +(``uuid5("file::placeholder")``) but the scanner's freshness gate filters +by ``user_id``, so for a file visible to *several* users the single shared +placeholder's ``user_id`` is overwritten by whoever scanned last and every other +user's scan sees "no record → re-queue", re-burning the (failing) parse on a loop. + +This module records a **durable, content-addressed, user-agnostic** dead-letter +marker instead. One marker point per document (a distinct deterministic ID, kept +separate from the in-flight placeholder), carrying the ``etag`` and an escalation +``tiers_sig`` (see ``document_processors.escalation.escalation_tiers_signature``). +The scanner consults it tenant-wide — for every user — and skips re-queuing while +BOTH still match, so the document is attempted once per content-version and never +loops. A content change (new ``etag``) or a config change that adds an escalation +tier (e.g. enabling OCR — new ``tiers_sig``) makes the marker stale and the +document retryable again. + +The marker carries ``is_placeholder=True`` so the existing search exclusion +(``get_placeholder_filter``) keeps it out of user-facing results with no extra +filter, plus ``dead_letter=True`` so the orphan-placeholder sweep and the scanner +can tell it apart from a volatile in-flight placeholder. Mirrors the fail-safe +philosophy of ``sharing_state``: a Qdrant error never aborts ingest — a failed +lookup degrades to "process normally", a failed write is logged, not raised. +""" + +import logging +import time +import uuid + +from qdrant_client import models +from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct + +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.embedding import get_embedding_service +from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client + +logger = logging.getLogger(__name__) + +# Payload flag distinguishing a durable dead-letter marker from an in-flight +# placeholder (both carry is_placeholder=True to inherit the search exclusion). +DEAD_LETTER_KEY = "dead_letter" + + +def _generate_dead_letter_id(doc_type: str, doc_id: str) -> str: + """Deterministic, user-agnostic point ID for a document's dead-letter marker. + + Distinct from the in-flight placeholder ID (``…:placeholder``) so the two can + coexist briefly and never collide; one marker per ``(doc_type, doc_id)``, so + a re-failure upserts in place rather than accumulating. + """ + return str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{doc_type}:{doc_id}:deadletter")) + + +def _dead_letter_filter(doc_id: str, doc_type: str) -> Filter: + """Match the dead-letter marker for one document (tenant-wide, no user_id).""" + return Filter( + must=[ + FieldCondition(key="doc_id", match=MatchValue(value=doc_id)), + FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), + FieldCondition(key=DEAD_LETTER_KEY, match=MatchValue(value=True)), + ] + ) + + +async def mark_dead_letter( + doc_id: str, + doc_type: str, + etag: str, + tiers_sig: str, + reason: str, + *, + file_path: str | None = None, +) -> None: + """Upsert a durable dead-letter marker for a terminally-failed document. + + Keyed by content (``etag``) + escalation config (``tiers_sig``); the scanner + skips re-queuing while both match. ``reason`` is the parse failure reason + (``timeout`` | ``oom`` | ``error``). Fail-safe: a Qdrant error is logged, not + raised — a missed mark just means the document is retried (the bounded prior + behaviour), never a crash. + """ + try: + qdrant_client = await get_qdrant_client() + settings = get_settings() + dimension = get_embedding_service().get_dimension() + + payload: dict = { + "doc_id": doc_id, + "doc_type": doc_type, + "is_placeholder": True, + DEAD_LETTER_KEY: True, + "etag": etag, + "tiers_sig": tiers_sig, + "reason": reason, + "failed_at": int(time.time()), + } + if doc_type == "file" and file_path: + payload["file_path"] = file_path + + point = PointStruct( + id=_generate_dead_letter_id(doc_type, doc_id), + vector={ + "dense": [0.0] * dimension, + "sparse": models.SparseVector(indices=[], values=[]), + }, + payload=payload, + ) + await qdrant_client.upsert( + collection_name=settings.get_collection_name(), + points=[point], + wait=True, + ) + logger.info( + "Dead-lettered %s_%s (reason=%s, etag=%s)", + doc_type, + doc_id, + reason, + etag, + ) + except Exception as e: + logger.warning( + "Failed to write dead-letter marker for %s_%s: %s", doc_type, doc_id, e + ) + # Don't raise — dead-lettering is best-effort; a miss just retries. + + +async def is_dead_lettered( + doc_id: str, + doc_type: str, + etag: str, + tiers_sig: str, +) -> bool: + """Whether this exact content is currently dead-lettered (skip re-queuing). + + Returns True only when a marker exists for ``(doc_id, doc_type)`` whose stored + ``etag`` AND ``tiers_sig`` both match the current values — so a content change + or a new escalation tier (e.g. OCR enabled) makes the document retryable + again. An empty ``etag`` is never dead-lettered (we cannot content-address it). + Fail-safe: a Qdrant error degrades to False (process normally), mirroring + ``sharing_state.claim_existing_index``. + """ + if not etag: + return False + try: + qdrant_client = await get_qdrant_client() + settings = get_settings() + points, _ = await qdrant_client.scroll( + collection_name=settings.get_collection_name(), + scroll_filter=_dead_letter_filter(doc_id, doc_type), + limit=1, + with_payload=True, + with_vectors=False, + ) + except Exception as e: + logger.warning( + "Dead-letter lookup failed for %s_%s (%s); processing normally", + doc_type, + doc_id, + e, + ) + return False + if not points: + return False + payload = dict(points[0].payload or {}) + return payload.get("etag") == etag and payload.get("tiers_sig") == tiers_sig + + +async def clear_dead_letter(doc_id: str, doc_type: str) -> None: + """Delete a document's dead-letter marker (on successful index / release). + + Idempotent and fail-safe: deleting a non-existent marker is a Qdrant no-op, + and an error is logged rather than raised so it never breaks the indexing + path that calls it. + """ + try: + qdrant_client = await get_qdrant_client() + settings = get_settings() + await qdrant_client.delete( + collection_name=settings.get_collection_name(), + points_selector=_dead_letter_filter(doc_id, doc_type), + ) + logger.debug("Cleared dead-letter marker for %s_%s", doc_type, doc_id) + except Exception as e: + logger.warning( + "Failed to clear dead-letter marker for %s_%s: %s", doc_type, doc_id, e + ) diff --git a/nextcloud_mcp_server/vector/placeholder.py b/nextcloud_mcp_server/vector/placeholder.py index af569ae6..b542083b 100644 --- a/nextcloud_mcp_server/vector/placeholder.py +++ b/nextcloud_mcp_server/vector/placeholder.py @@ -397,6 +397,13 @@ async def sweep_orphan_placeholders( orphan_ids = [] for point in points: payload = point.payload or {} + # Dead-letter markers (vector/dead_letter.py) reuse is_placeholder=True + # for the search exclusion but are DURABLE terminal-state records, not + # in-flight placeholders -- they carry no/foreign instance_id and must + # survive a Pod restart, so never sweep them as orphans. + if payload.get("dead_letter") is True: + kept += 1 + continue point_instance = payload.get("instance_id") if point_instance == _INSTANCE_ID: kept += 1 diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 8d517862..39ee8389 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -28,6 +28,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_dead_lettered, record_document_escalation, record_document_escalation_suppressed, record_document_parse_failed, @@ -43,6 +44,10 @@ from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter from nextcloud_mcp_server.usage import UsageEventStore from nextcloud_mcp_server.vector import payload_keys from nextcloud_mcp_server.vector._errors import format_exception_group +from nextcloud_mcp_server.vector.dead_letter import ( + clear_dead_letter, + mark_dead_letter, +) from nextcloud_mcp_server.vector.document_chunker import ( DocumentChunker, PageAwareChunker, @@ -993,8 +998,10 @@ async def _index_document( get_registry, ) from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415 + TIER_LADDER, BatchPending, EscalateError, + escalation_tiers_signature, ) registry = get_registry() @@ -1035,34 +1042,89 @@ async def _index_document( # 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). + # fail again. 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, + # The tier that produced this failed result: the worker's own + # tier on the per-tier path, else the deepest tier the inline + # pipeline reached (recorded as ``pipeline_tier``). + failing_tier = tier or result.metadata.get( + "pipeline_tier", TIER_LADDER[0] ) - 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", + terminal = ( + registry.next_available_tier(failing_tier, settings) is None + ) + if terminal: + # No higher tier can run (e.g. structured timed out with + # OCR off), so retrying just re-burns the same failing + # parse. Dead-letter the document tenant-wide + # (content-addressed, user-agnostic) so EVERY user's scan + # stops re-queuing it until its content (etag) or the + # escalation-tier set (e.g. OCR enabled -> new tiers_sig) + # changes. This fixes the multi-user placeholder + # ping-pong the per-user "failed" mark could not: a file + # shared by N users has ONE user-agnostic placeholder + # whose user_id is overwritten by the last scanner, so + # every other user re-queued it forever. + await mark_dead_letter( doc_task.doc_id, - exc_info=True, + doc_task.doc_type, + doc_task.etag or "", + escalation_tiers_signature(settings), + reason, + file_path=file_path, ) + record_document_dead_lettered(reason) + logger.warning( + "Permanent parse failure for %s (reason=%s); " + "dead-lettered (terminal tier=%s, no escalation) and " + "skipping index", + file_path, + reason, + failing_tier, + ) + # Drop the volatile in-flight placeholder; the durable + # marker is now the document's terminal-state record. + try: + await delete_placeholder_point( + doc_id=doc_task.doc_id, + doc_type=doc_task.doc_type, + user_id=doc_task.user_id, + ) + except Exception: + logger.debug( + "Could not delete placeholder for dead-lettered %s", + doc_task.doc_id, + exc_info=True, + ) + else: + # A higher tier exists; parse failures don't escalate to + # it today, so keep the legacy per-user "failed" + # placeholder mark (not dead-lettered -- a future change + # may route the failure to that tier). + 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. + logger.debug( + "Could not mark placeholder failed for %s", + doc_task.doc_id, + exc_info=True, + ) return False content = result.text @@ -1529,6 +1591,14 @@ async def _index_document( ) ) + # A successful (re-)index supersedes any prior terminal failure: clear a + # stale dead-letter marker (e.g. the file was fixed/replaced, or a new + # escalation tier finally parsed it) so it isn't left behind. Only files are + # ever dead-lettered (the mark lives in the file branch), so skip the extra + # Qdrant round-trip for the other doc types on the hot indexing path. + if doc_task.doc_type == "file": + await clear_dead_letter(doc_task.doc_id, doc_task.doc_type) + # Delete placeholder before writing real vectors # This prevents duplicates and cleans up the placeholder state try: diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index b00d4fe1..fdf381e7 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -27,6 +27,7 @@ from nextcloud_mcp_server.server.tag_exclusion import ( get_excluded_file_paths, is_path_excluded, ) +from nextcloud_mcp_server.vector.dead_letter import is_dead_lettered from nextcloud_mcp_server.vector.placeholder import ( query_document_metadata, write_placeholder_point, @@ -664,6 +665,16 @@ async def scan_user_documents( skipped, ) + # Escalation-tier fingerprint for the dead-letter skip below, computed + # once per scan. Lazy import: document_processors.__init__ pulls the + # heavy parse stack (pymupdf/_isolation, Unix-only ``resource``; #877), + # which the scanner (API role) must not load at module import. + from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415 + escalation_tiers_signature, + ) + + tiers_sig = escalation_tiers_signature(get_settings()) + for file_info in tagged_files: # Files are already filtered by MIME type in find_files_by_tag() file_count += 1 @@ -703,6 +714,24 @@ async def scan_user_documents( ) continue + # Tenant-wide dead-letter skip: a document that terminally failed + # parsing (no escalation tier) is recorded user-agnostically, so + # EVERY user's scan skips re-queuing it until its content (etag) or + # the escalation-tier set (tiers_sig, e.g. OCR enabled) changes. + # Unlike the per-user placeholder "failed" mark this is not + # defeated by a file shared across users -- whose single + # user-agnostic placeholder's user_id is overwritten by the last + # scanner, so every other user re-queued it on a loop. + if etag and await is_dead_lettered(file_id, "file", etag, tiers_sig): + _potentially_deleted.pop((user_id, file_id), None) + logger.debug( + "Skipping dead-lettered file %s (ID: %s) until content/" + "tier change", + file_path, + file_id, + ) + continue + if initial_sync: # Send everything on first sync - write placeholder first await write_placeholder_point( diff --git a/tests/unit/test_escalation_signature.py b/tests/unit/test_escalation_signature.py new file mode 100644 index 00000000..a9bc8922 --- /dev/null +++ b/tests/unit/test_escalation_signature.py @@ -0,0 +1,42 @@ +"""Unit tests for the escalation-tier signature used by dead-letter keying. + +``escalation_tiers_signature`` fingerprints the runtime escalation config so a +dead-lettered document becomes retryable when a new tier appears (e.g. an +operator enables OCR). It must be settings-derived (role-independent) and must +change when OCR is toggled. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from nextcloud_mcp_server.document_processors.escalation import ( + escalation_tiers_signature, +) + +pytestmark = pytest.mark.unit + + +def _settings(*, ocr: bool, engine: str = "pypdfium2") -> SimpleNamespace: + return SimpleNamespace(document_ocr_enabled=ocr, document_tier1_engine=engine) + + +def test_signature_is_stable_for_same_config() -> None: + assert escalation_tiers_signature( + _settings(ocr=False) + ) == escalation_tiers_signature(_settings(ocr=False)) + + +def test_enabling_ocr_changes_signature() -> None: + # Enabling OCR adds an escalation tier -> previously dead-lettered docs retry. + assert escalation_tiers_signature( + _settings(ocr=False) + ) != escalation_tiers_signature(_settings(ocr=True)) + + +def test_tier1_engine_change_changes_signature() -> None: + assert escalation_tiers_signature( + _settings(ocr=False, engine="pypdfium2") + ) != escalation_tiers_signature(_settings(ocr=False, engine="pymupdf")) diff --git a/tests/unit/vector/test_dead_letter.py b/tests/unit/vector/test_dead_letter.py new file mode 100644 index 00000000..6defa720 --- /dev/null +++ b/tests/unit/vector/test_dead_letter.py @@ -0,0 +1,170 @@ +"""Unit tests for content-addressed dead-letter markers. + +Covers vector/dead_letter.py: the durable, user-agnostic terminal-failure marker +that stops a multi-user shared file (whose single placeholder's user_id is +overwritten by the last scanner) from being re-queued forever. A marker is keyed +by content (``etag``) + escalation config (``tiers_sig``); a scan skips only while +both still match, so a content change or a newly-available tier (e.g. OCR enabled) +makes the document retryable again. + +Qdrant is reached via ``get_qdrant_client``/``get_settings``/``get_embedding_service``, +all monkeypatched here so the logic runs without a live Qdrant. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from nextcloud_mcp_server.vector import dead_letter as dl + +pytestmark = pytest.mark.unit + +_COLLECTION = "test_collection" + + +class _Settings: + def get_collection_name(self) -> str: + return _COLLECTION + + +def _point(payload: dict) -> SimpleNamespace: + """Stand-in for a qdrant_client Record (only id/payload are read).""" + return SimpleNamespace(id="pt", payload=payload) + + +@pytest.fixture +def client(monkeypatch) -> AsyncMock: + """An AsyncMock Qdrant client wired into dead_letter, plus stub deps. + + ``scroll`` defaults to "no points"; individual tests override + ``client.scroll.return_value``/``side_effect``. + """ + qc = AsyncMock() + qc.scroll.return_value = ([], None) + monkeypatch.setattr(dl, "get_qdrant_client", AsyncMock(return_value=qc)) + monkeypatch.setattr(dl, "get_settings", lambda: _Settings()) + monkeypatch.setattr( + dl, "get_embedding_service", lambda: SimpleNamespace(get_dimension=lambda: 4) + ) + return qc + + +def _must_conditions(flt) -> dict: + """Map FieldCondition key -> matched value for a Filter's ``must`` clause.""" + out = {} + for c in flt.must or []: + key = getattr(c, "key", None) + match = getattr(c, "match", None) + out[key] = getattr(match, "value", None) + return out + + +class TestDeadLetterId: + def test_user_agnostic_and_distinct_from_placeholder(self) -> None: + from nextcloud_mcp_server.vector.placeholder import _generate_placeholder_id + + dl_id = dl._generate_dead_letter_id("file", "520189") + # Deterministic + user-agnostic (depends only on doc_type:doc_id). + assert dl_id == dl._generate_dead_letter_id("file", "520189") + # Never collides with the in-flight placeholder for the same document. + assert dl_id != _generate_placeholder_id("file", "520189") + + +class TestMarkDeadLetter: + async def test_upserts_content_addressed_marker(self, client) -> None: + await dl.mark_dead_letter( + "520189", + "file", + "etag-1", + "ocr=0;t1=pypdfium2", + "timeout", + file_path="/Plans/big.pdf", + ) + client.upsert.assert_awaited_once() + point = client.upsert.await_args.kwargs["points"][0] + assert point.id == dl._generate_dead_letter_id("file", "520189") + payload = point.payload + assert payload["is_placeholder"] is True + assert payload[dl.DEAD_LETTER_KEY] is True + assert payload["etag"] == "etag-1" + assert payload["tiers_sig"] == "ocr=0;t1=pypdfium2" + assert payload["reason"] == "timeout" + assert payload["doc_id"] == "520189" + assert payload["file_path"] == "/Plans/big.pdf" + + async def test_failure_is_swallowed(self, client) -> None: + client.upsert.side_effect = RuntimeError("qdrant down") + # Best-effort: a write failure must not propagate (would crash the worker). + await dl.mark_dead_letter("1", "file", "e", "sig", "oom") + + +class TestIsDeadLettered: + def _marker(self, etag: str, tiers_sig: str) -> SimpleNamespace: + return _point( + { + "doc_id": "520189", + dl.DEAD_LETTER_KEY: True, + "etag": etag, + "tiers_sig": tiers_sig, + } + ) + + async def test_true_when_etag_and_sig_match(self, client) -> None: + client.scroll.return_value = ([self._marker("e1", "sig1")], None) + assert await dl.is_dead_lettered("520189", "file", "e1", "sig1") is True + + async def test_false_on_etag_change(self, client) -> None: + client.scroll.return_value = ([self._marker("e1", "sig1")], None) + # File content changed -> retryable. + assert await dl.is_dead_lettered("520189", "file", "e2", "sig1") is False + + async def test_false_on_tiers_sig_change(self, client) -> None: + client.scroll.return_value = ([self._marker("e1", "ocr=0;t1=pypdfium2")], None) + # OCR just enabled -> a new escalation tier exists -> retryable. + assert ( + await dl.is_dead_lettered("520189", "file", "e1", "ocr=1;t1=pypdfium2") + is False + ) + + async def test_false_when_no_marker(self, client) -> None: + client.scroll.return_value = ([], None) + assert await dl.is_dead_lettered("520189", "file", "e1", "sig1") is False + + async def test_false_on_empty_etag(self, client) -> None: + # Cannot content-address without an etag; never dead-lettered. + assert await dl.is_dead_lettered("520189", "file", "", "sig1") is False + client.scroll.assert_not_awaited() + + async def test_false_on_qdrant_error(self, client) -> None: + client.scroll.side_effect = RuntimeError("qdrant down") + # Degrade to "process normally" rather than aborting the scan. + assert await dl.is_dead_lettered("520189", "file", "e1", "sig1") is False + + async def test_filter_is_user_agnostic(self, client) -> None: + client.scroll.return_value = ([self._marker("e1", "sig1")], None) + await dl.is_dead_lettered("520189", "file", "e1", "sig1") + flt = client.scroll.await_args.kwargs["scroll_filter"] + conds = _must_conditions(flt) + assert conds == { + "doc_id": "520189", + "doc_type": "file", + dl.DEAD_LETTER_KEY: True, + } + assert "user_id" not in conds + + +class TestClearDeadLetter: + async def test_deletes_by_marker_filter(self, client) -> None: + await dl.clear_dead_letter("520189", "file") + client.delete.assert_awaited_once() + flt = client.delete.await_args.kwargs["points_selector"] + conds = _must_conditions(flt) + assert conds[dl.DEAD_LETTER_KEY] is True + assert conds["doc_id"] == "520189" + + async def test_failure_is_swallowed(self, client) -> None: + client.delete.side_effect = RuntimeError("qdrant down") + await dl.clear_dead_letter("520189", "file") diff --git a/tests/unit/vector/test_placeholder.py b/tests/unit/vector/test_placeholder.py index e7212b2e..c8017899 100644 --- a/tests/unit/vector/test_placeholder.py +++ b/tests/unit/vector/test_placeholder.py @@ -110,6 +110,26 @@ async def test_keeps_own_pod_placeholders(monkeypatch): client.delete.assert_not_awaited() +@pytest.mark.unit +async def test_keeps_dead_letter_markers(monkeypatch): + """Dead-letter markers (vector/dead_letter.py) reuse is_placeholder=True for + the search exclusion but are DURABLE terminal-state records, not in-flight + placeholders. They carry a foreign/absent instance_id, so without the + carve-out the sweep would delete them on every Pod restart and the + dead-lettered document would loop again. They MUST be kept.""" + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-new") + marker = SimpleNamespace( + id="dl-1", + payload={"is_placeholder": True, "dead_letter": True, "instance_id": "pod-old"}, + ) + client = _make_client([([marker], None)]) + + swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content") + + assert (swept, kept) == (0, 1) + client.delete.assert_not_awaited() + + @pytest.mark.unit async def test_noop_when_no_placeholders_exist(monkeypatch): """Cold-boot tenant with an empty collection — sweep does NOT diff --git a/tests/unit/vector/test_processor_dead_letter.py b/tests/unit/vector/test_processor_dead_letter.py new file mode 100644 index 00000000..a00dbcbc --- /dev/null +++ b/tests/unit/vector/test_processor_dead_letter.py @@ -0,0 +1,141 @@ +"""Unit tests for the processor's terminal-parse-failure dead-lettering. + +When a PDF parse fails permanently (isolated-worker timeout/OOM) and the failing +tier has NO higher escalation tier available (e.g. ``structured`` with OCR off), +``_index_document`` records a durable, content-addressed dead-letter marker +instead of the per-user ``status="failed"`` placeholder mark — the latter could +not stop the multi-user re-queue loop. A failure that still has a higher tier +keeps the legacy per-user mark. + +The real ``ProcessorRegistry`` singleton is used so the terminal decision +(``next_available_tier``) is exercised faithfully; only the parse itself, the +content fetch, and the Qdrant side-effects are mocked. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nextcloud_mcp_server.document_processors.base import ProcessingResult +from nextcloud_mcp_server.vector import processor +from nextcloud_mcp_server.vector.scanner import DocumentTask + +pytestmark = pytest.mark.unit + + +def _settings(*, ocr_enabled: bool) -> SimpleNamespace: + return SimpleNamespace( + document_ocr_enabled=ocr_enabled, + document_tier1_engine="pypdfium2", + get_collection_name=lambda: "c", + ) + + +def _file_task() -> DocumentTask: + return DocumentTask( + user_id="Demo-User", + doc_id="520189", + doc_type="file", + operation="index", + modified_at=0, + file_path="/Plans/big.pdf", + etag="etag-1", + ) + + +def _nc_client() -> MagicMock: + # MagicMock (typed Any) keeps the pre-commit ty-check happy where the real + # signature wants a NextcloudClient -- matching the other processor tests. + nc = MagicMock() + nc.webdav.read_file = AsyncMock(return_value=(b"%PDF-1.4", "application/pdf")) + return nc + + +def _patch_common(mocker, *, ocr_enabled: bool): + """Patch the shared seams; returns the spies for assertions.""" + mocker.patch.object( + processor, "get_settings", lambda: _settings(ocr_enabled=ocr_enabled) + ) + # Never a tenant-wide dedup hit (file was never indexed). + mocker.patch.object( + processor, "claim_existing_index", AsyncMock(return_value=False) + ) + spies = SimpleNamespace( + mark=mocker.patch.object(processor, "mark_dead_letter", AsyncMock()), + dead_metric=mocker.patch.object(processor, "record_document_dead_lettered"), + delete_ph=mocker.patch.object( + processor, "delete_placeholder_point", AsyncMock() + ), + update_ph=mocker.patch.object( + processor, "update_placeholder_status", AsyncMock() + ), + ) + return spies + + +async def test_terminal_failure_dead_letters(mocker): + """structured tier fails + OCR off (no higher tier) -> dead-letter, not mark.""" + spies = _patch_common(mocker, ocr_enabled=False) + # The per-tier worker runs the structured tier and the parse times out. + mocker.patch.object( + processor, + "_parse_pdf_tier", + AsyncMock( + return_value=ProcessingResult( + text="", + metadata={ + "parse_failed_reason": "timeout", + "pipeline_tier": "structured", + }, + processor="pymupdf", + success=False, + error="isolated parse failed (timeout)", + ) + ), + ) + + result = await processor._index_document( + _file_task(), _nc_client(), MagicMock(), tier="structured" + ) + + assert result is False + spies.mark.assert_awaited_once() + # Marker is content-addressed with this etag + the OCR-off tiers signature. + args = spies.mark.await_args.args + assert args[0] == "520189" and args[1] == "file" + assert args[2] == "etag-1" # etag + assert "ocr=0" in args[3] # tiers_sig + assert args[4] == "timeout" # reason + spies.dead_metric.assert_called_once_with("timeout") + spies.delete_ph.assert_awaited_once() # volatile placeholder dropped + spies.update_ph.assert_not_awaited() # NOT the legacy per-user failed mark + + +async def test_non_terminal_failure_keeps_legacy_mark(mocker): + """fast tier fails while structured is still available -> legacy failed mark.""" + spies = _patch_common(mocker, ocr_enabled=False) + mocker.patch.object( + processor, + "_parse_pdf_tier", + AsyncMock( + return_value=ProcessingResult( + text="", + metadata={"parse_failed_reason": "error", "pipeline_tier": "fast"}, + processor="pypdfium2", + success=False, + error="isolated parse failed (error)", + ) + ), + ) + + result = await processor._index_document( + _file_task(), _nc_client(), MagicMock(), tier="fast" + ) + + assert result is False + spies.update_ph.assert_awaited_once() # legacy per-user failed mark + spies.mark.assert_not_awaited() # NOT dead-lettered (structured can still run) + spies.dead_metric.assert_not_called() From cd348b32334b4c6fc3c0c5d4dc31fa56705fac96 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:17:38 +0200 Subject: [PATCH 2/6] fix(vector): clear dead-letter marker on delete, treat oversize as terminal Addresses round-1 review on PR #920: - Delete path now clears the file's dead-letter marker after release_document_for_user (whose principal-based filter misses the user-agnostic, principal-less marker), preventing orphan-marker accumulation for dead-lettered-then-deleted files. - Oversize PDFs (rejected by the pre-parse size guard, no pipeline_tier stamped) are now treated as terminal regardless of failing_tier -- no tier can parse an oversize file -- so they dead-letter instead of falling to the legacy per-user mark on the inline path. - Gate the success-path clear on a non-empty etag (an etag-less file can never have a marker, mirroring is_dead_lettered's early return). - dead_letter.py: payload typed dict[str, Any] (CLAUDE.md). Tests: oversize-terminal dead-letter and delete-path marker clear. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/dead_letter.py | 3 +- nextcloud_mcp_server/vector/processor.py | 21 ++++++-- .../unit/vector/test_processor_dead_letter.py | 51 +++++++++++++++++++ 3 files changed, 70 insertions(+), 5 deletions(-) diff --git a/nextcloud_mcp_server/vector/dead_letter.py b/nextcloud_mcp_server/vector/dead_letter.py index eca389de..245c08d4 100644 --- a/nextcloud_mcp_server/vector/dead_letter.py +++ b/nextcloud_mcp_server/vector/dead_letter.py @@ -30,6 +30,7 @@ lookup degrades to "process normally", a failed write is logged, not raised. import logging import time import uuid +from typing import Any from qdrant_client import models from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct @@ -88,7 +89,7 @@ async def mark_dead_letter( settings = get_settings() dimension = get_embedding_service().get_dimension() - payload: dict = { + payload: dict[str, Any] = { "doc_id": doc_id, "doc_type": doc_type, "is_placeholder": True, diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 39ee8389..e394da6f 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -626,6 +626,13 @@ async def process_document( await release_document_for_user( doc_task.doc_id, doc_task.doc_type, doc_task.user_id ) + # Drop any dead-letter marker for the file too: release only + # removes it when the last reader leaves (its filter misses the + # user-agnostic, principal-less marker), so without this a + # dead-lettered-then-deleted file would leave an orphan marker + # accumulating in Qdrant. Only files are ever dead-lettered. + if doc_task.doc_type == "file": + await clear_dead_letter(doc_task.doc_id, doc_task.doc_type) logger.info( "Deleted %s_%s for %s", doc_task.doc_type, @@ -1052,8 +1059,14 @@ async def _index_document( failing_tier = tier or result.metadata.get( "pipeline_tier", TIER_LADDER[0] ) + # An oversize PDF is rejected by the pre-parse size guard + # before any tier runs (no pipeline_tier stamped on the inline + # path) and no tier can ever parse it, so it is terminal + # regardless of failing_tier. Otherwise, terminal == no higher + # tier can run. terminal = ( - registry.next_available_tier(failing_tier, settings) is None + reason == "oversize" + or registry.next_available_tier(failing_tier, settings) is None ) if terminal: # No higher tier can run (e.g. structured timed out with @@ -1594,9 +1607,9 @@ async def _index_document( # A successful (re-)index supersedes any prior terminal failure: clear a # stale dead-letter marker (e.g. the file was fixed/replaced, or a new # escalation tier finally parsed it) so it isn't left behind. Only files are - # ever dead-lettered (the mark lives in the file branch), so skip the extra - # Qdrant round-trip for the other doc types on the hot indexing path. - if doc_task.doc_type == "file": + # ever dead-lettered, and only with a non-empty etag (is_dead_lettered + # early-returns without one), so skip the extra Qdrant round-trip otherwise. + if doc_task.doc_type == "file" and doc_task.etag: await clear_dead_letter(doc_task.doc_id, doc_task.doc_type) # Delete placeholder before writing real vectors diff --git a/tests/unit/vector/test_processor_dead_letter.py b/tests/unit/vector/test_processor_dead_letter.py index a00dbcbc..d1e9b81b 100644 --- a/tests/unit/vector/test_processor_dead_letter.py +++ b/tests/unit/vector/test_processor_dead_letter.py @@ -139,3 +139,54 @@ async def test_non_terminal_failure_keeps_legacy_mark(mocker): spies.update_ph.assert_awaited_once() # legacy per-user failed mark spies.mark.assert_not_awaited() # NOT dead-lettered (structured can still run) spies.dead_metric.assert_not_called() + + +async def test_oversize_failure_dead_letters_regardless_of_tier(mocker): + """An oversize PDF is terminal at any tier (no tier can parse it) -> dead-letter + even though a higher tier (structured) is nominally available above 'fast'.""" + spies = _patch_common(mocker, ocr_enabled=False) + mocker.patch.object( + processor, + "_parse_pdf_tier", + AsyncMock( + return_value=ProcessingResult( + text="", + metadata={"parse_failed_reason": "oversize"}, + processor="size_guard", + success=False, + error="PDF exceeds size cap", + ) + ), + ) + + result = await processor._index_document( + _file_task(), _nc_client(), MagicMock(), tier="fast" + ) + + assert result is False + spies.mark.assert_awaited_once() + assert spies.mark.await_args.args[4] == "oversize" # reason + spies.dead_metric.assert_called_once_with("oversize") + spies.update_ph.assert_not_awaited() + + +async def test_delete_clears_dead_letter_marker(mocker): + """Deleting a file must also drop its dead-letter marker, else a + dead-lettered-then-deleted file leaves an orphan accumulating in Qdrant + (release_document_for_user's filter misses the user-agnostic marker).""" + mocker.patch.object(processor, "get_qdrant_client", AsyncMock()) + mocker.patch.object(processor, "release_document_for_user", AsyncMock()) + clear = mocker.patch.object(processor, "clear_dead_letter", AsyncMock()) + + task = DocumentTask( + user_id="Demo-User", + doc_id="520189", + doc_type="file", + operation="delete", + modified_at=0, + file_path="/Plans/big.pdf", + etag="etag-1", + ) + await processor.process_document(task, MagicMock(), max_retries=1) + + clear.assert_awaited_once_with("520189", "file") From d72007194269576c4ceeaaa89c357029aff64c0f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:25:04 +0200 Subject: [PATCH 3/6] fix(vector): guard dead-letter on etag, harden marker filter Addresses round-2 review on PR #920: - Only dead-letter a terminal failure when the file has an etag to content-address the marker; without one, fall back to the legacy per-user placeholder mark (an etagless marker is unmatchable). + test. - _dead_letter_filter now also matches is_placeholder=True (redundant with dead_letter=True but lets Qdrant use the is_placeholder payload index). - TODO(deck-349) documenting the dead-lettered-then-deleted orphan-marker leak (out of scope; needs a marker sweep or TTL field) per reviewer. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/dead_letter.py | 18 ++++++++++- nextcloud_mcp_server/vector/processor.py | 16 ++++++---- tests/unit/vector/test_dead_letter.py | 1 + .../unit/vector/test_processor_dead_letter.py | 32 +++++++++++++++++++ 4 files changed, 59 insertions(+), 8 deletions(-) diff --git a/nextcloud_mcp_server/vector/dead_letter.py b/nextcloud_mcp_server/vector/dead_letter.py index 245c08d4..7bbcbdbc 100644 --- a/nextcloud_mcp_server/vector/dead_letter.py +++ b/nextcloud_mcp_server/vector/dead_letter.py @@ -25,6 +25,16 @@ filter, plus ``dead_letter=True`` so the orphan-placeholder sweep and the scanne can tell it apart from a volatile in-flight placeholder. Mirrors the fail-safe philosophy of ``sharing_state``: a Qdrant error never aborts ingest — a failed lookup degrades to "process normally", a failed write is logged, not raised. + +TODO(deck-349): a marker for a file that is dead-lettered and *then* deleted from +Nextcloud can be orphaned. The processor's delete path clears it, but the +scanner's grace-period deletion tracking only sees a file via its real indexed +points (filtered by ``user_id``); a dead-lettered file has only this +user-agnostic marker, so its disappearance enqueues no delete task and the marker +is never reached. Not a correctness issue (search excludes ``is_placeholder=True`` +and the etag check means a stale marker never blocks new content), but it +accumulates. A dedicated marker sweep or a TTL payload field would close it — +tracked as a follow-up, not done here. """ import logging @@ -57,11 +67,17 @@ def _generate_dead_letter_id(doc_type: str, doc_id: str) -> str: def _dead_letter_filter(doc_id: str, doc_type: str) -> Filter: - """Match the dead-letter marker for one document (tenant-wide, no user_id).""" + """Match the dead-letter marker for one document (tenant-wide, no user_id). + + Includes ``is_placeholder=True`` (redundant with ``dead_letter=True``, which + nothing else sets) so Qdrant can lean on the existing ``is_placeholder`` + payload index. + """ return Filter( must=[ FieldCondition(key="doc_id", match=MatchValue(value=doc_id)), FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), + FieldCondition(key="is_placeholder", match=MatchValue(value=True)), FieldCondition(key=DEAD_LETTER_KEY, match=MatchValue(value=True)), ] ) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index e394da6f..c6f90526 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -1068,7 +1068,7 @@ async def _index_document( reason == "oversize" or registry.next_available_tier(failing_tier, settings) is None ) - if terminal: + if terminal and doc_task.etag: # No higher tier can run (e.g. structured timed out with # OCR off), so retrying just re-burns the same failing # parse. Dead-letter the document tenant-wide @@ -1079,11 +1079,13 @@ async def _index_document( # ping-pong the per-user "failed" mark could not: a file # shared by N users has ONE user-agnostic placeholder # whose user_id is overwritten by the last scanner, so - # every other user re-queued it forever. + # every other user re-queued it forever. Requires an etag + # to content-address the marker; without one (rare) we + # fall back to the legacy per-user mark below. await mark_dead_letter( doc_task.doc_id, doc_task.doc_type, - doc_task.etag or "", + doc_task.etag, escalation_tiers_signature(settings), reason, file_path=file_path, @@ -1112,10 +1114,10 @@ async def _index_document( exc_info=True, ) else: - # A higher tier exists; parse failures don't escalate to - # it today, so keep the legacy per-user "failed" - # placeholder mark (not dead-lettered -- a future change - # may route the failure to that tier). + # Either a higher tier exists (parse failures don't + # escalate to it today) or there's no etag to + # content-address a dead-letter marker. Keep the legacy + # per-user "failed" placeholder mark. logger.warning( "Permanent parse failure for %s (reason=%s); marking " "failed and skipping index", diff --git a/tests/unit/vector/test_dead_letter.py b/tests/unit/vector/test_dead_letter.py index 6defa720..6529bba0 100644 --- a/tests/unit/vector/test_dead_letter.py +++ b/tests/unit/vector/test_dead_letter.py @@ -151,6 +151,7 @@ class TestIsDeadLettered: assert conds == { "doc_id": "520189", "doc_type": "file", + "is_placeholder": True, dl.DEAD_LETTER_KEY: True, } assert "user_id" not in conds diff --git a/tests/unit/vector/test_processor_dead_letter.py b/tests/unit/vector/test_processor_dead_letter.py index d1e9b81b..afea278d 100644 --- a/tests/unit/vector/test_processor_dead_letter.py +++ b/tests/unit/vector/test_processor_dead_letter.py @@ -170,6 +170,38 @@ async def test_oversize_failure_dead_letters_regardless_of_tier(mocker): spies.update_ph.assert_not_awaited() +async def test_terminal_failure_without_etag_uses_legacy_mark(mocker): + """A terminal failure with no etag can't be content-addressed, so fall back to + the legacy per-user placeholder mark instead of writing an unmatchable marker.""" + spies = _patch_common(mocker, ocr_enabled=False) + mocker.patch.object( + processor, + "_parse_pdf_tier", + AsyncMock( + return_value=ProcessingResult( + text="", + metadata={ + "parse_failed_reason": "timeout", + "pipeline_tier": "structured", + }, + processor="pymupdf", + success=False, + error="isolated parse failed (timeout)", + ) + ), + ) + task = _file_task() + task.etag = None # no content key + + result = await processor._index_document( + task, _nc_client(), MagicMock(), tier="structured" + ) + + assert result is False + spies.mark.assert_not_awaited() # no unmatchable marker written + spies.update_ph.assert_awaited_once() # legacy fallback + + async def test_delete_clears_dead_letter_marker(mocker): """Deleting a file must also drop its dead-letter marker, else a dead-lettered-then-deleted file leaves an orphan accumulating in Qdrant From a7f74617163ed3e75813fda15496371e36271e79 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:31:58 +0200 Subject: [PATCH 4/6] docs(vector): note tiers_sig extensibility, warn on dead-letter placeholder cleanup failure Round-3 review nits on PR #920 (none blocking): - escalation_tiers_signature: TODO noting future settings that can rescue a previously-terminal document (a toggleable llm tier, a raised oversize cap) should be folded into the signature so raising them auto-retries dead-letters. - Terminal-path placeholder cleanup: a delete failure here is real Qdrant I/O, not control-flow -- log at warning (was debug) for observability. Non-fatal (the durable marker is already written). Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/document_processors/escalation.py | 7 +++++++ nextcloud_mcp_server/vector/processor.py | 6 +++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py index 67af6c0a..75ca4e72 100644 --- a/nextcloud_mcp_server/document_processors/escalation.py +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -45,6 +45,13 @@ def escalation_tiers_signature(settings: Any) -> str: pin (``document_tier1_engine``). Enabling OCR changes the signature, so the pathological-but-OCR-recoverable documents dead-lettered while OCR was off are re-attempted automatically. + + TODO: when a future setting can make a previously-terminal document parseable, + fold it in here so raising it auto-retries existing dead-letters. Two known + candidates: a new escalation tier becoming toggleable (e.g. the reserved + ``llm`` rung in ``TIER_LADDER``), and a raised oversize cap (an oversize PDF is + always-terminal, so without the cap in this signature it stays dead-lettered + until its etag changes even after an operator allows bigger files). """ return ( f"ocr={int(bool(settings.document_ocr_enabled))};" diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index c6f90526..4c3f4076 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -1108,7 +1108,11 @@ async def _index_document( user_id=doc_task.user_id, ) except Exception: - logger.debug( + # A real Qdrant I/O failure (not control-flow): warn so + # it's observable. Non-fatal -- the durable dead-letter + # marker is already written, so the leftover volatile + # placeholder is merely redundant. + logger.warning( "Could not delete placeholder for dead-lettered %s", doc_task.doc_id, exc_info=True, From 114af7bf126e27122008db188d1019ee68285c92 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:38:12 +0200 Subject: [PATCH 5/6] docs(vector): document oversize dead-letter reason and failure-mode comments Round-4 review nits on PR #920 (none blocking): - record_document_dead_lettered: enumerate the oversize reason (added this PR) alongside timeout/oom/error in the docstring + counter comment. - Note the clear-dead-letter-before-upsert ordering implication (a transient upsert failure re-parses once, never a silent drop). - Clarify the orphan sweep's kept counter for tenant-wide dead-letter markers. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/observability/metrics.py | 5 +++-- nextcloud_mcp_server/vector/placeholder.py | 2 ++ nextcloud_mcp_server/vector/processor.py | 3 +++ 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index e7537964..4644ff74 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -316,7 +316,7 @@ document_parse_failed_total = Counter( document_dead_lettered_total = Counter( "astrolabe_document_dead_lettered_total", "Documents dead-lettered after a terminal parse failure (no escalation tier)", - ["reason"], # reason: timeout | oom | error + ["reason"], # reason: timeout | oom | error | oversize ) # Documents dropped after exhausting in-process indexing retries (the scanner @@ -806,7 +806,8 @@ def record_document_dead_lettered(reason: str) -> None: Args: reason: ``timeout`` | ``oom`` | ``error`` (the terminal parse failure - reason carried from the isolated worker). + reason carried from the isolated worker) or ``oversize`` (rejected by + the pre-parse size guard, which no tier can ever parse). """ document_dead_lettered_total.labels(reason=reason).inc() diff --git a/nextcloud_mcp_server/vector/placeholder.py b/nextcloud_mcp_server/vector/placeholder.py index b542083b..bb0e5b7e 100644 --- a/nextcloud_mcp_server/vector/placeholder.py +++ b/nextcloud_mcp_server/vector/placeholder.py @@ -402,6 +402,8 @@ async def sweep_orphan_placeholders( # in-flight placeholders -- they carry no/foreign instance_id and must # survive a Pod restart, so never sweep them as orphans. if payload.get("dead_letter") is True: + # Tenant-wide and always kept (not Pod-scoped); counted under + # ``kept`` only because the sweep's tally has no separate bucket. kept += 1 continue point_instance = payload.get("instance_id") diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 4c3f4076..4b7ac807 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -1615,6 +1615,9 @@ async def _index_document( # escalation tier finally parsed it) so it isn't left behind. Only files are # ever dead-lettered, and only with a non-empty etag (is_dead_lettered # early-returns without one), so skip the extra Qdrant round-trip otherwise. + # Cleared before the real-chunk upsert below: if that upsert then fails + # transiently, the document is re-queued and re-parses once on the next scan + # (an extra parse, never a silent drop) -- the safe ordering. if doc_task.doc_type == "file" and doc_task.etag: await clear_dead_letter(doc_task.doc_id, doc_task.doc_type) From 9b937549983e174a09daa83aaa2fb63f0abfbbf3 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:43:11 +0200 Subject: [PATCH 6/6] docs(metrics): clarify dead-letter counter counts attempts (fail-safe write) Round-5 review nit on PR #920 (non-blocking): record_document_dead_lettered increments alongside the fail-safe mark_dead_letter, so the counter measures the dead-letter attempt and can sit marginally above the live marker count if a Qdrant write fails. Note it in the docstring. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/observability/metrics.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 4644ff74..04dd279b 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -804,6 +804,11 @@ def record_document_parse_failed(reason: str) -> None: def record_document_dead_lettered(reason: str) -> None: """Record a document dead-lettered after a terminal parse failure. + Counts the dead-letter *attempt*: it is incremented alongside the + ``mark_dead_letter`` call, which is fail-safe (a Qdrant write error is logged, + not raised), so a transient write failure can leave this counter marginally + above the live marker count in Qdrant. + Args: reason: ``timeout`` | ``oom`` | ``error`` (the terminal parse failure reason carried from the isolated worker) or ``oversize`` (rejected by