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) <noreply@anthropic.com>

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",

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)
+    settings = get_settings()
-    stalled_after = get_settings().ingest_stalled_job_seconds
+    # 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.

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: