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: