From 35f8204a169c4fc398563a640be6ab708eb3b5d6 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 13 Jun 2026 13:34:03 +0200 Subject: [PATCH] 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)