From 53dde30d4ac6d33aacfdcd584907c7bd2af6ee75 Mon Sep 17 00:00:00 2001 From: yeejhyang <98456270+yeejhyang@users.noreply.github.com> Date: Wed, 16 Sep 2026 22:07:42 +0800 Subject: [PATCH] fix(models): preserve FIFO across request-admission handoff (#5459) * fix(models): make request admission enqueue atomic * test(models): cover atomic request admission FIFO entry * docs(models): record atomic admission FIFO invariant * fix(models): address request-admission review follow-ups --- .../harness/deerflow/models/AGENTS.md | 7 +++- .../deerflow/models/request_admission.py | 27 ++++++++----- ...est_model_request_admission_fifo_atomic.py | 39 +++++++++++++++++++ 3 files changed, 61 insertions(+), 12 deletions(-) create mode 100644 backend/tests/test_model_request_admission_fifo_atomic.py diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index 81398eb7d..2bf3fbf53 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -3,7 +3,10 @@ Request-admission waits follow the next scheduled admission and configured interval, capped at 50 ms; the cap must not become a minimum poll interval that limits high-RPM throughput. Local `AdmissionError` is structurally non-retriable -in LLM error handling regardless of its message text. +in LLM error handling regardless of its message text. Immediate admission and +joining the blocking FIFO are one lock-protected decision: do not split the +fast-path permit check from queue insertion, or an older caller can be overtaken +while handing off to the wait queue. - `create_chat_model(name, thinking_enabled)` instantiates LLM from config via reflection - Supports `thinking_enabled` flag with per-model `when_thinking_enabled` overrides @@ -12,7 +15,7 @@ in LLM error handling regardless of its message text. - Supports `supports_vision` flag for image understanding models - Config values starting with `$` resolved as environment variables - Missing provider modules surface actionable install hints from reflection resolvers (for example `uv add langchain-google-genai`) -- Optional `models[].request_admission` attaches a process-shared `BaseRateLimiter` at the model factory. Identical explicit groups share one FIFO across model instances, threads and event loops; implicit groups use the configured model name. Policies are immutable once registered and conflicting settings fail construction. A monotonic minimum interval spaces requests without idle-time burst credit; bounded waiters poll without occupying executor threads and unregister in `finally`. The factory strips the policy from provider kwargs and sets exposed SDK `max_retries=0` so middleware retries re-enter admission. This limits model invocations, not tokens or a distributed provider account; custom providers bypassing BaseChatModel hooks are outside the contract. Tests: `test_model_request_admission.py`. +- Optional `models[].request_admission` attaches a process-shared `BaseRateLimiter` at the model factory. Identical explicit groups share one FIFO across model instances, threads and event loops; implicit groups use the configured model name. Policies are immutable once registered and conflicting settings fail construction. A monotonic minimum interval spaces requests without idle-time burst credit; bounded waiters poll without occupying executor threads and unregister in `finally`. The factory strips the policy from provider kwargs and sets exposed SDK `max_retries=0` so middleware retries re-enter admission. This limits model invocations, not tokens or a distributed provider account; custom providers bypassing BaseChatModel hooks are outside the contract. Tests: `test_model_request_admission.py` and `test_model_request_admission_fifo_atomic.py`. ### Claude Code Credentials (`packages/harness/deerflow/models/credential_loader.py`) diff --git a/backend/packages/harness/deerflow/models/request_admission.py b/backend/packages/harness/deerflow/models/request_admission.py index ebe08e1c9..f46cff544 100644 --- a/backend/packages/harness/deerflow/models/request_admission.py +++ b/backend/packages/harness/deerflow/models/request_admission.py @@ -28,7 +28,7 @@ class RequestAdmission(BaseRateLimiter): self._lock = threading.Lock() self._waiters: deque[object] = deque() - def _try(self, ticket: object | None) -> bool: + def _try(self, ticket: object) -> bool: with self._lock: now = monotonic() if self._waiters and self._waiters[0] is not ticket: @@ -38,13 +38,20 @@ class RequestAdmission(BaseRateLimiter): self._next = now + self._interval return True - def _enqueue(self) -> object: - ticket = object() + def _try_or_enqueue(self, *, blocking: bool) -> tuple[bool, object | None]: + """Atomically admit immediately or join the FIFO before newcomers can pass.""" with self._lock: + now = monotonic() + if not self._waiters and now >= self._next: + self._next = now + self._interval + return True, None + if not blocking: + return False, None if len(self._waiters) >= self.config.max_queue_size: raise AdmissionError("LLM admission queue is full; reduce workload or increase queue capacity.") + ticket = object() self._waiters.append(ticket) - return ticket + return False, ticket def _remove(self, ticket: object) -> None: with self._lock: @@ -62,12 +69,12 @@ class RequestAdmission(BaseRateLimiter): return min(0.05, self._interval, until_next if until_next > 0 else self._interval, remaining) def acquire(self, *, blocking: bool = True) -> bool: - if self._try(None): + acquired, ticket = self._try_or_enqueue(blocking=blocking) + if acquired: return True - if not blocking: + if ticket is None: return False deadline = monotonic() + self.config.max_wait_seconds - ticket = self._enqueue() try: while True: delay = self._delay(deadline) @@ -78,12 +85,12 @@ class RequestAdmission(BaseRateLimiter): self._remove(ticket) async def aacquire(self, *, blocking: bool = True) -> bool: - if self._try(None): + acquired, ticket = self._try_or_enqueue(blocking=blocking) + if acquired: return True - if not blocking: + if ticket is None: return False deadline = monotonic() + self.config.max_wait_seconds - ticket = self._enqueue() try: while True: delay = self._delay(deadline) diff --git a/backend/tests/test_model_request_admission_fifo_atomic.py b/backend/tests/test_model_request_admission_fifo_atomic.py new file mode 100644 index 000000000..9fb0086d0 --- /dev/null +++ b/backend/tests/test_model_request_admission_fifo_atomic.py @@ -0,0 +1,39 @@ +"""Regression coverage for atomic request-admission FIFO entry.""" + +import pytest + +from deerflow.config.model_config import RequestAdmissionConfig +from deerflow.models import request_admission as admission + + +@pytest.fixture +def clock(monkeypatch): + now = [0.0] + monkeypatch.setattr(admission, "monotonic", lambda: now[0]) + return now + + +def test_blocking_waiter_reserves_fifo_position_before_newcomer_can_take_due_slot(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60)) + assert limiter.acquire(blocking=False) + + acquired, ticket = limiter._try_or_enqueue(blocking=True) + assert acquired is False + assert ticket is limiter._waiters[0] + + clock[0] = 1 + assert limiter.acquire(blocking=False) is False + + limiter._remove(ticket) + assert limiter.acquire(blocking=False) is True + + +def test_nonblocking_probe_never_joins_the_fifo(clock): + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60)) + assert limiter.acquire(blocking=False) + + acquired, ticket = limiter._try_or_enqueue(blocking=False) + + assert acquired is False + assert ticket is None + assert not limiter._waiters