From 9676bb31062413a73f8ee7b627d064a6e2734406 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 13:22:18 +0200 Subject: [PATCH 1/6] feat(ingest): per-tier escalation via procrastinate queue-hop Split external (procrastinate) document processing into per-tier queues so a document is attempted at most once per tier and requeued to the next tier's queue on a low-quality parse, using procrastinate's native retry. - escalation.py: TIER_LADDER (fast->structured->ocr) + EscalateError signal - registry: process_tier (one tier) + evaluate_escalation post-parse gate (reuses classify_from_text) + next_available_tier; shared _classify_result and _oversize_result with the inline pipeline - processor: process_document(tier=...) runs one tier and raises EscalateError before embed (junk text never indexed); inline memory path unchanged - queue/procrastinate: ingest-fast|structured|ocr queues; TieredEscalationStrategy (queue-hop on EscalateError, bounded same-tier transient retry); queue-aware task; producer defers to ingest-fast; per-queue counts + all-queue reclaim - cli: worker --tier {fast,structured,ocr} - billing: pages_ocr usage event + pipeline_tier metadata (paid OCR billed apart) - observability: astrolabe_ingest_queue_depth{queue,status} gauge + per-queue counts in nc_get_vector_sync_status / management status endpoint - config: INGEST_ESCALATION_ENABLED (default true), INGEST_TRANSIENT_MAX_ATTEMPTS INGEST_ESCALATION_ENABLED=false and INGEST_QUEUE=memory preserve prior behaviour. Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/management.py | 4 + nextcloud_mcp_server/cli.py | 43 ++- nextcloud_mcp_server/config.py | 18 ++ .../document_processors/escalation.py | 66 ++++ .../document_processors/registry.py | 296 ++++++++++++++---- nextcloud_mcp_server/models/semantic.py | 9 + nextcloud_mcp_server/observability/metrics.py | 32 ++ nextcloud_mcp_server/server/semantic.py | 1 + nextcloud_mcp_server/vector/ingest_status.py | 22 +- .../vector/metrics_publisher.py | 3 + nextcloud_mcp_server/vector/processor.py | 183 ++++++++++- .../vector/queue/procrastinate.py | 282 +++++++++++++++-- tests/unit/test_processor_metering.py | 50 +++ tests/unit/test_registry_tiering.py | 147 +++++++++ tests/unit/vector/test_parse_pdf_tier.py | 70 +++++ .../vector/test_procrastinate_producer.py | 62 +++- .../vector/test_tiered_escalation_strategy.py | 100 ++++++ 17 files changed, 1259 insertions(+), 129 deletions(-) create mode 100644 nextcloud_mcp_server/document_processors/escalation.py create mode 100644 tests/unit/vector/test_parse_pdf_tier.py create mode 100644 tests/unit/vector/test_tiered_escalation_strategy.py diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 0853ad2d..ee12ac2a 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/nextcloud_mcp_server/api/management.py @@ -338,6 +338,10 @@ async def get_vector_sync_status(request: Request) -> JSONResponse: if pending.job_counts is not None: # Per-status breakdown (todo/doing/failed/…) on the postgres backend. body["job_counts"] = pending.job_counts + if pending.job_counts_by_queue is not None: + # Per-tier-queue breakdown (Deck #323): where work sits across the + # ingest-fast / ingest-structured / ingest-ocr fleets. + body["job_counts_by_queue"] = pending.job_counts_by_queue return JSONResponse(body) except Exception as e: diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index 9f1d2283..a169ffaa 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -333,8 +333,18 @@ def _init_worker_observability(settings: Settings) -> None: default=None, help="Max concurrent jobs. Defaults to VECTOR_SYNC_PROCESSOR_WORKERS.", ) -def worker(concurrency: int | None): - """Run the ingest worker (Deck #183). +@click.option( + "--tier", + type=click.Choice(["fast", "structured", "ocr"]), + default=None, + help=( + "Run only this extraction tier's queue (Deck #323). Omit to drain ALL " + "tier queues in one process (single-Deployment / dev); set it to run one " + "tier per Deployment so the fleets scale independently." + ), +) +def worker(concurrency: int | None, tier: str | None): + """Run the ingest worker (Deck #183, per-tier fleets #323). \b Drains the per-tenant Postgres ingest queue (procrastinate): for each @@ -342,6 +352,13 @@ def worker(concurrency: int | None): embeds, and upserts into Qdrant. This is the scale-to-zero ``worker`` role of the api/worker split; run it as a separate Deployment from the API pod. + \b + With --tier the worker drains only that tier's queue (``ingest-``), so + a CPU-bound ``fast`` fleet, an in-cluster ``structured`` fleet, and a paid + ``ocr`` fleet scale independently. Without it, all tier queues are drained in + one process (handy for dev / a single Deployment). A low-quality parse hops + the job to the next tier's queue automatically (see TieredEscalationStrategy). + \b Requires INGEST_QUEUE=postgres (a PostgreSQL DATABASE_URL); procrastinate is Postgres-only. @@ -349,7 +366,7 @@ def worker(concurrency: int | None): \b Example: $ export DATABASE_URL=postgresql+asyncpg://mcp:mcp@db/mcp - $ nextcloud-mcp-server worker -c 4 + $ nextcloud-mcp-server worker -c 4 --tier fast """ import anyio # noqa: PLC0415 @@ -366,11 +383,21 @@ def worker(concurrency: int | None): _init_worker_observability(settings) from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 - INGEST_QUEUE_NAME, + ALL_INGEST_QUEUES, + LEGACY_INGEST_QUEUE, + TIER_QUEUES, apply_ingest_queue_schema, get_procrastinate_app, ) + # Which queues this process drains. A single tier -> just its queue; no tier + # -> every tier queue PLUS the legacy single queue, so a rolling upgrade + # never strands jobs deferred under the pre-#323 name. + if tier is not None: + queues = [TIER_QUEUES[tier]] + else: + queues = [*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE] + # This is the consumer side of the distributed (postgres) ingest backend. # Unlike the in-process anyio pool, the worker talks to procrastinate's App # directly (run_worker_async), so it does NOT go through IngestTransport — @@ -397,13 +424,15 @@ def worker(concurrency: int | None): # Structured log (not click.echo) so it lands in the JSON / OTel # pipeline like every other startup message. logger.info( - "Ingest worker started: queue=%s concurrency=%s delete_succeeded=%s", - INGEST_QUEUE_NAME, + "Ingest worker started: tier=%s queues=%s concurrency=%s " + "delete_succeeded=%s", + tier or "all", + queues, workers, settings.ingest_delete_succeeded_jobs, ) await app.run_worker_async( - queues=[INGEST_QUEUE_NAME], + queues=queues, concurrency=workers, install_signal_handlers=True, # Drop succeeded jobs (default) so the queue table stays lean and diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 3afe22ce..397ae470 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -234,6 +234,19 @@ _DEFAULTS: dict[str, Any] = { # queue-depth metric clean). Set false to retain succeeded rows for audit # (note: indexing success is also recorded in logs/metrics regardless). "ingest_delete_succeeded_jobs": True, + # Per-tier escalation on the procrastinate (postgres) ingest path (Deck + # #323). When true, a document that a tier cannot parse well is requeued onto + # the next tier's queue (fast -> structured -> ocr) via a native procrastinate + # queue-hop. When false the ``fast`` tier is terminal -- reproduces the + # pre-#323 behaviour where the cheap tier's output is indexed as-is. No effect + # on the in-process ``memory`` backend, which keeps the inline escalation. + "ingest_escalation_enabled": True, + # Global cap on SAME-tier retries for transient infra errors (doc fetch / + # embed / Qdrant blips) on the procrastinate path. Parse-quality failures + # escalate (one parse attempt per tier) and do NOT consume this budget; only + # whitelisted transient exceptions retry in place. Shared across tiers because + # a queue-hop cannot reset a per-tier counter (see TieredEscalationStrategy). + "ingest_transient_max_attempts": 5, "collection_metadata_source": "qdrant", # qdrant | api # CP base URL for COLLECTION_METADATA_SOURCE=api (e.g. http://control-plane). # Required only when the source is api. @@ -317,6 +330,7 @@ _dynaconf = Dynaconf( Validator("METRICS_PORT", gte=1, lte=65535), # Positive integers Validator("INGEST_STALLED_JOB_SECONDS", gte=1), + Validator("INGEST_TRANSIENT_MAX_ATTEMPTS", gte=1), Validator("VECTOR_SYNC_SCAN_INTERVAL", gte=1), Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1), Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1), @@ -856,6 +870,8 @@ class Settings: mcp_role: str = "all" # api | worker | all (Deck #183 two-pod model) ingest_stalled_job_seconds: int = 300 # crashed-worker reclaim threshold ingest_delete_succeeded_jobs: bool = True # drop succeeded ingest jobs + ingest_escalation_enabled: bool = True # per-tier queue-hop (Deck #323) + ingest_transient_max_attempts: int = 5 # same-tier transient-retry cap collection_metadata_source: str = "qdrant" # qdrant | api collection_metadata_api_url: str | None = None # CP URL when source=api embedding_gateway_url: str | None = None # required when provider=gateway @@ -1481,6 +1497,8 @@ def get_settings() -> Settings: "mcp_role": "MCP_ROLE", "ingest_stalled_job_seconds": "INGEST_STALLED_JOB_SECONDS", "ingest_delete_succeeded_jobs": "INGEST_DELETE_SUCCEEDED_JOBS", + "ingest_escalation_enabled": "INGEST_ESCALATION_ENABLED", + "ingest_transient_max_attempts": "INGEST_TRANSIENT_MAX_ATTEMPTS", "collection_metadata_source": "COLLECTION_METADATA_SOURCE", "collection_metadata_api_url": "COLLECTION_METADATA_API_URL", "embedding_gateway_url": "EMBEDDING_GATEWAY_URL", diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py new file mode 100644 index 00000000..c3ec30bd --- /dev/null +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -0,0 +1,66 @@ +"""Tier-escalation ladder + signal for the per-tier ingest fleet (Deck #323). + +The escalation ladder is the cheapest-first ordering of extraction tiers: + + fast -> structured -> ocr ( -> llm, reserved) + +It mirrors the ``tier`` vocabulary documented on +:meth:`DocumentProcessor.tier <.base.DocumentProcessor.tier>` and the +observability label set. On the *external* (procrastinate) ingest path each tier +runs on its own queue + worker fleet; a document that a tier cannot parse well is +**requeued onto the next tier's queue** rather than escalated inline. The +mechanism is a raised :class:`EscalateError` that the procrastinate retry +strategy turns into a native ``RetryDecision(queue=)`` queue-hop +(see ``vector/queue/procrastinate.py``). + +This module is deliberately free of any queue/transport dependency: it only +knows the *tier* vocabulary and the escalation signal. The tier -> queue-name +mapping lives in the queue layer, which imports :class:`EscalateError` from here +(document_processors never imports vector.queue, so there is no import cycle). +""" + +from __future__ import annotations + +# 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 next_tier(current: str) -> str | None: + """The next tier above ``current`` in the ladder, or ``None`` if terminal. + + Pure ordering only -- it does not consider whether the next tier is + *available* (a processor registered / OCR enabled). Callers that need + availability resolve it against the registry + settings (see + ``ProcessorRegistry.next_available_tier``); a tier with no escalation target + is terminal and its result is indexed as-is. + """ + try: + idx = TIER_LADDER.index(current) + except ValueError: + return None + nxt = idx + 1 + return TIER_LADDER[nxt] if nxt < len(TIER_LADDER) else None + + +class EscalateError(Exception): + """Raised when a tier's parse is too poor to index and a higher tier exists. + + Carries the tiers + reason so the procrastinate retry strategy can hop the + job to the next tier's queue and record + ``astrolabe_document_escalation_total{from_tier,to_tier,reason}``. It is a + control-flow signal, NOT a failure: it must propagate *before* chunk/embed so + the junk text is never indexed, and it must never be swallowed by a broad + ``except Exception`` on the indexing path. + + ``reason`` uses the existing escalation label vocabulary: + ``empty_text`` | ``low_confidence`` | ``unsupported`` | ``forced``. + """ + + def __init__(self, *, from_tier: str, to_tier: str, reason: str) -> None: + self.from_tier = from_tier + self.to_tier = to_tier + self.reason = reason + super().__init__( + f"escalate {from_tier}->{to_tier} (reason={reason})", + ) diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index b6c181a5..ff4b678e 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -14,7 +14,8 @@ from nextcloud_mcp_server.observability.metrics import ( from nextcloud_mcp_server.observability.tracing import trace_operation from .base import DocumentProcessor, ProcessingResult, ProcessorError -from .classifier import classify_from_text, image_coverage_per_page +from .classifier import DocClassification, classify_from_text, image_coverage_per_page +from .escalation import TIER_LADDER logger = logging.getLogger(__name__) @@ -202,32 +203,9 @@ class ProcessorRegistry: """ settings = get_settings() - # Pre-parse size guard: a pathologically large PDF (e.g. a 42 MB scanned - # DUDE) burns the OCR timeout for 0 chars. Fail fast with an explicit - # reason so the caller marks the placeholder "failed" instead of - # retrying. 0 disables the cap. This lives on the auto-tiered path only: - # an explicit processor_name="ocr" override (registry.process) bypasses - # _process_pdf entirely and is intentionally not size-gated (power-user - # escape hatch). Returning here also skips _run_processor, so the - # rejection is counted on astrolabe_document_parse_failed_total{oversize} - # (via vector/processor.py) but deliberately not on the parse-duration - # histogram -- there is no parse to time. - max_pdf_mb = settings.document_max_pdf_size_mb - if max_pdf_mb > 0 and len(content) > max_pdf_mb * 1024 * 1024: - size_mb = len(content) / (1024 * 1024) - logger.warning( - "PDF %s is %.1f MB (> %.1f MB cap); failing fast as oversize", - filename or "", - size_mb, - max_pdf_mb, - ) - return ProcessingResult( - text="", - metadata={"parse_failed_reason": "oversize"}, - processor="size_guard", - success=False, - error=(f"PDF exceeds size cap: {size_mb:.1f} MB > {max_pdf_mb:.1f} MB"), - ) + oversize = self._oversize_result(content, filename, settings) + if oversize is not None: + return oversize if settings.document_tier1_engine == "pymupdf": processor = self._pdf_processor_for_tier("structured") @@ -263,46 +241,11 @@ class ProcessorRegistry: # Tier-0 classification from the extraction (cheap: text-only, no PDF # re-open). Scan detection (image analysis, re-opens the PDF) runs only # when OCR + detect_scanned are enabled, so its cost is paid by - # OCR-opted-in tenants only. - classification = None - if settings.document_classify_enabled and result.success: - try: - image_coverage = None - if ( - settings.document_ocr_enabled - and settings.document_ocr_detect_scanned - ): - try: - image_coverage = image_coverage_per_page(content) - except Exception: - # Best-effort: fall back to text-only signals. WARNING - # (not DEBUG) so a systematic scan-detection failure on an - # OCR-enabled tenant is visible at LOG_LEVEL=INFO. - logger.warning( - "Scan detection failed for %s; using text-only signals", - filename or "", - exc_info=True, - ) - classification = classify_from_text( - result.text, - result.metadata.get("page_boundaries") or [], - min_text_quality=settings.document_ocr_min_text_quality, - min_page_chars=settings.document_ocr_min_page_chars, - page_fraction=settings.document_ocr_page_fraction, - image_coverage=image_coverage, - ) - record_document_classification( - classification.recommended_tier, - classification.flags, - classification.mean_text_quality, - classification.ocr_page_fraction, - ) - except Exception: - logger.warning( - "Tier-0 classification failed for %s", - filename or "", - exc_info=True, - ) + # OCR-opted-in tenants only. Shared with the external per-tier path via + # _classify_result. + classification = self._classify_result( + result, content, settings, record=True, filename=filename + ) # Escalate scanned / no-text-layer PDFs to OCR (tier-3) when enabled and # a provider is registered. The fast tier is terminal otherwise. Note: a @@ -355,6 +298,225 @@ class ProcessorRegistry: return result + def _oversize_result( + self, content: bytes, filename: str | None, settings: Any + ) -> ProcessingResult | None: + """Pre-parse size guard, shared by the inline and per-tier paths. + + A pathologically large PDF (e.g. a 42 MB scanned DUDE) burns the OCR + timeout for 0 chars. Return an explicit ``oversize`` failure so the + caller marks the placeholder "failed" instead of retrying; 0 disables the + cap. An explicit ``processor_name`` override (``registry.process``) + bypasses tiering entirely and is intentionally not size-gated (power-user + escape hatch). Skipping ``_run_processor`` means the rejection is counted + on ``astrolabe_document_parse_failed_total{oversize}`` (via + ``vector/processor.py``) but deliberately not on the parse-duration + histogram -- there is no parse to time. + """ + max_pdf_mb = settings.document_max_pdf_size_mb + if max_pdf_mb > 0 and len(content) > max_pdf_mb * 1024 * 1024: + size_mb = len(content) / (1024 * 1024) + logger.warning( + "PDF %s is %.1f MB (> %.1f MB cap); failing fast as oversize", + filename or "", + size_mb, + max_pdf_mb, + ) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "oversize"}, + processor="size_guard", + success=False, + error=(f"PDF exceeds size cap: {size_mb:.1f} MB > {max_pdf_mb:.1f} MB"), + ) + return None + + def _classify_result( + self, + result: ProcessingResult, + content: bytes, + settings: Any, + *, + record: bool, + filename: str | None = None, + ) -> DocClassification | None: + """Tier-0 classification of a parse result (text-only, cheap). + + Shared by the inline memory-backend pipeline (:meth:`_process_pdf`) and + the external per-tier path (:meth:`evaluate_escalation`). Returns + ``None`` when classification is disabled, the parse failed, or the + classifier raised -- best-effort, a classify failure must never break + indexing. ``record`` emits the classification metrics; set it only at the + FIRST classification of a document (the ``fast`` tier) so the per-doc + counters aren't multiplied across tiers. + """ + if not (settings.document_classify_enabled and result.success): + return None + try: + image_coverage = None + if settings.document_ocr_enabled and settings.document_ocr_detect_scanned: + try: + image_coverage = image_coverage_per_page(content) + except Exception: + # Best-effort: fall back to text-only signals. WARNING (not + # DEBUG) so a systematic scan-detection failure on an + # OCR-enabled tenant is visible at LOG_LEVEL=INFO. + logger.warning( + "Scan detection failed for %s; using text-only signals", + filename or "", + exc_info=True, + ) + classification = classify_from_text( + result.text, + result.metadata.get("page_boundaries") or [], + min_text_quality=settings.document_ocr_min_text_quality, + min_page_chars=settings.document_ocr_min_page_chars, + page_fraction=settings.document_ocr_page_fraction, + image_coverage=image_coverage, + ) + except Exception: + logger.warning( + "Tier-0 classification failed for %s", + filename or "", + exc_info=True, + ) + return None + if record: + record_document_classification( + classification.recommended_tier, + classification.flags, + classification.mean_text_quality, + classification.ocr_page_fraction, + ) + return classification + + def _tier_available(self, tier: str, settings: Any) -> bool: + """Whether ``tier`` can actually run a PDF parse right now. + + A tier is available when it has a registered PDF processor and is + enabled; the ``ocr`` tier additionally requires ``DOCUMENT_OCR_ENABLED`` + (so OCR stays opt-in and a misconfigured tenant never escalates to a + backend it hasn't turned on). + """ + if self._pdf_processor_for_tier(tier) is None: + return False + if tier == "ocr" and not settings.document_ocr_enabled: + return False + return True + + def next_available_tier( + self, current_tier: str, settings: Any, *, minimum: str | None = None + ) -> str | None: + """First escalation target above ``current_tier`` that can actually run. + + Walks the ladder strictly above ``current_tier`` (and not below + ``minimum``'s rung, when given) and returns the first + :meth:`_tier_available` tier. ``None`` means no higher tier can run -- + ``current_tier`` is then terminal and its result is indexed as-is. + """ + try: + cur_idx = TIER_LADDER.index(current_tier) + except ValueError: + return None + start_idx = cur_idx + 1 + if minimum is not None: + try: + start_idx = max(start_idx, TIER_LADDER.index(minimum)) + except ValueError: + pass + for tier in TIER_LADDER[start_idx:]: + if self._tier_available(tier, settings): + return tier + return None + + async def process_tier( + self, + content: bytes, + content_type: str, + filename: str | None, + tier: str, + options: dict[str, Any] | None = None, + progress_callback: ( + Callable[[float, float | None, str | None], Awaitable[None]] | None + ) = None, + ) -> ProcessingResult: + """Run exactly ONE extraction tier's processor on a PDF (external path). + + The per-tier procrastinate fleet calls this for the tier matching the + job's queue. Escalation to the next tier is decided separately by + :meth:`evaluate_escalation` and effected by the queue's retry strategy as + a queue-hop -- never inline here. ``escalated`` is set for any tier above + the cheapest so the parse span/metrics reflect an escalated attempt. + """ + oversize = self._oversize_result(content, filename, get_settings()) + if oversize is not None: + return oversize + processor = self._pdf_processor_for_tier(tier) + if processor is None: + raise ProcessorError( + f"No '{tier}'-tier PDF processor registered " + f"(available: {', '.join(self.list_processors())})" + ) + return await self._run_processor( + processor, + content, + content_type, + filename, + options, + progress_callback, + escalated=(tier != TIER_LADDER[0]), + ) + + def evaluate_escalation( + self, + result: ProcessingResult, + content: bytes, + current_tier: str, + settings: Any, + *, + filename: str | None = None, + ) -> tuple[str, str] | None: + """Decide whether ``current_tier``'s result must escalate (external path). + + Returns ``(to_tier, reason)`` when the parse is too poor to index and a + higher tier can run, else ``None`` (index the result as-is). Reuses the + tier-0 classifier as the post-parse quality gate, so the escalation + signal is identical to the inline pipeline's. + + A hard parse FAILURE (``result.success`` False) is never escalated: a + corrupt/encrypted PDF one engine can't open usually defeats the others + too (OCR reads the same bytes), so the caller marks it failed instead. + + Routing of the target tier: + + - ``total_chars == 0`` (scanned / no text layer) -> target the ``ocr`` + tier directly. Text-extractor tiers (``structured``) cannot conjure + text from a pure raster scan, so a structured hop would just be wasted. + - low-confidence but non-empty layer -> escalate to the next rung, so a + different in-cluster extractor can try before paying for OCR. + """ + classification = self._classify_result( + result, + content, + settings, + record=(current_tier == TIER_LADDER[0]), + filename=filename, + ) + if classification is None or classification.recommended_tier != "ocr": + return None + # A zero-page (empty/corrupt) PDF gains nothing from any tier. + if classification.page_count <= 0: + return None + if classification.total_chars == 0: + to_tier = self.next_available_tier(current_tier, settings, minimum="ocr") + reason = "empty_text" + else: + to_tier = self.next_available_tier(current_tier, settings) + reason = "low_confidence" + if to_tier is None: + return None + return (to_tier, reason) + async def _run_processor( self, processor: DocumentProcessor, diff --git a/nextcloud_mcp_server/models/semantic.py b/nextcloud_mcp_server/models/semantic.py index 35f67ddf..ff39493c 100644 --- a/nextcloud_mcp_server/models/semantic.py +++ b/nextcloud_mcp_server/models/semantic.py @@ -193,6 +193,15 @@ class VectorSyncStatusResponse(BaseResponse): "queue backend; None on the in-memory backend" ), ) + job_counts_by_queue: dict[str, dict[str, int]] | None = Field( + default=None, + description=( + "Per-tier-queue ingest job counts {queue: {status: count}} on the " + "postgres backend (Deck #323), so an operator can see whether work is " + "backed up on ingest-fast vs waiting on ingest-structured/ingest-ocr; " + "None on the in-memory backend" + ), + ) __all__ = [ diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index d6a37299..f506afbc 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -190,6 +190,21 @@ vector_sync_indexed_chunks = Gauge( "Total indexed chunks (non-placeholder points) in the vector store", ) +# Per-tier-queue ingest depth (Deck #323). One series per (queue, status) so an +# operator can see where work sits -- a ``fast`` backlog, docs waiting on +# ``ingest-structured``/``ingest-ocr``, or failures piling up per tier. KEDA +# scales each tier Deployment off the queue's ``todo`` depth via direct SQL; this +# gauge is the dashboard/alerting view of the same figures. Published by the +# periodic vector_sync metrics task from the procrastinate per-queue job counts. +ingest_queue_depth = Gauge( + "astrolabe_ingest_queue_depth", + "Ingest jobs per tier queue by status (todo/doing/failed)", + ["queue", "status"], +) +# The subset of statuses worth a gauge series; the rest (succeeded/cancelled/ +# aborted) are pruned from the queue table and uninteresting for operating. +_INGEST_DEPTH_STATUSES = ("todo", "doing", "failed") + qdrant_operations_total = Counter( "mcp_qdrant_operations_total", "Total Qdrant vector database operations", @@ -637,6 +652,23 @@ def update_vector_sync_indexed_chunks(count: int) -> None: vector_sync_indexed_chunks.set(count) +def update_ingest_queue_depth(by_queue: dict[str, dict[str, int]] | None) -> None: + """Set the per-tier-queue depth gauge from procrastinate job counts (#323). + + ``by_queue`` is ``{queue_name: {status: count}}`` (see + ``queue.procrastinate.get_ingest_job_counts_by_queue``). A queue missing a + status is set to 0 so a drained queue reads zero rather than going stale at + its last non-zero value. No-op on the memory backend (``by_queue`` is None). + """ + if not by_queue: + return + for queue, per_status in by_queue.items(): + for status in _INGEST_DEPTH_STATUSES: + ingest_queue_depth.labels(queue=queue, status=status).set( + per_status.get(status, 0) + ) + + def record_document_parse( processor: str, tier: str, diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 382236a4..0696fbcc 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -1099,6 +1099,7 @@ def configure_semantic_tools(mcp: FastMCP): enabled=True, ingest_queue=settings.ingest_queue, job_counts=pending.job_counts, + job_counts_by_queue=pending.job_counts_by_queue, ) except Exception as e: diff --git a/nextcloud_mcp_server/vector/ingest_status.py b/nextcloud_mcp_server/vector/ingest_status.py index 6f470525..f1cd7689 100644 --- a/nextcloud_mcp_server/vector/ingest_status.py +++ b/nextcloud_mcp_server/vector/ingest_status.py @@ -29,6 +29,10 @@ class IngestPending: # Per-status counts (todo/doing/failed/…) on the postgres backend; None on # the memory backend, which has no durable per-status breakdown. job_counts: dict[str, int] | None = None + # Per-tier-queue breakdown ``{queue: {status: count}}`` on the postgres + # backend (Deck #323); None on the memory backend. Feeds the per-tier status + # surface + the astrolabe_ingest_queue_depth gauge. + job_counts_by_queue: dict[str, dict[str, int]] | None = None async def get_ingest_pending( @@ -47,13 +51,27 @@ async def get_ingest_pending( """ if ingest_queue == "postgres": counts: dict[str, int] = {} - if task_producer is not None and hasattr(task_producer, "job_counts"): + by_queue: dict[str, dict[str, int]] | None = None + # Prefer the per-queue breakdown (Deck #323) and aggregate from it, so the + # fleet-wide totals and the per-tier view always agree. Fall back to the + # aggregated call for any producer that predates job_counts_by_queue. + if task_producer is not None and hasattr(task_producer, "job_counts_by_queue"): + try: + by_queue = await task_producer.job_counts_by_queue() + for per_status in by_queue.values(): + for status, value in per_status.items(): + counts[status] = counts.get(status, 0) + value + except Exception as e: + logger.warning("Failed to read ingest job counts by queue: %s", e) + elif task_producer is not None and hasattr(task_producer, "job_counts"): try: counts = await task_producer.job_counts() except Exception as e: logger.warning("Failed to read ingest job counts: %s", e) pending = counts.get("todo", 0) + counts.get("doing", 0) - return IngestPending(pending=pending, job_counts=counts) + return IngestPending( + pending=pending, job_counts=counts, job_counts_by_queue=by_queue + ) if document_receive_stream is None: return IngestPending(pending=0) diff --git a/nextcloud_mcp_server/vector/metrics_publisher.py b/nextcloud_mcp_server/vector/metrics_publisher.py index 87998201..2958ff72 100644 --- a/nextcloud_mcp_server/vector/metrics_publisher.py +++ b/nextcloud_mcp_server/vector/metrics_publisher.py @@ -29,6 +29,7 @@ from qdrant_client.models import FieldCondition, Filter, MatchValue from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.observability.metrics import ( + update_ingest_queue_depth, update_vector_sync_indexed_chunks, update_vector_sync_indexed_documents, update_vector_sync_pending_documents, @@ -96,6 +97,8 @@ async def publish_vector_sync_metrics( # Keep the legacy gauge meaningful on every consumer path, not just the # single-user one — existing dashboards/alerts reference it. update_vector_sync_queue_size(pending.pending) + # Per-tier-queue depth (Deck #323): None on the memory backend (no-op). + update_ingest_queue_depth(pending.job_counts_by_queue) except Exception as exc: # noqa: BLE001 — metrics must not break ingest logger.warning("Failed to publish pending-documents gauge: %s", exc) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index ddc5da73..6da08d79 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -6,7 +6,7 @@ Processes documents from stream: fetches content, generates embeddings, stores i import logging import time import uuid -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast import anyio import httpx @@ -14,6 +14,12 @@ from anyio.abc import TaskStatus from anyio.streams.memory import MemoryObjectReceiveStream from qdrant_client.models import PointStruct +if TYPE_CHECKING: + # Type-only: the document stack is heavy (pymupdf/_isolation) and must stay + # off processor.py's import path (#877); the runtime import is lazy. + from nextcloud_mcp_server.document_processors.base import ProcessingResult + from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry + from nextcloud_mcp_server.acl_hash import compute_acl_hash from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.config import get_settings @@ -21,6 +27,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_escalation, record_document_parse_failed, record_embedding, record_embedding_tokens, @@ -106,6 +113,59 @@ def _drop_reason(exc: BaseException) -> str: return "other" +def _is_pdf(content_type: str) -> bool: + """Whether a MIME type is a PDF (parameter-tolerant).""" + return content_type.split(";")[0].strip().lower() == "application/pdf" + + +async def _parse_pdf_tier( + registry: "ProcessorRegistry", + content: bytes, + content_type: str, + filename: str | None, + tier: str, + settings: Any, +) -> "ProcessingResult": + """Run a single extraction tier and apply the post-parse escalation gate. + + The external per-tier ingest path (Deck #323): the procrastinate worker for + ``tier`` parses with exactly that tier, then either returns the result to + index or raises ``EscalateError`` to hand the document to the next tier's + queue (the queue's retry strategy turns the raise into a native queue-hop). + The escalation metric is recorded here, at the decision point. + + A hard parse failure (``result.success`` False) is returned as-is, not + escalated -- a corrupt/encrypted/oversize PDF that one engine can't open + usually defeats the others too; the caller marks it failed. This preserves + the "OCR is an enhancement, never worse than off" invariant: a tenant who has + not enabled a higher tier (or has no processor for it) simply indexes the + cheap tier's output. + """ + # Lazy import: keep the document stack (pymupdf/_isolation) off the module + # load path; this runs only on the per-tier worker, which needs it anyway. + from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415 + EscalateError, + ) + + result = await registry.process_tier(content, content_type, filename, tier) + if result.success: + decision = registry.evaluate_escalation( + result, content, tier, settings, filename=filename + ) + if decision is not None: + to_tier, reason = decision + record_document_escalation(tier, to_tier, reason) + logger.info( + "Escalating %s %s->%s (reason=%s)", + filename or "", + tier, + to_tier, + reason, + ) + raise EscalateError(from_tier=tier, to_tier=to_tier, reason=reason) + return result + + def assign_page_numbers(chunks, page_boundaries): """Assign page numbers to chunks based on page boundaries. @@ -173,6 +233,7 @@ async def record_indexing_usage( token_count: int, total_chars: int, page_count: int | None, + pipeline_tier: str | None = None, ) -> None: """Record the billable usage events for one embedded document. @@ -213,6 +274,11 @@ async def record_indexing_usage( "doc_type": doc_type, "user_id": user_id, "total_chars": total_chars, + # Which extraction tier produced the parsed pages (Deck #323). Carried so + # the CP rollup / a future per-tier price can attribute parsing cost to + # the tier that incurred it (paid OCR vs CPU-cheap fast). None for text + # doc types, which are never parsed. + "pipeline_tier": pipeline_tier, } try: store = await UsageEventStore.shared() @@ -242,6 +308,18 @@ async def record_indexing_usage( metadata=metadata, enabled=True, ) + # Paid-OCR pages are metered as a SEPARATE line (Deck #323) so the + # expensive tier's cost is billable independently of CPU-cheap parsing + # -- pages_embedded counts all parsed pages, pages_ocr only the OCR + # tier's. Gated on the tier so it's emitted exactly when the doc was + # actually OCR'd; the same page_count guard above applies. + if pipeline_tier == "ocr": + await store.record_usage_event( + metric="pages_ocr", + value=page_count, + metadata=metadata, + enabled=True, + ) except Exception: # Reached only when shared()/store construction itself raises # (record_usage_event swallows its own write failures). Metering is on, @@ -393,7 +471,11 @@ async def _reconcile_tag_event( async def process_document( - doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3 + doc_task: DocumentTask, + nc_client: NextcloudClient, + *, + max_retries: int = 3, + tier: str | None = None, ): """ Process a single document: fetch, tokenize, embed, store in Qdrant. @@ -407,6 +489,11 @@ async def process_document( (3) suits the in-process SQLite pool, which has no durable retry. The procrastinate worker passes ``1`` so durable retry is owned by the queue (and survives worker crashes), avoiding compounding 3×N retries. + tier: Extraction tier to run for PDFs on the external per-tier path (Deck + #323) -- the procrastinate worker passes the tier matching its queue. + ``None`` (the default, used by the in-process/memory pool) runs the + inline tiered pipeline (``registry.process``: fast -> OCR escalation + in one call) and never raises ``EscalateError``. Retry layering: the embedding provider adds its own transient retry (5 attempts, 2s→60s backoff — card 309) *inside* each of these attempts. On the @@ -415,6 +502,19 @@ async def process_document( re-picked on the next scan; the procrastinate path (max_retries=1) caps it at one outer attempt (~30s) and defers. Don't stack a third retry layer here. """ + # EscalateError is a control-flow signal that arises ONLY on the per-tier + # external path (tier set). Bind the class lazily there so the in-process / + # memory path never pulls the document stack at call time (mirrors the lazy + # get_registry import; see #877). When tier is None it can't be raised, so + # the guards below stay inert. + escalate_error_cls: type[BaseException] | None = None + if tier is not None: + from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415 + EscalateError, + ) + + escalate_error_cls = EscalateError + start_time = time.time() logger.debug( @@ -484,7 +584,9 @@ async def process_document( for attempt in range(max_retries): try: - indexed = await _index_document(doc_task, nc_client, qdrant_client) + indexed = await _index_document( + doc_task, nc_client, qdrant_client, tier=tier + ) # A permanent parse failure returns False: it was already # recorded (document_parse_failed_total + the registry's @@ -506,6 +608,14 @@ async def process_document( return # Success except Exception as e: + # An escalation signal is control flow, not a failure: + # propagate it untouched so the procrastinate retry strategy + # can hop the job to the next tier's queue. Never retry it + # in-process and never count it as a drop. + if escalate_error_cls is not None and isinstance( + e, escalate_error_cls + ): + raise if attempt < max_retries - 1: logger.warning( "Retry %s/%s for %s_%s: %s", @@ -556,7 +666,12 @@ async def process_document( record_ingest_dropped(reason) raise - except Exception: + except Exception as e: + # An escalation signal must reach the procrastinate retry strategy + # un-recorded -- it is neither a processing success nor an error + # (the hop is its own event, counted via record_document_escalation). + if escalate_error_cls is not None and isinstance(e, escalate_error_cls): + raise # Single processing-error call site: catches exhausted-retry # re-raises, delete failures, and setup errors (get_qdrant_client / # get_settings) — each counted exactly once. A failed delete is not @@ -571,11 +686,20 @@ async def process_document( async def _index_document( - doc_task: DocumentTask, nc_client: NextcloudClient, qdrant_client + doc_task: DocumentTask, + nc_client: NextcloudClient, + qdrant_client, + *, + tier: str | None = None, ) -> bool | None: """ Index a single document (called by process_document with retry). + ``tier`` selects the external per-tier PDF path (Deck #323): when set and the + file is a PDF, exactly that tier is parsed and a low-quality result raises + ``EscalateError`` to hand the document to the next tier's queue. ``None`` + (default) runs the inline tiered pipeline (``registry.process``). + Returns ``False`` when a permanent parse failure means nothing was indexed (the caller must then skip the success metrics); ``None`` otherwise. @@ -800,22 +924,40 @@ async def _index_document( "vector_sync.file_size": len(content_bytes), }, ): - # The registry runs the tiered PDF pipeline (tier-0 classify -> - # tier-1 fast -> OCR escalation) and records classification metrics. - # Imported lazily so module import doesn't pull in the document stack - # (document_processors -> _isolation, Unix-only ``resource``; see #877). + # The registry runs the tiered PDF pipeline and records + # classification metrics. Imported lazily so module import doesn't + # pull in the document stack (document_processors -> _isolation, + # Unix-only ``resource``; see #877). from nextcloud_mcp_server.document_processors import ( # noqa: PLC0415 get_registry, ) + from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415 + EscalateError, + ) registry = get_registry() try: - result = await registry.process( - content=content_bytes, - content_type=content_type, - filename=file_path, - ) + # External per-tier path (Deck #323): run only this worker's tier + # for PDFs and let a low-quality parse raise EscalateError (a + # queue-hop to the next tier). Everything else -- non-PDF files, + # and the in-process/memory pool (tier is None) -- runs the inline + # tiered pipeline (fast -> OCR escalation in one call). + if tier is not None and _is_pdf(content_type): + result = await _parse_pdf_tier( + registry, + content_bytes, + content_type, + file_path, + tier, + settings, + ) + else: + result = await registry.process( + content=content_bytes, + 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 @@ -881,6 +1023,11 @@ async def _index_document( ) else: logger.debug("No page_boundaries in metadata for %s", file_path) + except EscalateError: + # Control-flow signal (per-tier path): re-raise untouched so the + # queue hops the job to the next tier. NOT a "failed to process" + # error -- don't log it as one. + raise except Exception as e: logger.error("Failed to process file %s: %s", file_path, e) raise @@ -1037,6 +1184,14 @@ async def _index_document( and not isinstance(raw_page_count, bool) else None ), + # Tier that produced the parsed pages (registry stamps it on the + # result metadata); text doc types stay "fast". Narrow defensively + # to str|None — file_metadata is loosely typed (Any values). + pipeline_tier=( + pt + if isinstance(pt := file_metadata.get("pipeline_tier"), str) + else None + ), ) async def generate_sparse_embeddings(): diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index e7b13f92..d37b5af6 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -5,11 +5,14 @@ This replaces NATS JetStream and the old Postgres-queue stub. The MCP server now owns *both* sides of ingest: - **Producer** (API role / scanner) — :class:`ProcrastinateTaskProducer.send` - *defers* one ``ingest:process_document`` job per changed document into the - per-tenant Postgres (the same app DB; procrastinate manages its own tables). -- **Consumer** (worker role) — ``nextcloud-mcp-server worker`` runs - :func:`procrastinate.App.run_worker`, which drains the ``ingest`` queue and - invokes the existing :func:`process_document` pipeline. + *defers* one ``ingest:process_document`` job per changed document onto the + cheapest tier's queue (``ingest-fast``) in the per-tenant Postgres (the same + app DB; procrastinate manages its own tables). +- **Consumer** (worker role) — ``nextcloud-mcp-server worker [--tier T]`` runs + :func:`procrastinate.App.run_worker`, which drains its tier's queue and invokes + the existing :func:`process_document` pipeline. A parse too poor to index hops + the job to the next tier's queue (see :class:`TieredEscalationStrategy`), so + cheap CPU parsing and paid OCR run on independently-scaled fleets (Deck #323). Design notes: @@ -35,9 +38,17 @@ from datetime import datetime, timezone from types import TracebackType from typing import TYPE_CHECKING -from procrastinate import App, Blueprint, JobContext, PsycopgConnector, RetryStrategy +from procrastinate import ( + App, + BaseRetryStrategy, + Blueprint, + JobContext, + PsycopgConnector, + RetryDecision, +) from procrastinate.connector import BaseConnector from procrastinate.exceptions import AlreadyEnqueued +from procrastinate.jobs import Job from ...config import get_procrastinate_conninfo, get_settings from ..scanner import DocumentTask @@ -47,14 +58,53 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Single queue for document ingest. KEDA scales the worker Deployment on the -# depth of this queue (``SELECT count(*) FROM procrastinate_jobs WHERE -# queue_name='ingest' AND status='todo'``). -INGEST_QUEUE_NAME = "ingest" +# One queue per extraction tier (Deck #323), aligned cheapest-first with +# document_processors.escalation.TIER_LADDER. Each queue is drained by its own +# worker Deployment + KEDA ScaledObject (``SELECT count(*) FROM +# procrastinate_jobs WHERE queue_name= AND status='todo'``), so a +# CPU-bound ``fast`` fleet, an in-cluster ``structured`` fleet, and a paid +# network-bound ``ocr`` fleet scale (and fail) independently. +INGEST_QUEUE_FAST = "ingest-fast" +INGEST_QUEUE_STRUCTURED = "ingest-structured" +INGEST_QUEUE_OCR = "ingest-ocr" + +# tier -> queue. The producer always defers onto the cheapest tier's queue; a +# low-quality parse hops the job up the ladder via the retry strategy below. +TIER_QUEUES: dict[str, str] = { + "fast": INGEST_QUEUE_FAST, + "structured": INGEST_QUEUE_STRUCTURED, + "ocr": INGEST_QUEUE_OCR, +} +_QUEUE_TIERS: dict[str, str] = {queue: tier for tier, queue in TIER_QUEUES.items()} +ALL_INGEST_QUEUES: tuple[str, ...] = tuple(TIER_QUEUES.values()) +# New jobs start here; ``ocr`` is reached only by escalation. +DEFAULT_INGEST_QUEUE = INGEST_QUEUE_FAST + +# Legacy single-queue name (pre-#323). A rolling upgrade may still have jobs +# parked on it; a worker can be told to drain it alongside the tier queues, and +# the job-count / reclaim helpers include it so nothing is stranded. +LEGACY_INGEST_QUEUE = "ingest" +# Back-compat alias for callers that imported the old single-queue constant. +INGEST_QUEUE_NAME = DEFAULT_INGEST_QUEUE + +# Queues the job-count + reclaim helpers sweep (tier queues + the legacy one). +_MANAGED_QUEUES: tuple[str, ...] = (*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE) + # Blueprint namespace → registered task names are prefixed ``ingest:``. _NAMESPACE = "ingest" INGEST_TASK_NAME = f"{_NAMESPACE}:process_document" + +def tier_for_queue(queue: str | None) -> str: + """Tier a worker on ``queue`` should run. Unknown/legacy -> ``fast``. + + The queue-aware task uses this to pick which single tier to parse with: the + job's current queue *is* its tier. A job on the legacy ``ingest`` queue (or + any unrecognised queue) defaults to the cheapest tier. + """ + return _QUEUE_TIERS.get(queue or "", "fast") + + # A crashed worker leaves its job in ``doing``; reclaim it once its (per-worker) # heartbeat is this many seconds stale. The default is sized well above the # longest expected ``process_document`` (PDF render + embedding) so a slow-but- @@ -69,6 +119,7 @@ INGEST_TASK_NAME = f"{_NAMESPACE}:process_document" # Blueprint cannot be added to more than one App — which the tests (in-memory + # real Postgres) and any re-init path require. async def process_document_task( + context: JobContext, *, user_id: str, doc_id: str, @@ -80,12 +131,26 @@ async def process_document_task( etag: str | None = None, owner_id: str | None = None, ) -> None: - """Worker entry: rebuild the DocumentTask, resolve creds, run the pipeline.""" + """Worker entry: rebuild the DocumentTask, resolve creds, run the pipeline. + + Queue-aware (Deck #323): the tier this worker runs is the tier of the job's + current queue. A low-quality parse raises ``EscalateError``, which the + :class:`TieredEscalationStrategy` turns into a queue-hop to the next tier. + When per-tier escalation is disabled (``INGEST_ESCALATION_ENABLED=false``), + ``tier`` stays ``None`` and the inline pipeline runs (fast -> OCR in one + call), reproducing the pre-#323 single-queue behaviour. + """ # Local imports avoid a heavy import chain at blueprint-definition time # (this module is also imported by the API pod just to defer jobs). from ..oauth_sync import NotProvisionedError # noqa: PLC0415 from ..processor import process_document # noqa: PLC0415 + tier = ( + tier_for_queue(context.job.queue) + if get_settings().ingest_escalation_enabled + else None + ) + task = DocumentTask( user_id=user_id, doc_id=doc_id, @@ -111,7 +176,7 @@ async def process_document_task( try: # Durable retry is procrastinate's job; disable the in-process loop. - await process_document(task, nc_client, max_retries=1) + await process_document(task, nc_client, max_retries=1, tier=tier) finally: await nc_client.close() @@ -127,9 +192,11 @@ async def reclaim_stalled_ingest_jobs(context: JobContext, timestamp: int) -> No retry_at = datetime.now(tz=timezone.utc) stalled_after = get_settings().ingest_stalled_job_seconds reclaimed = 0 - for job in await manager.get_stalled_jobs( - queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=stalled_after - ): + # queue=None sweeps every queue, so an orphaned job on any tier queue is + # reclaimed regardless of which tier's worker happens to run this periodic. + # retry_job_by_id_async keeps the job on its own queue, so a stalled ``ocr`` + # job re-runs on ``ingest-ocr`` (the ocr fleet), not the reclaiming worker's. + for job in await manager.get_stalled_jobs(seconds_since_heartbeat=stalled_after): if job.id is None: continue await manager.retry_job_by_id_async(job_id=job.id, retry_at=retry_at) @@ -163,6 +230,121 @@ async def _resolve_client(user_id: str) -> NextcloudClient: return await get_user_client_basic_auth(user_id, host) +def _first_leaf(exc: BaseException) -> BaseException: + """Descend nested ExceptionGroups to the first concrete leaf exception. + + An anyio task group can wrap the real cause (and nest groups); the retry + strategy classifies on the leaf, mirroring ``processor._drop_reason``. + """ + while isinstance(exc, BaseExceptionGroup) and exc.exceptions: + exc = exc.exceptions[0] + return exc + + +def _is_transient_infra_error(exc: BaseException) -> bool: + """Whether ``exc`` is a transient infra blip worth a SAME-tier retry. + + Mirrors the retryable subset of ``processor._drop_reason``: doc-fetch / + embed / Qdrant timeouts, connection drops, rate limits, and 5xx. A parse + that is merely *poor* never reaches here -- that path raises + ``EscalateError`` (handled separately) -- so this is purely about + infrastructure that should recover on its own. Imports are lazy: this only + runs in the worker, and the module is also imported by the API pod to defer. + """ + import httpx # noqa: PLC0415 + + if isinstance(exc, (httpx.TimeoutException, httpx.ConnectError)): + return True + try: + import openai # noqa: PLC0415 + + if isinstance( + exc, + ( + openai.APITimeoutError, + openai.APIConnectionError, + openai.RateLimitError, + ), + ): + return True + if isinstance(exc, openai.APIStatusError): + return exc.status_code >= 500 + except ImportError: # pragma: no cover -- openai is a hard dependency + pass + if type(exc).__module__.startswith("qdrant_client"): + return True + return False + + +class TieredEscalationStrategy(BaseRetryStrategy): + """Native procrastinate retry that escalates across tier queues (Deck #323). + + Three outcomes, decided from the raised exception: + + - ``EscalateError`` -> ``RetryDecision(queue=)``: the SAME + job hops to the next fleet's queue and is parsed once by that tier. This is + how a document is "requeued on a failed parse" -- once per tier, with no + same-tier parse retry. + - a whitelisted transient infra error (doc fetch / embed / Qdrant blip) -> + same-queue exponential backoff, while under ``max_transient_attempts``. + - anything else, the transient cap is reached, or the target tier is unknown + -> ``None`` (no retry); the placeholder was already marked failed by the + pipeline, and the next scan re-picks the document. + + Per-tier attempt accounting is intentionally approximate: a queue-hop can't + reset ``job.attempts`` (procrastinate has no per-tier counter), so parse + escalations *do* advance the same counter the transient cap reads. Because a + parse escalation hops (it never retries in place) the "once per parse per + tier" guarantee is structural; the cap is just a generous global ceiling on + transient churn across the whole lineage, not an exact per-tier count. + """ + + def __init__(self, *, max_transient_attempts: int) -> None: + self._max_transient_attempts = max_transient_attempts + + def get_retry_decision( + self, *, exception: BaseException, job: Job + ) -> RetryDecision | None: + # Lazy import: EscalateError lives in the document stack, which the API + # pod (it also builds this App to defer) must not load. get_retry_decision + # runs only in the worker, where the stack is already imported. + from ...document_processors.escalation import EscalateError # noqa: PLC0415 + + exc = _first_leaf(exception) + + if isinstance(exc, EscalateError): + queue = TIER_QUEUES.get(exc.to_tier) + if queue is None: + # Unknown target tier: don't strand the job on a queue no worker + # drains -- stop and let the placeholder/next scan handle it. + logger.error( + "ingest.escalate_unknown_tier from=%s to=%s", + exc.from_tier, + exc.to_tier, + ) + return None + logger.info( + "ingest.escalate from=%s to=%s reason=%s queue=%s", + exc.from_tier, + exc.to_tier, + exc.reason, + queue, + ) + # Immediate hop -- the next tier's fleet should pick it up at once. + return RetryDecision(queue=queue, retry_in={"seconds": 0}) + + if ( + _is_transient_infra_error(exc) + and job.attempts < self._max_transient_attempts + ): + # 4, 8, 16, ... seconds, capped at 5 min. attempts is >=1 here (the + # failing attempt is counted), so attempts-1 makes the first wait 4s. + wait = min(4 * (2 ** max(0, job.attempts - 1)), 300) + return RetryDecision(retry_in={"seconds": wait}) + + return None + + def _build_ingest_blueprint() -> Blueprint: """Create a fresh Blueprint with the ingest tasks registered. @@ -172,13 +354,22 @@ def _build_ingest_blueprint() -> Blueprint: bp = Blueprint() # Durable retry owned by the queue (survives worker crashes); the in-process # retry loop in process_document is disabled on this path via max_retries=1. - bp.task( + # The task's default queue is the cheapest tier; the producer defers there + # explicitly and the strategy hops a job up the ladder on a poor parse. + bp.task( # type: ignore[no-matching-overload] name="process_document", - queue=INGEST_QUEUE_NAME, - retry=RetryStrategy(max_attempts=5, exponential_wait=4), + queue=DEFAULT_INGEST_QUEUE, + pass_context=True, + # procrastinate's RetryValue type only admits RetryStrategy, but a custom + # BaseRetryStrategy subclass is the documented extension point (and is + # accepted at runtime by get_retry_strategy). The annotation is just too + # narrow, hence the ignore. + retry=TieredEscalationStrategy( + max_transient_attempts=get_settings().ingest_transient_max_attempts + ), )(process_document_task) reclaim = bp.task( - name="reclaim_stalled_jobs", queue=INGEST_QUEUE_NAME, pass_context=True + name="reclaim_stalled_jobs", queue=DEFAULT_INGEST_QUEUE, pass_context=True )(reclaim_stalled_ingest_jobs) bp.periodic(cron="*/5 * * * *", periodic_id="reclaim_stalled_ingest")(reclaim) return bp @@ -276,20 +467,43 @@ async def apply_ingest_queue_schema( _JOB_STATUSES = ("todo", "doing", "succeeded", "failed", "cancelled", "aborted") -async def get_ingest_job_counts(app: App | None = None) -> dict[str, int]: - """Return ingest job counts by status (``todo``/``doing``/``failed``/…). +async def get_ingest_job_counts_by_queue( + app: App | None = None, +) -> dict[str, dict[str, int]]: + """Per-queue ingest job counts by status (Deck #323). - Reads procrastinate's per-queue stats via the manager API (not hand-written - SQL) so a future schema bump doesn't silently break the status surface. The - manager flattens its per-status ``stats`` into top-level row keys, so we read - the known status keys directly. Assumes the app's connector is already open. + Returns ``{queue_name: {status: count}}`` for the managed ingest queues (the + per-tier queues + the legacy single queue) that have rows. Reads + procrastinate's per-queue stats via the manager API (not hand-written SQL) so + a future schema bump doesn't silently break the status surface. Assumes the + app's connector is already open. Feeds the per-tier status surface + the + ``astrolabe_ingest_queue_depth`` gauge. """ app = app or get_procrastinate_app() - counts: dict[str, int] = {} - for row in await app.job_manager.list_queues_async(queue=INGEST_QUEUE_NAME): + by_queue: dict[str, dict[str, int]] = {} + for row in await app.job_manager.list_queues_async(): + name = row.get("name") + if name not in _MANAGED_QUEUES: + continue + per = by_queue.setdefault(name, {}) for status in _JOB_STATUSES: if status in row: - counts[status] = counts.get(status, 0) + int(row[status]) + per[status] = per.get(status, 0) + int(row[status]) + return by_queue + + +async def get_ingest_job_counts(app: App | None = None) -> dict[str, int]: + """Aggregate ingest job counts by status across all managed queues. + + Fleet-wide totals summed over the per-tier queues + the legacy queue, so + ``pending = todo + doing`` reflects all outstanding ingest work regardless of + which tier a document currently sits on. Per-queue breakdown: + :func:`get_ingest_job_counts_by_queue`. + """ + counts: dict[str, int] = {} + for per in (await get_ingest_job_counts_by_queue(app)).values(): + for status, value in per.items(): + counts[status] = counts.get(status, 0) + value return counts @@ -336,7 +550,13 @@ class ProcrastinateTaskProducer: async def send(self, task: DocumentTask, /) -> None: key = _doc_queueing_lock(task) - deferrer = self._app.configure_task(INGEST_TASK_NAME, queueing_lock=key) + # Always defer onto the cheapest tier's queue; the escalation strategy + # hops the job up the ladder on a poor parse. queueing_lock is a global + # partial-unique on status='todo', so a doc mid-escalation on a higher + # tier still dedupes a fresh enqueue here -- no double-processing. + deferrer = self._app.configure_task( + INGEST_TASK_NAME, queue=DEFAULT_INGEST_QUEUE, queueing_lock=key + ) try: await deferrer.defer_async(**asdict(task)) except AlreadyEnqueued: @@ -358,6 +578,10 @@ class ProcrastinateTaskProducer: """Ingest job counts by status (for the vector-sync status surface).""" return await get_ingest_job_counts(self._app) + async def job_counts_by_queue(self) -> dict[str, dict[str, int]]: + """Per-tier-queue ingest job counts by status (Deck #323).""" + return await get_ingest_job_counts_by_queue(self._app) + def clone(self) -> ProcrastinateTaskProducer: return self diff --git a/tests/unit/test_processor_metering.py b/tests/unit/test_processor_metering.py index 1b9dedc7..01f612f1 100644 --- a/tests/unit/test_processor_metering.py +++ b/tests/unit/test_processor_metering.py @@ -176,3 +176,53 @@ async def test_store_failure_is_swallowed(monkeypatch): total_chars=9, page_count=2, ) + + +@pytest.mark.unit +async def test_ocr_tier_records_pages_ocr(store_spy): + """OCR-tier pages are metered as a separate pages_ocr line (Deck #323).""" + await processor.record_indexing_usage( + enabled=True, + provider="mistral", + model="mistral-embed", + doc_type="file", + user_id="alice", + chunk_count=20, + token_count=900, + total_chars=40000, + page_count=8, + pipeline_tier="ocr", + ) + by_metric = { + c.kwargs["metric"]: c.kwargs["value"] + for c in store_spy.record_usage_event.await_args_list + } + # pages_ocr fires IN ADDITION to pages_embedded for OCR-tier pages. + assert by_metric == { + "tokens_embedded": 900, + "pages_embedded": 8, + "pages_ocr": 8, + } + # pipeline_tier is threaded into the billing metadata for CP attribution. + for c in store_spy.record_usage_event.await_args_list: + assert c.kwargs["metadata"]["pipeline_tier"] == "ocr" + + +@pytest.mark.unit +async def test_fast_tier_does_not_record_pages_ocr(store_spy): + """A CPU-cheap fast-tier parse must NOT incur the paid pages_ocr line.""" + await processor.record_indexing_usage( + enabled=True, + provider="mistral", + model="mistral-embed", + doc_type="file", + user_id="alice", + chunk_count=10, + token_count=500, + total_chars=20000, + page_count=4, + pipeline_tier="fast", + ) + metrics = {c.kwargs["metric"] for c in store_spy.record_usage_event.await_args_list} + assert "pages_ocr" not in metrics + assert metrics == {"tokens_embedded", "pages_embedded"} diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index c111763d..0c0a4da6 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -241,3 +241,150 @@ async def test_no_ocr_escalation_when_disabled(monkeypatch): res = await r.process(b"%PDF-1.7", "application/pdf") # Fast tier is terminal when OCR is disabled. assert res.processor == "fast" + + +# --- Per-tier external path (Deck #323) ------------------------------------- + + +async def test_process_tier_runs_named_tier(monkeypatch): + """process_tier runs exactly the requested tier's processor, not priority.""" + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings()) + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr", "ocr"), 5), + ) + res = await r.process_tier(b"%PDF-1.7", "application/pdf", "f.pdf", "structured") + assert res.processor == "structured" + + +async def test_process_tier_unknown_tier_raises(monkeypatch): + from nextcloud_mcp_server.document_processors.base import ProcessorError + + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings()) + r = _registry((_Fake("fast", "fast"), 20)) + with pytest.raises(ProcessorError, match="structured"): + await r.process_tier(b"%PDF-1.7", "application/pdf", "f.pdf", "structured") + + +async def test_process_tier_oversize_fails_fast(monkeypatch): + """The size guard applies on the per-tier path too (before any parse).""" + monkeypatch.setattr( + reg_mod, "get_settings", lambda: _Settings(max_pdf_size_mb=0.001) + ) + r = _registry((_Fake("ocr", "ocr"), 5)) + res = await r.process_tier(b"x" * 4096, "application/pdf", "big.pdf", "ocr") + assert res.success is False + assert res.metadata["parse_failed_reason"] == "oversize" + + +def test_next_available_tier_walks_ladder(): + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr", "ocr"), 5), + ) + # ocr disabled -> structured is the only target above fast. + s = _Settings(ocr=False) + assert r.next_available_tier("fast", s) == "structured" + assert r.next_available_tier("structured", s) is None # ocr gated off + # ocr enabled -> reachable; minimum skips the structured rung. + s_ocr = _Settings(ocr=True) + assert r.next_available_tier("structured", s_ocr) == "ocr" + assert r.next_available_tier("fast", s_ocr, minimum="ocr") == "ocr" + + +def test_next_available_tier_skips_unregistered(): + # No structured processor -> fast escalates straight to ocr. + r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5)) + assert r.next_available_tier("fast", _Settings(ocr=True)) == "ocr" + + +def test_evaluate_escalation_good_text_indexes(monkeypatch): + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry( + (_Fake("fast", "fast", text="This is clean readable prose text."), 20), + (_Fake("ocr", "ocr"), 5), + ) + res = ProcessingResult( + text="This is clean readable prose text.", + metadata={ + "page_count": 1, + "page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 34}], + }, + processor="fast", + ) + assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None + + +def test_evaluate_escalation_empty_jumps_to_ocr(monkeypatch): + """A scanned (no-text-layer) result targets ocr directly, skipping structured.""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr", "ocr"), 5), + ) + res = ProcessingResult( + text="", + metadata={ + "page_count": 1, + "page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 0}], + }, + processor="fast", + ) + decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) + assert decision == ("ocr", "empty_text") + + +def test_evaluate_escalation_lowconf_goes_to_structured(monkeypatch): + """A junk-but-non-empty layer escalates to the next rung (structured).""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + junk = "x" * 40 # one long token, no whitespace -> quality ~0 + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr", "ocr"), 5), + ) + res = ProcessingResult( + text=junk, + metadata={ + "page_count": 1, + "page_boundaries": [ + {"page": 1, "start_offset": 0, "end_offset": len(junk)} + ], + }, + processor="fast", + ) + decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) + assert decision == ("structured", "low_confidence") + + +def test_evaluate_escalation_failure_not_escalated(monkeypatch): + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5)) + res = ProcessingResult( + text="", + metadata={"parse_failed_reason": "error"}, + processor="fast", + success=False, + ) + assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None + + +def test_evaluate_escalation_terminal_when_no_higher_tier(monkeypatch): + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + # Only fast registered -> nowhere to escalate even on junk text. + r = _registry((_Fake("fast", "fast"), 20)) + junk = "y" * 40 + res = ProcessingResult( + text=junk, + metadata={ + "page_count": 1, + "page_boundaries": [ + {"page": 1, "start_offset": 0, "end_offset": len(junk)} + ], + }, + processor="fast", + ) + assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None diff --git a/tests/unit/vector/test_parse_pdf_tier.py b/tests/unit/vector/test_parse_pdf_tier.py new file mode 100644 index 00000000..cc445ea3 --- /dev/null +++ b/tests/unit/vector/test_parse_pdf_tier.py @@ -0,0 +1,70 @@ +"""Unit tests for the per-tier PDF parse + escalation gate (Deck #323). + +``processor._parse_pdf_tier`` runs one tier and either returns the result to +index or raises ``EscalateError`` (a queue-hop). These exercise the decision +without standing up the full ingest pipeline. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nextcloud_mcp_server.document_processors.base import ProcessingResult +from nextcloud_mcp_server.document_processors.escalation import EscalateError +from nextcloud_mcp_server.vector import processor + +pytestmark = pytest.mark.unit + + +def _registry(result: ProcessingResult, decision): + reg = MagicMock() + reg.process_tier = AsyncMock(return_value=result) + reg.evaluate_escalation = MagicMock(return_value=decision) + return reg + + +async def test_good_parse_returns_result(monkeypatch): + rec = MagicMock() + monkeypatch.setattr(processor, "record_document_escalation", rec) + result = ProcessingResult(text="clean", metadata={}, processor="fast") + reg = _registry(result, decision=None) + out = await processor._parse_pdf_tier( + reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object() + ) + assert out is result + rec.assert_not_called() + + +async def test_low_quality_parse_raises_escalate(monkeypatch): + rec = MagicMock() + monkeypatch.setattr(processor, "record_document_escalation", rec) + result = ProcessingResult(text="", metadata={}, processor="fast") + reg = _registry(result, decision=("ocr", "empty_text")) + with pytest.raises(EscalateError) as ei: + await processor._parse_pdf_tier( + reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object() + ) + assert ei.value.from_tier == "fast" + assert ei.value.to_tier == "ocr" + assert ei.value.reason == "empty_text" + # The escalation is recorded at the decision point. + rec.assert_called_once_with("fast", "ocr", "empty_text") + + +async def test_hard_failure_returns_result_without_escalating(monkeypatch): + rec = MagicMock() + monkeypatch.setattr(processor, "record_document_escalation", rec) + result = ProcessingResult( + text="", + metadata={"parse_failed_reason": "oversize"}, + processor="size_guard", + success=False, + ) + reg = _registry(result, decision=("ocr", "empty_text")) + out = await processor._parse_pdf_tier( + reg, b"%PDF", "application/pdf", "big.pdf", "fast", settings=object() + ) + # success=False short-circuits: the gate is never consulted, no escalation. + assert out is result + reg.evaluate_escalation.assert_not_called() + rec.assert_not_called() diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index f472044a..156ce073 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/tests/unit/vector/test_procrastinate_producer.py @@ -3,6 +3,7 @@ Uses procrastinate's in-memory connector so no live Postgres is required. """ +from types import SimpleNamespace from typing import cast from unittest.mock import AsyncMock @@ -15,6 +16,11 @@ from nextcloud_mcp_server.vector.scanner import DocumentTask pytestmark = pytest.mark.unit +def _ctx(queue: str = pq.INGEST_QUEUE_FAST) -> JobContext: + """Minimal JobContext stand-in: the task only reads ``context.job.queue``.""" + return cast(JobContext, SimpleNamespace(job=SimpleNamespace(queue=queue))) + + @pytest.fixture def app(): """An App bound to the in-memory connector with the ingest tasks.""" @@ -94,18 +100,21 @@ class TestProcessDocumentTask: captured["user_id"] = user_id return fake_client - async def fake_process(task, nc_client, *, max_retries): + async def fake_process(task, nc_client, *, max_retries, tier): captured["task"] = task captured["nc_client"] = nc_client captured["max_retries"] = max_retries + captured["tier"] = tier monkeypatch.setattr(pq, "_resolve_client", fake_resolve) monkeypatch.setattr( "nextcloud_mcp_server.vector.processor.process_document", fake_process ) - # Calling the Task runs its wrapped function in-process. + # Calling the Task runs its wrapped function in-process. The job is on the + # ocr queue, so the queue-aware task must derive tier="ocr". await pq.process_document_task( + _ctx(pq.INGEST_QUEUE_OCR), user_id="alice", doc_id="42", doc_type="note", @@ -120,6 +129,8 @@ class TestProcessDocumentTask: assert captured["task"].etag == "e1" # Worker disables the in-process retry loop; durable retry is the queue's. assert captured["max_retries"] == 1 + # Tier is derived from the job's queue (escalation enabled by default). + assert captured["tier"] == "ocr" fake_client.close.assert_awaited_once() async def test_pipeline_error_propagates_and_closes_client(self, monkeypatch): @@ -130,7 +141,7 @@ class TestProcessDocumentTask: async def fake_resolve(user_id): return fake_client - async def fake_process(task, nc_client, *, max_retries): + async def fake_process(task, nc_client, *, max_retries, tier): raise RuntimeError("transient qdrant failure") monkeypatch.setattr(pq, "_resolve_client", fake_resolve) @@ -140,6 +151,7 @@ class TestProcessDocumentTask: with pytest.raises(RuntimeError, match="transient qdrant failure"): await pq.process_document_task( + _ctx(), user_id="alice", doc_id="42", doc_type="note", @@ -167,6 +179,7 @@ class TestProcessDocumentTask: # Returns cleanly (job succeeds as a no-op); pipeline never runs. await pq.process_document_task( + _ctx(), user_id="ghost", doc_id="9", doc_type="note", @@ -188,7 +201,8 @@ class TestReclaimStalledJobs: class FakeManager: async def get_stalled_jobs(self, queue=None, seconds_since_heartbeat=0): - assert queue == pq.INGEST_QUEUE_NAME + # Reclaim sweeps EVERY queue (Deck #323), so no queue filter. + assert queue is None return [Job(1), Job(2), Job(None)] # None id is skipped async def retry_job_by_id_async(self, job_id, retry_at): @@ -208,27 +222,55 @@ class TestReclaimStalledJobs: class TestGetIngestJobCounts: async def test_aggregates_stats_rows(self): class FakeManager: - async def list_queues_async(self, queue=None): - assert queue == pq.INGEST_QUEUE_NAME + async def list_queues_async(self, queue=None, **kwargs): + # Counts now aggregate across all managed queues (Deck #323), so + # the helper lists every queue and filters by name itself. + assert queue is None # procrastinate flattens per-status stats into top-level keys. return [ { - "name": "ingest", - "jobs_count": 6, + "name": "ingest-fast", + "jobs_count": 4, "todo": 3, "doing": 1, "succeeded": 0, + "failed": 0, + "cancelled": 0, + "aborted": 0, + }, + { + "name": "ingest-ocr", + "jobs_count": 2, + "todo": 0, + "doing": 0, + "succeeded": 0, "failed": 2, "cancelled": 0, "aborted": 0, - } + }, + { + # An unmanaged queue must NOT pollute ingest counts. + "name": "some-other-queue", + "jobs_count": 9, + "todo": 9, + "doing": 0, + "succeeded": 0, + "failed": 0, + "cancelled": 0, + "aborted": 0, + }, ] class FakeApp: job_manager = FakeManager() counts = await pq.get_ingest_job_counts(cast(App, FakeApp())) - assert counts["todo"] == 3 + assert counts["todo"] == 3 # only ingest-* queues, not some-other-queue assert counts["doing"] == 1 assert counts["failed"] == 2 assert counts["succeeded"] == 0 + + by_queue = await pq.get_ingest_job_counts_by_queue(cast(App, FakeApp())) + assert set(by_queue) == {"ingest-fast", "ingest-ocr"} + assert by_queue["ingest-fast"]["todo"] == 3 + assert by_queue["ingest-ocr"]["failed"] == 2 diff --git a/tests/unit/vector/test_tiered_escalation_strategy.py b/tests/unit/vector/test_tiered_escalation_strategy.py new file mode 100644 index 00000000..215eafc6 --- /dev/null +++ b/tests/unit/vector/test_tiered_escalation_strategy.py @@ -0,0 +1,100 @@ +"""Unit tests for the per-tier escalation primitives (Deck #323). + +Covers the tier-ladder helpers + EscalateError (document_processors.escalation) +and the procrastinate TieredEscalationStrategy that turns a raised exception +into a queue-hop / same-tier retry / give-up decision. +""" + +import httpx +import pytest +from procrastinate.jobs import Job + +import nextcloud_mcp_server.vector.queue.procrastinate as pq +from nextcloud_mcp_server.document_processors.escalation import ( + TIER_LADDER, + EscalateError, + next_tier, +) + +pytestmark = pytest.mark.unit + + +def _job(queue: str = pq.INGEST_QUEUE_FAST, attempts: int = 1) -> Job: + return Job( + id=1, + queue=queue, + task_name=pq.INGEST_TASK_NAME, + lock=None, + queueing_lock=None, + attempts=attempts, + ) + + +class TestLadder: + def test_next_tier_ordering(self): + assert next_tier("fast") == "structured" + assert next_tier("structured") == "ocr" + assert next_tier("ocr") is None # terminal + assert next_tier("unknown") is None + + def test_ladder_is_cheapest_first(self): + assert TIER_LADDER == ("fast", "structured", "ocr") + + def test_tier_for_queue(self): + assert pq.tier_for_queue(pq.INGEST_QUEUE_OCR) == "ocr" + assert pq.tier_for_queue(pq.INGEST_QUEUE_STRUCTURED) == "structured" + # Legacy / unknown / None all fall back to the cheapest tier. + assert pq.tier_for_queue(pq.LEGACY_INGEST_QUEUE) == "fast" + assert pq.tier_for_queue(None) == "fast" + + +class TestTieredEscalationStrategy: + def _strategy(self, max_transient: int = 5): + return pq.TieredEscalationStrategy(max_transient_attempts=max_transient) + + def test_escalate_hops_to_target_queue(self): + exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text") + decision = self._strategy().get_retry_decision(exception=exc, job=_job()) + assert decision is not None + assert decision.queue == pq.INGEST_QUEUE_OCR + + def test_escalate_to_structured(self): + exc = EscalateError( + from_tier="fast", to_tier="structured", reason="low_confidence" + ) + decision = self._strategy().get_retry_decision(exception=exc, job=_job()) + assert decision is not None + assert decision.queue == pq.INGEST_QUEUE_STRUCTURED + + def test_escalate_unknown_tier_gives_up(self): + exc = EscalateError(from_tier="ocr", to_tier="bogus", reason="low_confidence") + decision = self._strategy().get_retry_decision(exception=exc, job=_job()) + assert decision is None + + def test_escalate_unwraps_exception_group(self): + exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text") + group = ExceptionGroup("wrapped", [exc]) + decision = self._strategy().get_retry_decision(exception=group, job=_job()) + assert decision is not None + assert decision.queue == pq.INGEST_QUEUE_OCR + + def test_transient_retries_same_queue_under_cap(self): + decision = self._strategy(max_transient=5).get_retry_decision( + exception=httpx.ConnectError("refused"), job=_job(attempts=1) + ) + assert decision is not None + # Same-tier retry: no queue override (stays on its current queue). + assert decision.queue is None + assert decision.retry_at is not None + + def test_transient_gives_up_over_cap(self): + decision = self._strategy(max_transient=5).get_retry_decision( + exception=httpx.ConnectError("refused"), job=_job(attempts=5) + ) + assert decision is None + + def test_non_transient_error_gives_up(self): + decision = self._strategy().get_retry_decision( + exception=ValueError("permanent"), job=_job(attempts=1) + ) + assert decision is None From 35f8204a169c4fc398563a640be6ab708eb3b5d6 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 13:34:03 +0200 Subject: [PATCH 2/6] fix(ingest): address review round 1 (reclaim queue + tests) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Register the periodic stalled-job reclaim on a dedicated ingest-maintenance queue that every worker drains (any --tier), so reclaim still fires when the fast fleet is scaled to zero and only ocr workers run. procrastinate's periodic-defer dedup keeps it single-run across drainers. - escalation: mark `unsupported`/`forced` reason labels as reserved (not raised). - processor: note that options/progress_callback are intentionally not threaded through _parse_pdf_tier yet (symmetric with the inline path). - tests: assert TieredEscalationStrategy backoff progression (4/8/16/…/300s); cover get_ingest_pending per-queue aggregation + the legacy job_counts fallback; add an external-path zero-page no-escalation case; use the canonical INGEST_QUEUE_FAST instead of the back-compat alias. Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/cli.py | 10 ++++-- .../document_processors/escalation.py | 6 ++-- nextcloud_mcp_server/vector/processor.py | 4 +++ .../vector/queue/procrastinate.py | 14 +++++++- tests/unit/test_registry_tiering.py | 12 +++++++ tests/unit/vector/test_ingest_status.py | 32 ++++++++++++++++--- .../vector/test_procrastinate_producer.py | 3 +- .../vector/test_tiered_escalation_strategy.py | 16 ++++++++++ 8 files changed, 86 insertions(+), 11 deletions(-) diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index a169ffaa..9c191ad1 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -384,6 +384,7 @@ def worker(concurrency: int | None, tier: str | None): from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 ALL_INGEST_QUEUES, + INGEST_QUEUE_MAINTENANCE, LEGACY_INGEST_QUEUE, TIER_QUEUES, apply_ingest_queue_schema, @@ -392,11 +393,14 @@ def worker(concurrency: int | None, tier: str | None): # Which queues this process drains. A single tier -> just its queue; no tier # -> every tier queue PLUS the legacy single queue, so a rolling upgrade - # never strands jobs deferred under the pre-#323 name. + # never strands jobs deferred under the pre-#323 name. Every worker also + # drains the maintenance queue so the periodic stalled-job reclaim fires + # regardless of which tier(s) are scaled up (procrastinate dedups the + # periodic, so multiple drainers don't multiply the reclaim). if tier is not None: - queues = [TIER_QUEUES[tier]] + queues = [TIER_QUEUES[tier], INGEST_QUEUE_MAINTENANCE] else: - queues = [*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE] + queues = [*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE, INGEST_QUEUE_MAINTENANCE] # This is the consumer side of the distributed (postgres) ingest backend. # Unlike the in-process anyio pool, the worker talks to procrastinate's App diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py index c3ec30bd..a49bf6df 100644 --- a/nextcloud_mcp_server/document_processors/escalation.py +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -53,8 +53,10 @@ class EscalateError(Exception): the junk text is never indexed, and it must never be swallowed by a broad ``except Exception`` on the indexing path. - ``reason`` uses the existing escalation label vocabulary: - ``empty_text`` | ``low_confidence`` | ``unsupported`` | ``forced``. + ``reason`` uses the existing escalation label vocabulary. This PR raises + ``empty_text`` (scanned / no text layer) and ``low_confidence`` (junk text + layer); ``unsupported`` and ``forced`` are reserved for future callers and + not raised yet. """ def __init__(self, *, from_tier: str, to_tier: str, reason: str) -> None: diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 6da08d79..190f207f 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -147,6 +147,10 @@ async def _parse_pdf_tier( EscalateError, ) + # options / progress_callback are not threaded here -- the indexing caller + # passes neither today, and the inline path (registry.process) omits them + # too. Forward them if a tier processor ever needs per-call tuning (e.g. OCR + # DPI); keeping the two paths symmetric until then. result = await registry.process_tier(content, content_type, filename, tier) if result.success: decision = registry.evaluate_escalation( diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index d37b5af6..9f926da6 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -87,6 +87,14 @@ LEGACY_INGEST_QUEUE = "ingest" # Back-compat alias for callers that imported the old single-queue constant. INGEST_QUEUE_NAME = DEFAULT_INGEST_QUEUE +# Maintenance queue carrying ONLY the periodic stalled-job reclaim (no document +# jobs). Every worker drains it regardless of --tier, so the reclaim fires even +# in an asymmetric deployment where the fast fleet is scaled to zero and only +# ocr workers run. procrastinate's periodic-defer dedup ensures exactly one +# worker runs each tick even when many drain this queue. Kept off document +# queues so tier isolation (which fleet processes which docs) is preserved. +INGEST_QUEUE_MAINTENANCE = "ingest-maintenance" + # Queues the job-count + reclaim helpers sweep (tier queues + the legacy one). _MANAGED_QUEUES: tuple[str, ...] = (*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE) @@ -368,8 +376,12 @@ def _build_ingest_blueprint() -> Blueprint: max_transient_attempts=get_settings().ingest_transient_max_attempts ), )(process_document_task) + # Reclaim runs on the dedicated maintenance queue (every worker drains it), + # not a tier queue -- otherwise an ocr-only deployment (fast scaled to zero) + # would never fire the periodic and orphaned ``doing`` jobs would never be + # reclaimed. The task itself sweeps ALL queues (get_stalled_jobs(queue=None)). reclaim = bp.task( - name="reclaim_stalled_jobs", queue=DEFAULT_INGEST_QUEUE, pass_context=True + name="reclaim_stalled_jobs", queue=INGEST_QUEUE_MAINTENANCE, pass_context=True )(reclaim_stalled_ingest_jobs) bp.periodic(cron="*/5 * * * *", periodic_id="reclaim_stalled_ingest")(reclaim) return bp diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index 0c0a4da6..1c0f6dc4 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -388,3 +388,15 @@ def test_evaluate_escalation_terminal_when_no_higher_tier(monkeypatch): processor="fast", ) assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None + + +def test_evaluate_escalation_zero_page_does_not_escalate(monkeypatch): + """A zero-page (empty/corrupt) PDF never escalates on the external path.""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5)) + res = ProcessingResult( + text="", + metadata={"page_count": 0, "page_boundaries": []}, + processor="fast", + ) + assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None diff --git a/tests/unit/vector/test_ingest_status.py b/tests/unit/vector/test_ingest_status.py index 27923cd2..71d8fd20 100644 --- a/tests/unit/vector/test_ingest_status.py +++ b/tests/unit/vector/test_ingest_status.py @@ -11,21 +11,45 @@ pytestmark = pytest.mark.unit class TestGetIngestPending: - async def test_postgres_reads_job_counts(self): + async def test_postgres_aggregates_by_queue(self): + """Per-queue counts are summed into fleet-wide totals (Deck #323).""" producer = AsyncMock() - producer.job_counts.return_value = {"todo": 5, "doing": 2, "failed": 1} + producer.job_counts_by_queue.return_value = { + "ingest-fast": {"todo": 5, "doing": 1}, + "ingest-ocr": {"doing": 1, "failed": 1}, + } result = await get_ingest_pending( task_producer=producer, document_receive_stream=None, ingest_queue="postgres", ) - assert result.pending == 7 # todo + doing + assert result.pending == 7 # todo(5) + doing(1+1) assert result.job_counts == {"todo": 5, "doing": 2, "failed": 1} + assert result.job_counts_by_queue == { + "ingest-fast": {"todo": 5, "doing": 1}, + "ingest-ocr": {"doing": 1, "failed": 1}, + } + + async def test_postgres_falls_back_to_aggregate_counts(self): + """A producer without job_counts_by_queue uses the aggregated call.""" + + class LegacyProducer: + async def job_counts(self): + return {"todo": 5, "doing": 2, "failed": 1} + + result = await get_ingest_pending( + task_producer=LegacyProducer(), + document_receive_stream=None, + ingest_queue="postgres", + ) + assert result.pending == 7 + assert result.job_counts == {"todo": 5, "doing": 2, "failed": 1} + assert result.job_counts_by_queue is None async def test_postgres_degrades_to_zero_on_error(self): producer = AsyncMock() - producer.job_counts.side_effect = RuntimeError("db down") + producer.job_counts_by_queue.side_effect = RuntimeError("db down") result = await get_ingest_pending( task_producer=producer, diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index 156ce073..370eb22f 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/tests/unit/vector/test_procrastinate_producer.py @@ -48,7 +48,8 @@ class TestProcrastinateTaskProducer: assert len(jobs) == 1 job = jobs[0] assert job["task_name"] == pq.INGEST_TASK_NAME - assert job["queue_name"] == pq.INGEST_QUEUE_NAME + # New jobs are deferred onto the cheapest tier's queue. + assert job["queue_name"] == pq.INGEST_QUEUE_FAST assert job["queueing_lock"] == "alice:note:42" assert job["lock"] is None # no execution lock (crash-deadlock guard) assert job["args"]["doc_id"] == "42" diff --git a/tests/unit/vector/test_tiered_escalation_strategy.py b/tests/unit/vector/test_tiered_escalation_strategy.py index 215eafc6..bbb08eff 100644 --- a/tests/unit/vector/test_tiered_escalation_strategy.py +++ b/tests/unit/vector/test_tiered_escalation_strategy.py @@ -5,6 +5,8 @@ and the procrastinate TieredEscalationStrategy that turns a raised exception into a queue-hop / same-tier retry / give-up decision. """ +from datetime import datetime, timezone + import httpx import pytest from procrastinate.jobs import Job @@ -87,6 +89,20 @@ class TestTieredEscalationStrategy: assert decision.queue is None assert decision.retry_at is not None + def test_transient_backoff_progression(self): + # min(4 * 2**(attempts-1), 300): 4, 8, 16, ... capped at 300s. + strat = self._strategy(max_transient=100) + for attempts, expected in [(1, 4), (2, 8), (3, 16), (4, 32), (20, 300)]: + decision = strat.get_retry_decision( + exception=httpx.ConnectError("x"), job=_job(attempts=attempts) + ) + assert decision is not None and decision.retry_at is not None + delta = (decision.retry_at - datetime.now(timezone.utc)).total_seconds() + # retry_at = now + wait; allow a small window for execution time. + assert expected - 2 <= delta <= expected + 1, ( + f"attempts={attempts}: delta={delta:.2f}s, expected≈{expected}s" + ) + def test_transient_gives_up_over_cap(self): decision = self._strategy(max_transient=5).get_retry_decision( exception=httpx.ConnectError("refused"), job=_job(attempts=5) From e7c0c23486d33cd8c3640053b7717fdf44fec1cd Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 13:45:01 +0200 Subject: [PATCH 3/6] fix(ingest): address review round 2 (stale gauge + hygiene) - metrics: update_ingest_queue_depth now pre-zeroes every managed ingest queue before applying live counts, so a queue that drains to empty (and drops out of procrastinate's list_queues_async) reads 0 instead of sticking at its last non-zero value (ghost backlog in Grafana/alerts). Adds a regression test. - procrastinate: comment that _is_transient_infra_error treats all qdrant errors as transient deliberately (bounded same-tier retry; over-broad is acceptable). - escalation: note next_tier is the building block; production routing uses ProcessorRegistry.next_available_tier. - tests: add evaluate_escalation fast+ocr-only low-confidence -> ocr case. Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../document_processors/escalation.py | 9 +++--- nextcloud_mcp_server/observability/metrics.py | 21 ++++++++++-- .../vector/queue/procrastinate.py | 7 ++++ tests/unit/test_ingest_queue_depth_metric.py | 32 +++++++++++++++++++ tests/unit/test_registry_tiering.py | 20 ++++++++++++ 5 files changed, 82 insertions(+), 7 deletions(-) create mode 100644 tests/unit/test_ingest_queue_depth_metric.py diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py index a49bf6df..7e39d765 100644 --- a/nextcloud_mcp_server/document_processors/escalation.py +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -30,10 +30,11 @@ def next_tier(current: str) -> str | None: """The next tier above ``current`` in the ladder, or ``None`` if terminal. Pure ordering only -- it does not consider whether the next tier is - *available* (a processor registered / OCR enabled). Callers that need - availability resolve it against the registry + settings (see - ``ProcessorRegistry.next_available_tier``); a tier with no escalation target - is terminal and its result is indexed as-is. + *available* (a processor registered / OCR enabled). **Production routing uses + ``ProcessorRegistry.next_available_tier``**, which layers availability on top + of this ordering; ``next_tier`` itself is the underlying building block + (referenced directly by tests). A tier with no escalation target is terminal + and its result is indexed as-is. """ try: idx = TIER_LADDER.index(current) diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index f506afbc..71dc4728 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -656,12 +656,27 @@ def update_ingest_queue_depth(by_queue: dict[str, dict[str, int]] | None) -> Non """Set the per-tier-queue depth gauge from procrastinate job counts (#323). ``by_queue`` is ``{queue_name: {status: count}}`` (see - ``queue.procrastinate.get_ingest_job_counts_by_queue``). A queue missing a - status is set to 0 so a drained queue reads zero rather than going stale at - its last non-zero value. No-op on the memory backend (``by_queue`` is None). + ``queue.procrastinate.get_ingest_job_counts_by_queue``). No-op on the memory + backend (``by_queue`` is None). + + Every managed queue is zeroed first: ``list_queues_async`` stops returning a + queue once it has no jobs, so a queue that drained to empty drops out of + ``by_queue`` entirely. Without the pre-zero its gauge series would stick at + its last non-zero value (ghost backlog in Grafana/alerts) instead of reading + 0. The live counts then overwrite the zeros for queues that still have work. """ if not by_queue: return + # Lazy import to keep observability decoupled from the queue layer at module + # load (and sidestep any import cycle); both names are public constants. + from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 + ALL_INGEST_QUEUES, + LEGACY_INGEST_QUEUE, + ) + + for queue in (*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE): + for status in _INGEST_DEPTH_STATUSES: + ingest_queue_depth.labels(queue=queue, status=status).set(0) for queue, per_status in by_queue.items(): for status in _INGEST_DEPTH_STATUSES: ingest_queue_depth.labels(queue=queue, status=status).set( diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index 9f926da6..bc511f44 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -279,6 +279,13 @@ def _is_transient_infra_error(exc: BaseException) -> bool: return exc.status_code >= 500 except ImportError: # pragma: no cover -- openai is a hard dependency pass + # Deliberately over-broad: this treats ALL qdrant_client exceptions as + # transient (not just timeouts/5xx). In a healthy cluster qdrant errors are + # transient, and a bounded same-tier retry is cheap; a genuinely permanent + # qdrant fault (e.g. schema mismatch) just exhausts the transient cap and + # then gives up. So unlike _drop_reason (which only *labels* the cause), this + # may add a few retries on a non-retriable qdrant error -- an acceptable + # trade for not having to enumerate qdrant's non-retriable status codes. if type(exc).__module__.startswith("qdrant_client"): return True return False diff --git a/tests/unit/test_ingest_queue_depth_metric.py b/tests/unit/test_ingest_queue_depth_metric.py new file mode 100644 index 00000000..9ff5a5d5 --- /dev/null +++ b/tests/unit/test_ingest_queue_depth_metric.py @@ -0,0 +1,32 @@ +"""Unit test for the per-tier ingest-queue-depth gauge (Deck #323). + +Guards the round-2 fix: a queue that drains to empty (and so drops out of +procrastinate's ``list_queues_async``) must read 0, not its last non-zero value. +""" + +import pytest + +from nextcloud_mcp_server.observability.metrics import update_ingest_queue_depth + +pytestmark = pytest.mark.unit + +_METRIC = "astrolabe_ingest_queue_depth" + + +def test_drained_queue_zeroes_not_stale(metric_sample): + # ocr has a backlog this tick. + update_ingest_queue_depth({"ingest-ocr": {"todo": 4}}) + assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == 4.0 + + # Next tick ocr has drained → procrastinate omits it from by_queue entirely. + update_ingest_queue_depth({"ingest-fast": {"todo": 1}}) + # The gauge must read 0 for the drained queue, not the stale 4. + assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == 0.0 + assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == 1.0 + + +def test_none_is_noop(metric_sample): + update_ingest_queue_depth({"ingest-fast": {"doing": 2}}) + # Memory backend passes None → must not wipe the last published values. + update_ingest_queue_depth(None) + assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "doing"}) == 2.0 diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index 1c0f6dc4..7c343b14 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -400,3 +400,23 @@ def test_evaluate_escalation_zero_page_does_not_escalate(monkeypatch): processor="fast", ) assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None + + +def test_evaluate_escalation_lowconf_to_ocr_when_no_structured(monkeypatch): + """fast+ocr only: a low-confidence parse routes straight to ocr (skips the + unregistered structured rung), not to None.""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + junk = "z" * 40 # non-empty but junk -> recommended ocr, total_chars > 0 + r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5)) + res = ProcessingResult( + text=junk, + metadata={ + "page_count": 1, + "page_boundaries": [ + {"page": 1, "start_offset": 0, "end_offset": len(junk)} + ], + }, + processor="fast", + ) + decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) + assert decision == ("ocr", "low_confidence") From 44f72839ed449ebbc7ca448fc6159664358d1fab Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 14:00:20 +0200 Subject: [PATCH 4/6] fix(ingest): address review round 3 + SonarCloud reliability gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests: make the transient-backoff progression assertion load-independent by bracketing the get_retry_decision call with before/after timestamps instead of measuring against a second datetime.now() (no freezegun dependency). - tests: use pytest.approx for the ingest-queue-depth gauge assertions — SonarCloud python:S1244 (float == ) was a MAJOR reliability finding that tripped the new_reliability_rating quality gate. - processor: tighten the EscalateError lazy-bind comment (file processing already imports the document stack via get_registry; the gating only spares the delete / text-doc paths and module-load time). Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/processor.py | 9 +++++---- tests/unit/test_ingest_queue_depth_metric.py | 19 ++++++++++++++----- .../vector/test_tiered_escalation_strategy.py | 14 ++++++++++---- 3 files changed, 29 insertions(+), 13 deletions(-) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 190f207f..05619b18 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -507,10 +507,11 @@ async def process_document( one outer attempt (~30s) and defers. Don't stack a third retry layer here. """ # EscalateError is a control-flow signal that arises ONLY on the per-tier - # external path (tier set). Bind the class lazily there so the in-process / - # memory path never pulls the document stack at call time (mirrors the lazy - # get_registry import; see #877). When tier is None it can't be raised, so - # the guards below stay inert. + # external path (tier set). Bind the class lazily here, and only when a tier + # is set, so the document stack is never imported at *module load* (the #877 + # invariant) nor on the delete / text-doc call paths (file processing already + # imports it via get_registry regardless). When tier is None it can't be + # raised, so the guards below stay inert. escalate_error_cls: type[BaseException] | None = None if tier is not None: from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415 diff --git a/tests/unit/test_ingest_queue_depth_metric.py b/tests/unit/test_ingest_queue_depth_metric.py index 9ff5a5d5..0fb1a169 100644 --- a/tests/unit/test_ingest_queue_depth_metric.py +++ b/tests/unit/test_ingest_queue_depth_metric.py @@ -5,6 +5,7 @@ procrastinate's ``list_queues_async``) must read 0, not its last non-zero value. """ import pytest +from pytest import approx from nextcloud_mcp_server.observability.metrics import update_ingest_queue_depth @@ -14,19 +15,27 @@ _METRIC = "astrolabe_ingest_queue_depth" def test_drained_queue_zeroes_not_stale(metric_sample): - # ocr has a backlog this tick. + # ocr has a backlog this tick. (pytest.approx: the gauge sample is a float.) update_ingest_queue_depth({"ingest-ocr": {"todo": 4}}) - assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == 4.0 + assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == approx( + 4 + ) # Next tick ocr has drained → procrastinate omits it from by_queue entirely. update_ingest_queue_depth({"ingest-fast": {"todo": 1}}) # The gauge must read 0 for the drained queue, not the stale 4. - assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == 0.0 - assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == 1.0 + assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == approx( + 0 + ) + assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == approx( + 1 + ) def test_none_is_noop(metric_sample): update_ingest_queue_depth({"ingest-fast": {"doing": 2}}) # Memory backend passes None → must not wipe the last published values. update_ingest_queue_depth(None) - assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "doing"}) == 2.0 + assert metric_sample( + _METRIC, {"queue": "ingest-fast", "status": "doing"} + ) == approx(2) diff --git a/tests/unit/vector/test_tiered_escalation_strategy.py b/tests/unit/vector/test_tiered_escalation_strategy.py index bbb08eff..aa976efb 100644 --- a/tests/unit/vector/test_tiered_escalation_strategy.py +++ b/tests/unit/vector/test_tiered_escalation_strategy.py @@ -91,16 +91,22 @@ class TestTieredEscalationStrategy: def test_transient_backoff_progression(self): # min(4 * 2**(attempts-1), 300): 4, 8, 16, ... capped at 300s. + # procrastinate sets retry_at = utcnow() + wait at call time. Bracketing + # the call with before/after makes the assertion exact and independent of + # runner load: with before <= call_now <= after, we have + # (retry_at - after) <= wait <= (retry_at - before). strat = self._strategy(max_transient=100) for attempts, expected in [(1, 4), (2, 8), (3, 16), (4, 32), (20, 300)]: + before = datetime.now(timezone.utc) decision = strat.get_retry_decision( exception=httpx.ConnectError("x"), job=_job(attempts=attempts) ) + after = datetime.now(timezone.utc) assert decision is not None and decision.retry_at is not None - delta = (decision.retry_at - datetime.now(timezone.utc)).total_seconds() - # retry_at = now + wait; allow a small window for execution time. - assert expected - 2 <= delta <= expected + 1, ( - f"attempts={attempts}: delta={delta:.2f}s, expected≈{expected}s" + lo = (decision.retry_at - after).total_seconds() + hi = (decision.retry_at - before).total_seconds() + assert lo <= expected <= hi, ( + f"attempts={attempts}: expected={expected}s not in [{lo:.3f}, {hi:.3f}]" ) def test_transient_gives_up_over_cap(self): From ce53e21ead827c0e98258e21465e0f462f43ccc3 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 14:10:01 +0200 Subject: [PATCH 5/6] fix(ingest): zero queue-depth gauge on all-queues-drained (review round 4) - metrics: update_ingest_queue_depth guarded on `not by_queue`, which conflated None (memory backend no-op) with {} (postgres, ALL queues drained). When every queue drains at once, get_ingest_job_counts_by_queue returns {} and the pre-zero loop was skipped, leaving a stale ghost backlog in the gauge. Guard on `by_queue is None` only; add an all-drained regression test. - procrastinate: note that INGEST_TRANSIENT_MAX_ATTEMPTS is snapshotted at blueprint-build time (restart to pick up changes). Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/observability/metrics.py | 16 ++++++++++------ .../vector/queue/procrastinate.py | 3 +++ tests/unit/test_ingest_queue_depth_metric.py | 14 ++++++++++++++ 3 files changed, 27 insertions(+), 6 deletions(-) diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 71dc4728..74dab2a5 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -656,16 +656,20 @@ def update_ingest_queue_depth(by_queue: dict[str, dict[str, int]] | None) -> Non """Set the per-tier-queue depth gauge from procrastinate job counts (#323). ``by_queue`` is ``{queue_name: {status: count}}`` (see - ``queue.procrastinate.get_ingest_job_counts_by_queue``). No-op on the memory - backend (``by_queue`` is None). + ``queue.procrastinate.get_ingest_job_counts_by_queue``). No-op only on the + memory backend (``by_queue is None``); an empty dict (postgres backend with + every queue drained) still runs the pre-zero so the gauge reads 0. Every managed queue is zeroed first: ``list_queues_async`` stops returning a queue once it has no jobs, so a queue that drained to empty drops out of - ``by_queue`` entirely. Without the pre-zero its gauge series would stick at - its last non-zero value (ghost backlog in Grafana/alerts) instead of reading - 0. The live counts then overwrite the zeros for queues that still have work. + ``by_queue`` entirely (and when ALL drain, ``by_queue`` is ``{}``). Without + the pre-zero its gauge series would stick at its last non-zero value (ghost + backlog in Grafana/alerts) instead of reading 0. The live counts then + overwrite the zeros for queues that still have work. """ - if not by_queue: + # ``is None`` not ``not by_queue``: an empty dict means "postgres, all queues + # drained" and MUST still zero the gauge -- only None (memory) is the no-op. + if by_queue is None: return # Lazy import to keep observability decoupled from the queue layer at module # load (and sidestep any import cycle); both names are public constants. diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index bc511f44..4b191beb 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -379,6 +379,9 @@ def _build_ingest_blueprint() -> Blueprint: # BaseRetryStrategy subclass is the documented extension point (and is # accepted at runtime by get_retry_strategy). The annotation is just too # narrow, hence the ignore. + # Settings are snapshotted here at blueprint-build time (build_app, first + # use), so a restart is needed to pick up INGEST_TRANSIENT_MAX_ATTEMPTS + # changes -- intentional: the strategy lives for the App's lifetime. retry=TieredEscalationStrategy( max_transient_attempts=get_settings().ingest_transient_max_attempts ), diff --git a/tests/unit/test_ingest_queue_depth_metric.py b/tests/unit/test_ingest_queue_depth_metric.py index 0fb1a169..1665d392 100644 --- a/tests/unit/test_ingest_queue_depth_metric.py +++ b/tests/unit/test_ingest_queue_depth_metric.py @@ -39,3 +39,17 @@ def test_none_is_noop(metric_sample): assert metric_sample( _METRIC, {"queue": "ingest-fast", "status": "doing"} ) == approx(2) + + +def test_all_queues_drained_empty_dict_zeroes(metric_sample): + # postgres backend with every queue drained → get_ingest_job_counts_by_queue + # returns {} (list_queues_async drops empty queues). An empty dict is NOT the + # memory-backend no-op: it must still zero every managed queue's gauge. + update_ingest_queue_depth({"ingest-fast": {"todo": 9}}) + assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == approx( + 9 + ) + update_ingest_queue_depth({}) + assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == approx( + 0 + ) From 392cd49bd30503e5819deda851fc4514b260c54c Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 14:19:11 +0200 Subject: [PATCH 6/6] fix(ingest): stagger stalled-job reclaim to avoid thundering herd (round 5) A stall is often systemic (a Qdrant/embedding outage stalls every in-flight job), so reclaiming the whole batch at now() every */5min tick would thundering-herd a recovering dependency, bypassing TieredEscalationStrategy's per-job backoff. reclaim_stalled_ingest_jobs now offsets retry_at by a fixed delay (INGEST_RECLAIM_RETRY_DELAY_SECONDS, default 30s; 0 = legacy immediate). Also document the hot-vs-restart flag asymmetry: INGEST_ESCALATION_ENABLED is re-read per job; INGEST_TRANSIENT_MAX_ATTEMPTS is snapshotted at worker startup. Deck #323. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 13 +++++++++++++ nextcloud_mcp_server/vector/queue/procrastinate.py | 14 +++++++++++--- tests/unit/vector/test_procrastinate_producer.py | 8 +++++++- 3 files changed, 31 insertions(+), 4 deletions(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 397ae470..d99af469 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -240,13 +240,23 @@ _DEFAULTS: dict[str, Any] = { # queue-hop. When false the ``fast`` tier is terminal -- reproduces the # pre-#323 behaviour where the cheap tier's output is indexed as-is. No effect # on the in-process ``memory`` backend, which keeps the inline escalation. + # HOT: re-read per job (process_document_task), so it takes effect on the next + # job -- unlike INGEST_TRANSIENT_MAX_ATTEMPTS, which is snapshotted at worker + # startup and needs a restart. "ingest_escalation_enabled": True, # Global cap on SAME-tier retries for transient infra errors (doc fetch / # embed / Qdrant blips) on the procrastinate path. Parse-quality failures # escalate (one parse attempt per tier) and do NOT consume this budget; only # whitelisted transient exceptions retry in place. Shared across tiers because # a queue-hop cannot reset a per-tier counter (see TieredEscalationStrategy). + # Snapshotted at worker startup (blueprint build); restart to change it. "ingest_transient_max_attempts": 5, + # Delay (seconds) before a reclaimed stalled job is re-run. A stall is often + # systemic (Qdrant/embedding outage), so reclaiming every crashed job at + # now() would thundering-herd a recovering dependency every reclaim tick + # (*/5min), bypassing TieredEscalationStrategy's per-job backoff. A small + # fixed delay staggers the retry. 0 = immediate (legacy behaviour). + "ingest_reclaim_retry_delay_seconds": 30, "collection_metadata_source": "qdrant", # qdrant | api # CP base URL for COLLECTION_METADATA_SOURCE=api (e.g. http://control-plane). # Required only when the source is api. @@ -331,6 +341,7 @@ _dynaconf = Dynaconf( # Positive integers Validator("INGEST_STALLED_JOB_SECONDS", gte=1), Validator("INGEST_TRANSIENT_MAX_ATTEMPTS", gte=1), + Validator("INGEST_RECLAIM_RETRY_DELAY_SECONDS", gte=0), Validator("VECTOR_SYNC_SCAN_INTERVAL", gte=1), Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1), Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1), @@ -872,6 +883,7 @@ class Settings: ingest_delete_succeeded_jobs: bool = True # drop succeeded ingest jobs ingest_escalation_enabled: bool = True # per-tier queue-hop (Deck #323) ingest_transient_max_attempts: int = 5 # same-tier transient-retry cap + ingest_reclaim_retry_delay_seconds: int = 30 # stagger reclaimed-job retries collection_metadata_source: str = "qdrant" # qdrant | api collection_metadata_api_url: str | None = None # CP URL when source=api embedding_gateway_url: str | None = None # required when provider=gateway @@ -1499,6 +1511,7 @@ def get_settings() -> Settings: "ingest_delete_succeeded_jobs": "INGEST_DELETE_SUCCEEDED_JOBS", "ingest_escalation_enabled": "INGEST_ESCALATION_ENABLED", "ingest_transient_max_attempts": "INGEST_TRANSIENT_MAX_ATTEMPTS", + "ingest_reclaim_retry_delay_seconds": "INGEST_RECLAIM_RETRY_DELAY_SECONDS", "collection_metadata_source": "COLLECTION_METADATA_SOURCE", "collection_metadata_api_url": "COLLECTION_METADATA_API_URL", "embedding_gateway_url": "EMBEDDING_GATEWAY_URL", diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index 4b191beb..c93c9d10 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -34,7 +34,7 @@ from __future__ import annotations import logging from dataclasses import asdict -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from types import TracebackType from typing import TYPE_CHECKING @@ -197,8 +197,16 @@ async def reclaim_stalled_ingest_jobs(context: JobContext, timestamp: int) -> No periodic-run marker (unused). """ manager = context.app.job_manager - retry_at = datetime.now(tz=timezone.utc) - stalled_after = get_settings().ingest_stalled_job_seconds + settings = get_settings() + # Stagger the re-run rather than retry at now(): a stall is often systemic (a + # Qdrant / embedding outage stalls every in-flight job), so reclaiming the + # whole batch immediately every tick would thundering-herd a recovering + # dependency, bypassing TieredEscalationStrategy's per-job backoff. The fixed + # delay spreads them out; 0 restores the legacy immediate retry. + retry_at = datetime.now(tz=timezone.utc) + timedelta( + seconds=settings.ingest_reclaim_retry_delay_seconds + ) + stalled_after = settings.ingest_stalled_job_seconds reclaimed = 0 # queue=None sweeps every queue, so an orphaned job on any tier queue is # reclaimed regardless of which tier's worker happens to run this periodic. diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index 370eb22f..32342d37 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/tests/unit/vector/test_procrastinate_producer.py @@ -192,9 +192,10 @@ class TestProcessDocumentTask: class TestReclaimStalledJobs: async def test_reclaims_each_stalled_job(self): - from datetime import datetime + from datetime import datetime, timezone retried: list[int] = [] + retry_ats: list[datetime] = [] class Job: def __init__(self, id): @@ -209,6 +210,7 @@ class TestReclaimStalledJobs: async def retry_job_by_id_async(self, job_id, retry_at): assert isinstance(retry_at, datetime) retried.append(job_id) + retry_ats.append(retry_at) class FakeApp: job_manager = FakeManager() @@ -216,8 +218,12 @@ class TestReclaimStalledJobs: class Ctx: app = FakeApp() + before = datetime.now(tz=timezone.utc) await pq.reclaim_stalled_ingest_jobs(cast(JobContext, Ctx()), timestamp=0) assert retried == [1, 2] + # Reclaimed jobs are staggered into the future (default 30s) rather than + # retried at now(), so a systemic outage doesn't thundering-herd. + assert all((ra - before).total_seconds() >= 25 for ra in retry_ats) class TestGetIngestJobCounts: