fix(cascade): 限制批次内并发执行的 handler 数量

src/everos/memory/cascade/orchestrator.py

@@ -33,6 +33,7 @@
 from .scanner import CascadeScanner
 from .watcher import CascadeWatcher
 from .worker import (
+    DEFAULT_HANDLER_CONCURRENCY,
     DEFAULT_OPTIMIZE_HEARTBEAT_SECONDS,
     DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS,
     DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS,
@@ -153,6 +154,7 @@ class CascadeConfig:
 
     scan_interval_seconds: float = 30.0
     worker_batch_size: int = 50
+    worker_handler_concurrency: int = DEFAULT_HANDLER_CONCURRENCY
     worker_max_retry: int = 3
     worker_poll_interval_seconds: float = 1.0
     worker_retry_backoff_seconds: float = 2.0
@@ -200,6 +202,7 @@ def __init__(
         self._worker = CascadeWorker(
             self._handlers,
             batch_size=self._config.worker_batch_size,
+            handler_concurrency=self._config.worker_handler_concurrency,
             max_retry=self._config.worker_max_retry,
             poll_interval_seconds=self._config.worker_poll_interval_seconds,
             retry_backoff_seconds=self._config.worker_retry_backoff_seconds,

src/everos/memory/cascade/worker.py

@@ -26,9 +26,10 @@
 bounds the total retry budget so a persistently-failing row cannot
 retry forever.
 
-Batch processing is concurrent inside a batch (``asyncio.gather``);
-ordering across rows is best-effort — the LSN gives a deterministic
-prefix but the handlers themselves are independent.
+Batch processing is bounded-concurrent inside a batch; rows are claimed in
+batches of 50 but only a small fixed number of handlers execute at once.
+Ordering across rows is best-effort — the LSN gives a deterministic prefix
+but the handlers themselves are independent.
 
 After a batch completes, each kind that mutated its LanceDB table is
 passed to :meth:`_schedule_optimize` — a per-kind throttle + trailing
@@ -73,6 +74,30 @@
 
 # Conservative defaults — surface in settings if tuning is needed.
 DEFAULT_BATCH_SIZE = 50
+DEFAULT_HANDLER_CONCURRENCY = 4
+"""Maximum handler executions in flight on one worker.
+
+The bound is per worker, not per :meth:`CascadeWorker.drain_once` call: a
+server drains from two places at once -- the background loop and whatever
+calls ``drain_fully`` (``POST /cascade/quiesce`` does, before ``stop``) --
+and a semaphore built per call would let those two overlap at twice this
+number, in exactly the backlog-plus-quiesce case the bound exists for.
+
+A daily-log handler first resolves existing rows by ``md_path``. On fragmented
+Lance tables that lookup is a native flat scan: 50 simultaneous handlers grew
+the production-data replay past 8.29 GiB and did not finish in 80 seconds,
+while 1/2/4/8-way peaks were about 0.26/0.61/0.98/1.73 GiB. Keeping the claim
+batch at 50 preserves queue semantics; bounding only execution prevents native
+queries that outlive Python cancellation from multiplying memory and file
+handles after a timeout.
+
+A slot is held across :meth:`_process_one`, inline retry backoff included, so
+a row failing against a down provider parks one slot for
+``retry_backoff * max_retry * (max_retry + 1) / 2`` seconds (12s at the
+defaults) with no native query in flight. That is deliberate -- backing off
+harder against a provider that is already failing is the behaviour we want --
+but it is why a fully-failing batch drains slower than it did unbounded.
+"""
 DEFAULT_MAX_RETRY = 3
 DEFAULT_POLL_INTERVAL_SECONDS = 1.0
 DEFAULT_RETRY_BACKOFF_SECONDS = 2.0
@@ -450,6 +475,7 @@ def __init__(
         handlers: dict[str, Handler],
         *,
         batch_size: int = DEFAULT_BATCH_SIZE,
+        handler_concurrency: int = DEFAULT_HANDLER_CONCURRENCY,
         max_retry: int = DEFAULT_MAX_RETRY,
         poll_interval_seconds: float = DEFAULT_POLL_INTERVAL_SECONDS,
         retry_backoff_seconds: float = DEFAULT_RETRY_BACKOFF_SECONDS,
@@ -465,8 +491,12 @@ def __init__(
             DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS
         ),
     ) -> None:
+        if handler_concurrency <= 0:
+            raise ValueError("handler_concurrency must be positive")
         self._handlers = handlers
         self._batch_size = batch_size
+        self._handler_concurrency = handler_concurrency
+        self._handler_semaphore = asyncio.Semaphore(handler_concurrency)
         self._max_retry = max_retry
         self._poll_interval = poll_interval_seconds
         self._retry_backoff = retry_backoff_seconds
@@ -497,7 +527,11 @@ async def start(self) -> None:
         self._rebuild_task = self._spawn_loop(
             "rebuild", self._rebuild_loop, "cascade-worker-rebuild"
         )
-        logger.info("cascade_worker_started", batch_size=self._batch_size)
+        logger.info(
+            "cascade_worker_started",
+            batch_size=self._batch_size,
+            handler_concurrency=self._handler_concurrency,
+        )
 
     def _spawn_loop(
         self,
@@ -655,7 +689,12 @@ async def drain_once(self, *, kinds: set[str] | None = None) -> int:
         )
         if not batch:
             return 0
-        results = await asyncio.gather(*(self._process_one(row) for row in batch))
+
+        async def process_bounded(row: MdChangeState) -> str | None:
+            async with self._handler_semaphore:
+                return await self._process_one(row)
+
+        results = await asyncio.gather(*(process_bounded(row) for row in batch))
         touched_kinds = {kind for kind in results if kind is not None}
         for kind in touched_kinds:
             self._schedule_optimize(kind)

tests/unit/test_memory/test_cascade/test_worker.py

@@ -95,6 +95,23 @@ async def handle_deleted(self, md_path: str) -> HandlerOutcome:
         )
 
 
+class _ConcurrencyTrackingHandler(_OkHandler):
+    """Records peak in-flight handler calls for the batch bound regression."""
+
+    def __init__(self) -> None:
+        self.active = 0
+        self.peak = 0
+
+    async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome:
+        self.active += 1
+        self.peak = max(self.peak, self.active)
+        try:
+            await asyncio.sleep(0.01)
+            return await super().handle_added_or_modified(md_path)
+        finally:
+            self.active -= 1
+
+
 class _BareExceptionHandler(_OkHandler):
     async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome:
         raise RuntimeError("unexpected boom")
@@ -137,6 +154,63 @@ async def test_ok_handler_marks_done(patched_repo: _FakeRepo) -> None:
     assert patched_repo.failed == []
 
 
+async def test_large_batch_limits_concurrent_handler_execution(
+    patched_repo: _FakeRepo,
+) -> None:
+    """Claim 50 rows, but never execute more than four handlers at once."""
+    patched_repo.batch = [_Row(md_path=f"{i}.md") for i in range(50)]
+    handler = _ConcurrencyTrackingHandler()
+    w = CascadeWorker({"episode": handler}, retry_backoff_seconds=0)
+
+    processed = await w.drain_once()
+
+    assert processed == 50
+    assert len(patched_repo.done) == 50
+    # ``==``, not ``<=``: a bound accidentally set to 1 would serialize the
+    # whole batch and still satisfy ``<=``.
+    assert handler.peak == 4
+
+
+class _PerCallFakeRepo(_FakeRepo):
+    """Serves one prepared batch per claim, so two drains can overlap."""
+
+    def __init__(self, batches: list[list[_Row]]) -> None:
+        super().__init__(batch=[])
+        self._batches = list(batches)
+
+    async def claim_pending_batch(
+        self, _limit: int, *, kinds: set[str] | None = None
+    ) -> list[_Row]:
+        return self._batches.pop(0) if self._batches else []
+
+
+async def test_concurrent_drains_share_one_handler_bound(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The bound is per worker, so two overlapping drains do not double it.
+
+    ``POST /cascade/quiesce`` calls ``drain_fully`` before ``stop``, while the
+    background loop is still draining -- two ``drain_once`` calls on one
+    worker. A semaphore built per call would let that reach 8.
+    """
+    from everos.memory.cascade import worker as worker_mod
+
+    repo = _PerCallFakeRepo(
+        [
+            [_Row(md_path=f"a{i}.md") for i in range(50)],
+            [_Row(md_path=f"b{i}.md") for i in range(50)],
+        ]
+    )
+    monkeypatch.setattr(worker_mod, "md_change_state_repo", repo)
+    handler = _ConcurrencyTrackingHandler()
+    w = CascadeWorker({"episode": handler}, retry_backoff_seconds=0)
+
+    processed = await asyncio.gather(w.drain_once(), w.drain_once())
+
+    assert processed == [50, 50]
+    assert handler.peak == 4
+
+
 async def test_bare_exception_marked_permanent(patched_repo: _FakeRepo) -> None:
     """Anything that isn't ExternalServiceError counts as unrecoverable."""
     patched_repo.batch = [_Row(md_path="a.md")]