fix: never reset live concurrency counters (cap leaked on long streams)

[lloydmak99]

Problem

The per-organization, per-model concurrency cap (DEFAULT_CONCURRENT_LIMIT = 64, overridable via /v1/admin/organizations/{org_id}/concurrent-limit) did not reliably cap concurrency. An organization could run well above its limit on a single cloud-api instance.

Root cause. CompletionServiceImpl.concurrent_counts was a moka::future::Cache<(org_id, model_id), Arc<AtomicU32>> built with time_to_live(600s) and max_capacity(100_000). moka's TTL is measured from insertion and ignores whether requests still hold the counter. Ten minutes after a key was first used, the entry expired; the next try_acquire_concurrent_slot ran get_with and created a fresh AtomicU32(0), admitting up to limit more requests while the older requests were still streaming. When those older requests finished they decremented the orphaned Arc, so the new counter also undercounted for its whole life. Capacity eviction could do the same. The doc comment on CONCURRENT_COUNT_TTL_SECS already admitted the limit "can be temporarily exceeded"; the TTL was added in #447 as a backstop against leaked slots, before the RAII guards in the same PR made a leak impossible on the guarded paths.

Evidence (2026-09-14, OpenRouter org 0d16829b-…, model z-ai/glm-5.3-flash).

• 15,153 "Organization concurrent request limit exceeded for model" warnings, all from cpu01, all current_count=64 limit=64, one model_id. cloud-api.near.ai resolved only to cpu01 (40.160.1.150, re-verified today), so one instance and one counter should have held this traffic to 64.
• Concurrency rebuilt from inference-proxy terminal logs: a peak of 92 in flight at 19:19:28Z, of which about 65 were admitted within 19:18:46–19:18:57Z right after minutes of continuous 64/64 rejections while ~26 older requests were still running. That is exactly the signature of a counter reset: a burst of limit admissions on top of the live requests.
• A peak of 77 requests producing tokens at 18:53:16Z with a single rejection between 18:15Z and 19:13Z: the counter admitted requests 65–77, i.e. it was undercounting after an earlier reset.
• OpenRouter streams are long (served duration p90 ≈ 274 s, max ≈ 880 s), so counters were routinely live when the 600 s TTL fired.

Changes

Design: a locked registry of live holders (option a), no TTL, no capacity eviction.

New module crates/services/src/completions/concurrency.rs:

• ConcurrencySlots is a std::sync::Mutex<HashMap<(org_id, model_id), KeyState>> where each KeyState holds a map of slot_id → Holder { acquired_at, reported }. The in-flight count for a key is holders.len(), read and updated under the one lock, so acquire, release and key removal are atomic with respect to each other: two counters for the same key can never coexist, and a key with a live holder can never be removed because it is non-empty. No .await ever happens under the lock, and the lock is taken with unwrap_or_else(|e| e.into_inner()) so a poisoned mutex can never panic inside Drop.
• try_acquire(key, limit) -> Result<ConcurrencySlot, current_count> admits only when holders.len() < limit. A rejection inserts nothing, so probing idle keys does not grow the registry.
• ConcurrencySlot is a non-Clone RAII handle whose Drop removes its holder and drops the key when it becomes empty. Idle keys are therefore reclaimed on release, keeping memory bounded (requirement 3) without any sweep or TTL.
• sweep(now, threshold) -> SweepReport snapshots in-use total, tracked keys, the busiest key, and slots held longer than threshold, marking each long-held slot as reported so it is logged once.
• spawn_slot_monitor runs the sweep every 60 s on the Tokio runtime (skipped when there is none), holding only a Weak to the registry. Each tick records three histograms with the environment tag only (no per-org cardinality): cloud_api.concurrent_slots.in_use, cloud_api.concurrent_slots.max_per_key, cloud_api.concurrent_slots.over_threshold, and emits warn! (IDs only) for each slot held longer than 30 minutes: "Concurrent request slot held longer than the leak threshold". Real leaks now surface instead of being papered over.

Why not option b (keep moka with a custom Expiry): moka decides expiry at create/read/update time and cannot exempt entries from capacity eviction per entry, so "never evict while count > 0" would need re-insert tricks on every acquire and still leave a window. The registry is ~150 lines of plain code with obviously atomic semantics and one lock op per acquire and per release, which is negligible against a network hop per request.

crates/services/src/completions/mod.rs:

• Removed CONCURRENT_COUNT_TTL_SECS and the moka concurrent_counts cache; the service now holds concurrent_slots: Arc<ConcurrencySlots> and starts the monitor in new().
• try_acquire_concurrent_slot returns a ConcurrencySlot. The rejection warn log (message, current_count, limit, IDs), the 429 error text and record_error call are byte-identical to before, so existing alerts keep working.
• ConcurrentSlotGuard wraps Option<ConcurrencySlot> and keeps disarm(); its manual Drop is gone because dropping the slot releases it. InterceptStream.concurrent_counter became concurrent_slot: Option<ConcurrencySlot>, released first thing in its Drop at exactly the point the old decrement happened. All seven acquire sites keep their shape and early-return ordering, so release still happens exactly once on normal completion, on error, on client disconnect (stream dropped) and on panic, via RAII only. No manual decrements remain anywhere.

crates/services/src/completions/ports.rs: ConcurrentRequestGuard (used by the Anthropic /v1/messages route) now owns a ConcurrencySlot; its manual Drop is gone.

crates/services/src/metrics/consts.rs: the three metric names above.

Unchanged: ORG_LIMIT_CACHE_TTL_SECS = 300 and the org-limit cache, DEFAULT_CONCURRENT_LIMIT, invalidate_org_concurrent_limit, all limit semantics. The counter is still per instance.

Compatibility with #975 (fleet concurrency leases). #975 keeps the local counter as its ConcurrentSlot::Local / Shadowed { local } fallback and keeps ConcurrentSlotGuard + disarm(). On rebase, Local(Arc<AtomicU32>) becomes Local(ConcurrencySlot) and release_local becomes a plain drop; ConcurrentRequestGuard already moves to a slot field in that PR. The metric names here (cloud_api.concurrent_slots.*) deliberately do not collide with #975's cloud_api.concurrency.*.

Second-leak audit (requested). Every guarded path was traced and no early release or lost slot was found:

• Streaming chat: every early return between acquire and disarm() is covered by the armed guard, and the span from disarm() to the InterceptStream literal contains no .await, so it cannot be cancelled there. Provider retry/fallback happens entirely before the stream is returned (retry_with_fallback_caps); there is no mid-stream provider swap.
• /v1/chat/completions and /v1/completions move the InterceptStream into Body::from_stream unbuffered, so the slot lives exactly as long as the client body. /v1/responses takes one slot per agent turn and drops it promptly on client disconnect (send failure breaks the loop).
• Anthropic /v1/messages: the guard is moved into NativeUsageStream before the first await and released in finish_billing on EOF, error or drop; the non-streaming branch holds it until the body is fully read. count_tokens generates nothing and correctly takes no slot.
• Alias resolution (resolve_and_get_model) returns the canonical model.id, so aliases and canonical names share one counter. n > 1 is one slot, one provider call.
• audio_transcription wraps the provider future in tokio::time::timeout in place; nothing is spawned that could outlive the guard.

Gaps found that are not leaks and are left for a follow-up because they are policy changes (they would introduce 429s on paths that never had admission control): POST /v1/images/generations, POST /v1/images/edits, and the /v1/responses image branch call the provider pool directly and never take a slot; the auto-redact PII classification issued inside /v1/chat/completions (auto_redact::detect) also reaches the pool without a slot. Also pre-existing in /v1/responses: a client disconnect is only detected when a text delta fails to send; failures from reasoning or citation deltas are logged and ignored, so a request that is still emitting only reasoning keeps its slot until the provider stream ends. Correctly counted, just released late. Minor: the audio route keys the slot on the canonical model.id but sends request.model (possibly an alias) upstream; the count is right, only the outbound name differs.

Validation

Two independent reviews of the commit: an adversarial review of the registry and every guard site (no bug found; admission and release are atomic under one lock, no cancellation window between disarm() and the InterceptStream literal, no double release, registry growth bounded by live requests), and a system-level review confirming exactly one CompletionServiceImpl and therefore one registry per process (OHTTP loopback re-enters the same router), every capped route taking its slot, the streaming body owning the InterceptStream until the client goes away, and the admin PATCH invalidating the org-limit cache immediately.

Tests added (all Tokio paused-time where time matters):

• concurrency::tests::limit_holds_for_requests_held_past_600s_and_900s and tests::service_limit_holds_for_requests_held_past_600s_and_900s (through CompletionServiceImpl::try_acquire_concurrent_slot with a pinned org limit): fill the limit, advance 601 s and then past 1000 s, the next acquire is still rejected; release one, exactly one more is admitted.
• concurrency::tests::stress_never_exceeds_limit_and_ends_at_zero: 64 tasks × 50 iterations on 4 worker threads, limit 8; observed in-flight never exceeds 8, admitted + rejected == attempts, registry ends at 0 in-use and 0 keys.
• tests::dropping_intercept_stream_midway_releases_slot, tests::provider_error_before_first_chunk_releases_slot, tests::successful_stream_holds_slot_until_dropped (proves disarm() handed the slot to the stream rather than leaking it), tests::panic_in_guarded_scope_releases_slot, tests::panic_in_spawned_task_releases_slot, tests::test_intercept_stream_releases_slot_on_drop.
• tests::streaming_cap_holds_past_600s_through_create_chat_completion_stream: through the real public entry point with the mock provider, three never-polled streams fill a limit of 3; after 700 s and again after 1000 s of paused time the next request is rejected with RateLimitExceeded; dropping one stream admits exactly one more; dropping all leaves the registry empty.
• concurrency::tests::idle_key_is_reclaimed_and_restarts_from_zero_without_touching_other_keys, concurrency::tests::test_concurrent_limit_different_orgs_and_models_independent.
• concurrency::tests::sweep_reports_long_held_slots_once and concurrency::tests::slot_monitor_records_slot_histograms (metrics carry the environment tag only).

Commands run locally on the pinned toolchain (1.92.0):

cargo fmt --all -- --check                                   ok
cargo clippy --all-targets --all-features -- -D warnings     ok, no warnings
cargo nextest run --lib --bins                               1542 passed, 4 skipped
cargo nextest run --test integration_tests                   8 passed, 1 skipped
cargo nextest run --test e2e_all  (Postgres 15 in Docker)    761 passed, 11 skipped

Rollout Notes

• No config change needed; no migration; no new environment variables. Behaviour change is strictly "the configured limit is now enforced for requests of any length".
• Expect the OpenRouter org (and any other org that runs long streams near its cap) to see more 429s than before, because the cap is now real. That is the intended outcome; raising a limit is the existing admin PATCH.
• Confirm in production after deploy:
  • cloud_api.concurrent_slots.max_per_key ≤ the highest configured org limit (64 unless raised), and cloud_api.concurrent_slots.in_use ≤ that limit × number of (org, model) pairs active.
  • From inference-proxy logs, in-flight requests per org and model ≤ limit × number of cloud-api instances serving that org.
  • cloud_api.concurrent_slots.over_threshold stays at 0 and no "Concurrent request slot held longer than the leak threshold" warnings appear; if they do, the org/model IDs in the log point at the leaking path.
• Rollback: revert the commit. The registry has no persisted state.
• Out of scope, unchanged: retry/overload handling, 429 mapping, API-key rate limits, default limit values.

[ironloopai]

Review · Status

🟩 Completed

IronLoop completed the review and posted it to GitHub.

Result

Open submitted review →

Run details

• Run: ba241264-974f-484e-bc18-c6790508af3a
• Base: main at 5c7cab7
• Head: fix/never-reset-live-concurrency-counters at 685dc57
• Created: 2026-09-15 04:16 UTC
• Updated: 2026-09-15 04:40 UTC

_{Automatic trigger · attempt 1 of 3 · completed in 23m 43s}

[claude]

Review: fix: never reset live concurrency counters

Reviewed statically (this runner cannot invoke cargo, so I did not re-run the build or suite — the CI results stand on their own). No prior review threads existed on this PR, so nothing here duplicates earlier feedback.

No critical issues found. The core change is correct: holders.len() under a single Mutex makes admission, release, and key removal mutually atomic; a key with a live holder is non-empty and therefore unremovable; a rejection inserts nothing; and ConcurrencySlot is non-Clone with a Drop-only release. I traced the specific failure modes:

• Slot IDs come from one process-wide AtomicU64, so a stale release(key, id) can never remove another key's holder.
• No re-entrancy: try_acquire drops the guard before constructing the slot, and KeyState/Holder have no Drop, so keys.remove(&key) under the lock cannot call back into release.
• lock() via unwrap_or_else(|e| e.into_inner()) correctly keeps Drop panic-free during unwinding.
• u32::try_from(len).unwrap_or(u32::MAX) saturates into a rejection — fails closed.
• Per-instance state only, so rolling updates need no migration or coordination.
• CompletionServiceImpl::new is only called from init_domain_services_with_pool (crates/api/src/lib.rs:463), which is async, so Handle::try_current() always succeeds and the monitor genuinely starts in production.
• Release sites all hold up: InterceptStream::drop releases at exactly the old decrement point; the /v1/responses per-turn completion_stream is scoped to the loop body so turns do not stack slots; the Anthropic /v1/messages guard is released in finish_billing before its billing.take() early return.

Non-blocking notes, roughly in order of operational value:

1. timeout_seconds is now the only upper bound on a slot — worth confirming it is set for every backend serving a capped route.
Dropping the TTL removes the self-healing property, which is the right call, but it changes the failure mode: a genuinely stuck slot is now stuck until the pod restarts. There is no admin drain path, and 64 stuck slots hard-lock an org/model. The backstop is the provider request timeout. For openai_compatible — the OpenRouter path in the bug report — crates/inference_providers/src/non_attested/external/openai_compatible.rs:33-37 sets only connect_timeout/pool_idle_timeout, with no read_timeout; the bound comes solely from reqwest's total .timeout(timeout) at line 233. Chutes has a real streaming read_timeout. A pass over the configured timeout_seconds per backend before deploy seems worthwhile, since that value is now the worst-case hold.

2. Consider raising the affected org's limit before rolling this out, not after.
The evidence in the description has that org peaking at ~92 in flight against a limit of 64. The moment the cap becomes real, that is roughly a third fewer concurrent requests plus a step change in 429s. The rollout notes treat "raising a limit is the existing admin PATCH" as the remedy, but applying it reactively means the fix lands as a visible capacity cut first. Pre-raising makes the deploy a no-op for that org and lets you tighten deliberately afterwards.

3. sweep is O(total in-flight) while holding the one global mutex.
Every acquire and release process-wide contends on the same lock, and once a minute the sweep walks every holder under it. At current volumes this is genuinely negligible and not worth changing now, but if in-flight ever reaches tens of thousands it is the thing that will bite. The cheap fix later is to snapshot counts under the lock and do the long-held scan outside it, or shard the map.

4. Small caveat on the rollout check for max_per_key.
Because the registry enforces the cap atomically under one lock, max_per_key <= configured limit holds by construction — that check cannot fail even if something is wrong, so it validates less than it looks like it does. The load-bearing signals are over_threshold and the "held longer than the leak threshold" warn. Also, both gauges are sampled once per 60 s, so bursts between ticks are invisible in in_use.

Metric cardinality (environment tag only) and the logging are clean under CLAUDE.md — IDs only, no titles, content, or credentials. Leaving the uncapped paths (/v1/images/*, the /v1/responses image branch, auto_redact::detect) out of scope is the right call; they are policy changes, not leaks.

✅

[github-actions]

🔍 OpenCodeReview found 1 issue(s) in this PR.

• ✅ 1 posted as inline comment(s)
• 📝 0 posted as summary

⚠️ 1 warning(s) occurred during review.

[PierreLeGuen]

No confirmed issues found.

Checks: git diff --check 5c7cab7: passed.; cargo +1.92.0 fmt --all -- --check: passed.; cargo +1.92.0 test -p services --lib -- --test-threads=4: 669 passed, 0 failed, 1 ignored.

[ironloopai]

Review · Summary

Found one medium-severity resource-retention issue.

Findings: 🟠 Medium 1

_{Code-specific findings are attached to the diff.}

Validation

• ✅ Concurrency registry tests — Six focused registry tests passed, covering admission, release, contention, long-held slots, and monitoring.
• ✅ Long-stream service regression — The service-level long-lived streaming-cap regression test passed.

Review details

• Run: ba241264-974f-484e-bc18-c6790508af3a
• Attempts: 1