From 6f74fe67b377d9c1c62b82110b249ad56e8f4d9c Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Wed, 27 May 2026 12:10:42 +0300 Subject: [PATCH] Cap claimed-row count at max_concurren --- docs/ingester.md | 5 +++++ haiku_rag_slim/haiku/rag/ingester/workers/pool.py | 11 ++++++----- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/docs/ingester.md b/docs/ingester.md index b00153de..44c407ad 100644 --- a/docs/ingester.md +++ b/docs/ingester.md @@ -186,6 +186,11 @@ once `claim_timeout_s` elapses. - `claim_timeout_s` must exceed the longest legitimate job duration; a shorter value lets the reaper resurrect in-flight jobs. - `worker_count <= max_concurrent`; extras stall in the semaphore. +- `max_concurrent` should match downstream capacity. docling-serve + processes one task per instance, so `max_concurrent` above the number + of `providers.docling_serve.base_url` entries over-subscribes the + fleet — extra submissions queue inside docling-serve and inflate + `claimed_at` duration toward `claim_timeout_s`. - `poll_idle_interval_s`: lower = faster pickup, more SQLite churn. - `reaper_interval_s`: worst-case post-crash reclaim is `claim_timeout_s + reaper_interval_s`. diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 14970076..75026e42 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -132,12 +132,13 @@ class WorkerPool: if self._breaker.is_open: await self._sleep_or_stop(self._poll_idle_s) continue - job = await self._jobs.claim_next(worker_id) - if job is None: - await self._sleep_or_stop(self._poll_idle_s) - continue - + # Semaphore wraps claim + process: at most max_concurrent workers + # hold a claimed job at any one time. async with self._semaphore: + job = await self._jobs.claim_next(worker_id) + if job is None: + await self._sleep_or_stop(self._poll_idle_s) + continue await self._process(job) async def _reaper_loop(self) -> None: