feat(llm): per-provider bounded fail-open concurrency cap (Workstream C phase 3) - #347
Merged
Merged
Conversation
…t (Workstream C C3 wiring)
📝 WalkthroughWalkthroughLiteLLM text-generation and embedding requests now use a shared, environment-configured per-provider concurrency limiter with bounded acquisition, fail-open behavior, and permit cleanup. Tests cover provider isolation, unknown providers, logging, and integration seams. ChangesProvider concurrency limiting
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant TextGenerationOrEmbedding
participant provider_slot
participant LiteLLM
participant ProviderSemaphore
TextGenerationOrEmbedding->>provider_slot: provider_slot(model)
provider_slot->>LiteLLM: resolve provider
provider_slot->>ProviderSemaphore: acquire bounded slot
ProviderSemaphore-->>provider_slot: permit or fail-open timeout
provider_slot-->>TextGenerationOrEmbedding: execute request context
TextGenerationOrEmbedding->>LiteLLM: completion or embedding call
TextGenerationOrEmbedding->>ProviderSemaphore: release acquired slot
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
reflexio/server/llm/_litellm_text_generation.py (1)
794-797: 🧹 Nitpick | 🔵 TrivialFallback requests bypass the fallback provider's concurrency cap.
Because
provider_slotwraps the entire_completion_with_hard_timeoutcall, any fallback models executed internally bylitellm's fallback ladder will continue to hold the primary provider's concurrency slot while completely bypassing the fallback provider's concurrency limit.This aligns with the stated decision ("429 recovery stays with the fallback ladder (Decision A)"), but it means that during a sustained primary outage, the fallback provider could temporarily receive a burst of concurrent requests that exceeds its configured limit. Consider noting this behavior for operational monitoring and downstream load management.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/llm/_litellm_text_generation.py` around lines 794 - 797, Document the concurrency behavior around _call_and_parse and provider_slot: fallback models invoked internally by _completion_with_hard_timeout do not acquire the fallback provider’s concurrency cap while the primary slot remains held. Note this for operational monitoring and downstream load management, without changing the existing fallback-ladder behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/server/llm/test_provider_concurrency.py`:
- Around line 13-34: Update test_caps_concurrent_holders_per_provider to avoid
importlib.reload(pc) and environment mutation; use monkeypatch.setattr on the
module-level concurrency configuration, matching the approach in
test_fail_open_emits_log. Keep the existing timeout patch and semaphore
assertions, and remove both reload calls so subsequent tests retain their
original module state.
---
Nitpick comments:
In `@reflexio/server/llm/_litellm_text_generation.py`:
- Around line 794-797: Document the concurrency behavior around _call_and_parse
and provider_slot: fallback models invoked internally by
_completion_with_hard_timeout do not acquire the fallback provider’s concurrency
cap while the primary slot remains held. Note this for operational monitoring
and downstream load management, without changing the existing fallback-ladder
behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: ee28073a-c6f0-4274-bfee-2c98bcf2143d
📒 Files selected for processing (4)
reflexio/server/llm/_litellm_embedding.pyreflexio/server/llm/_litellm_text_generation.pyreflexio/server/llm/_provider_concurrency.pytests/server/llm/test_provider_concurrency.py
Comment on lines
+13
to
+34
| def test_caps_concurrent_holders_per_provider(monkeypatch): | ||
| # Force a small cap and a short acquire timeout via reload. | ||
| monkeypatch.setenv("REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY", "2") | ||
| import importlib | ||
|
|
||
| importlib.reload(pc) | ||
| monkeypatch.setattr(pc, "_ACQUIRE_TIMEOUT_SECONDS", 0.3) | ||
| # Force a deterministic provider key (avoid network/model lookups). | ||
| monkeypatch.setattr(pc, "_provider_key", lambda _m: "openai") | ||
|
|
||
| sem = pc._get_semaphore("openai") | ||
| assert sem._value == 2 # BoundedSemaphore initial permits | ||
| # Hold both permits. | ||
| with pc.provider_slot("gpt-x"), pc.provider_slot("gpt-x"): | ||
| assert sem._value == 0 | ||
| # A third acquire must FAIL OPEN after the bounded timeout (not block forever). | ||
| t0 = time.monotonic() | ||
| with pc.provider_slot("gpt-x"): | ||
| waited = time.monotonic() - t0 | ||
| assert waited >= 0.3 # waited the bounded timeout, then proceeded | ||
| importlib.reload(pc) | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Avoid module reloading to prevent test pollution.
Using importlib.reload(pc) inside a test re-evaluates module-level variables. Because the second reload(pc) at the end of the test executes before monkeypatch restores the environment variables, the module will remain initialized with REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY = 2 for any subsequent tests executed by the same worker.
To avoid test pollution, replace the environment variable and reload approach with monkeypatch.setattr, mirroring the safer strategy you used in test_fail_open_emits_log.
♻️ Proposed refactor
def test_caps_concurrent_holders_per_provider(monkeypatch):
- # Force a small cap and a short acquire timeout via reload.
- monkeypatch.setenv("REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY", "2")
- import importlib
-
- importlib.reload(pc)
+ # Force a small cap and a short acquire timeout.
+ monkeypatch.setattr(pc, "REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY", 2)
monkeypatch.setattr(pc, "_ACQUIRE_TIMEOUT_SECONDS", 0.3)
# Force a deterministic provider key (avoid network/model lookups).
monkeypatch.setattr(pc, "_provider_key", lambda _m: "openai")
+ pc._semaphores.clear()
sem = pc._get_semaphore("openai")
assert sem._value == 2 # BoundedSemaphore initial permits
# Hold both permits.
with pc.provider_slot("gpt-x"), pc.provider_slot("gpt-x"):
assert sem._value == 0
# A third acquire must FAIL OPEN after the bounded timeout (not block forever).
t0 = time.monotonic()
with pc.provider_slot("gpt-x"):
waited = time.monotonic() - t0
assert waited >= 0.3 # waited the bounded timeout, then proceeded
- importlib.reload(pc)
+ pc._semaphores.clear()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def test_caps_concurrent_holders_per_provider(monkeypatch): | |
| # Force a small cap and a short acquire timeout via reload. | |
| monkeypatch.setenv("REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY", "2") | |
| import importlib | |
| importlib.reload(pc) | |
| monkeypatch.setattr(pc, "_ACQUIRE_TIMEOUT_SECONDS", 0.3) | |
| # Force a deterministic provider key (avoid network/model lookups). | |
| monkeypatch.setattr(pc, "_provider_key", lambda _m: "openai") | |
| sem = pc._get_semaphore("openai") | |
| assert sem._value == 2 # BoundedSemaphore initial permits | |
| # Hold both permits. | |
| with pc.provider_slot("gpt-x"), pc.provider_slot("gpt-x"): | |
| assert sem._value == 0 | |
| # A third acquire must FAIL OPEN after the bounded timeout (not block forever). | |
| t0 = time.monotonic() | |
| with pc.provider_slot("gpt-x"): | |
| waited = time.monotonic() - t0 | |
| assert waited >= 0.3 # waited the bounded timeout, then proceeded | |
| importlib.reload(pc) | |
| def test_caps_concurrent_holders_per_provider(monkeypatch): | |
| # Force a small cap and a short acquire timeout. | |
| monkeypatch.setattr(pc, "REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY", 2) | |
| monkeypatch.setattr(pc, "_ACQUIRE_TIMEOUT_SECONDS", 0.3) | |
| # Force a deterministic provider key (avoid network/model lookups). | |
| monkeypatch.setattr(pc, "_provider_key", lambda _m: "openai") | |
| pc._semaphores.clear() | |
| sem = pc._get_semaphore("openai") | |
| assert sem._value == 2 # BoundedSemaphore initial permits | |
| # Hold both permits. | |
| with pc.provider_slot("gpt-x"), pc.provider_slot("gpt-x"): | |
| assert sem._value == 0 | |
| # A third acquire must FAIL OPEN after the bounded timeout (not block forever). | |
| t0 = time.monotonic() | |
| with pc.provider_slot("gpt-x"): | |
| waited = time.monotonic() - t0 | |
| assert waited >= 0.3 # waited the bounded timeout, then proceeded | |
| pc._semaphores.clear() |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/server/llm/test_provider_concurrency.py` around lines 13 - 34, Update
test_caps_concurrent_holders_per_provider to avoid importlib.reload(pc) and
environment mutation; use monkeypatch.setattr on the module-level concurrency
configuration, matching the approach in test_fail_open_emits_log. Keep the
existing timeout patch and semaphore assertions, and remove both reload calls so
subsequent tests retain their original module state.
This was referenced Jul 17, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Per-provider, per-instance bounded fail-open concurrency cap for remote LLM calls
(
REFLEXIO_LLM_PROVIDER_MAX_CONCURRENCY, default 8). Part of Scalability Workstream C (C3).Why
At 4–8 tasks, a publish burst fans N tasks' LLM calls at one provider with no local ceiling →
provider-429 storms. This caps concurrent in-flight calls per provider per instance.
Design (per the enterprise design doc §5, reviewed 2 rounds)
reflexio/server/llm/_provider_concurrency.py: a lazily-built per-providerBoundedSemaphore+provider_slot(model)context manager.self._completion_with_hard_timeout(...)in_call_and_parse— the subprocess spawns inside that call, so the permit is held in theparent (a child-side semaphore would cap nothing). Embedding wraps the in-process
litellm.embedding(...). Two distinct seams.acquire(timeout=30s); on timeout it proceeds WITHOUT apermit + logs
llm_provider_cap_saturated— never blocks unboundedly (that would re-open thehung-provider stall class PYTHON-FASTAPI-62 through the limiter).
num_retriesuntouched.embedders — bounded by their own knob.
Tests
per-provider isolation, unknown-provider uncapped, saturation logs.
provider_slot. Parent placement is structural (wrap around_completion_with_hard_timeout, outside the fork).560 passedacrosstests/server/llm/(4 unrelated errors need an LLM API key at fixture setup)..env.templatedoc + the deployment-doctrine 3-knob disambiguation land in the paired enterprise PR.Summary by CodeRabbit
New Features
Reliability
Tests