Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/changelog/fragments/7778.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
feat(resilience): atomically enforce cumulative global, provider, and account concurrency limits
23 changes: 23 additions & 0 deletions docs/plans/7778-hierarchical-admission.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
# #7778 hierarchical admission cleanup plan

1. Lock the existing single-key semaphore contract and the new atomic multi-key
contract with focused tests: no partial reservations, FIFO queueing, abort,
timeout, queue-full, idempotent release, stats, and cleanup.
2. Generalize the existing account semaphore in place. Keep `acquire()` as a
compatibility wrapper around `acquireMany()`; do not add a second scheduler
or a dependency.
3. Replace the account-only acquisition in `chatCore` with one cumulative
global/provider/account acquisition immediately before `withRateLimit`.
Reacquire the whole set whenever account rotation changes the connection,
and retain the release through streaming completion.
4. Extend the existing resilience settings pipeline (types, defaults,
normalization, schema, API response, UI, and translations) with the global
and provider caps. Relabel the old Bottleneck concurrency control as
connection/quota-scope concurrency so its real scope is explicit.
5. Run focused tests, lint, typecheck, static checks, and the full test suite;
document the behavioral change in the changelog.

Behavior intentionally preserved: zero/null concurrency bypasses a gate,
account-only callers keep using `acquire()`, blocked-account controls retain
their key format and API, and provider rate-limit queue behavior remains
unchanged.
37 changes: 29 additions & 8 deletions open-sse/handlers/chatCore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,7 @@ import {
} from "../services/rateLimitManager.ts";
import * as localLimiterErrors from "../services/rateLimitManager/errors.ts";
import {
acquire as acquireAccountSemaphore,
acquireMany as acquireConcurrencyGates,
markBlocked as markAccountSemaphoreBlocked,
} from "../services/accountSemaphore.ts";
import { lockModel, lockModelIfPerModelQuota } from "../services/accountFallback.ts";
Expand Down Expand Up @@ -447,6 +447,7 @@ import { extractFacts } from "@/lib/memory/extraction";
import { handleToolCallExecution } from "@/lib/skills/interception";
import { MEMORY_BUILTIN_TOOL_NAMES } from "@/lib/skills/memoryBuiltins";
import { OMNIROUTE_RESPONSE_HEADERS } from "@/shared/constants/headers";
import { resolveProviderId } from "@/shared/constants/providers";
import { getClaudeCodeCompatibleRequestDefaults } from "@/lib/providers/requestDefaults";
import {
buildClaudeCodeCompatibleRequest,
Expand Down Expand Up @@ -525,6 +526,7 @@ export async function handleChatCore({
managedLease = null,
}) {
let { provider, model, extendedContext } = modelInfo;
const resilienceSettings = resolveResilienceSettings(cachedSettings);
if (!skipResourcePressureGuard) {
try {
const pressureGuard = checkResourcePressureGuard();
Expand Down Expand Up @@ -3051,6 +3053,10 @@ export async function handleChatCore({
connectionId: attemptConnectionId,
credentials: execCreds,
});
const canonicalProviderKey = resolveProviderId(String(provider).trim().toLowerCase());
const providerConcurrency =
resilienceSettings.providerQuotaOverrides[canonicalProviderKey]
?.providerConcurrency ?? 0;

trace("pre_semaphore", {
semaphoreKey: accountSemaphoreKey,
Expand All @@ -3061,13 +3067,27 @@ export async function handleChatCore({
stage: "waiting_account_slot",
});
}
const releaseAccountSemaphore =
accountSemaphoreKey && accountSemaphoreMaxConcurrency != null
? await acquireAccountSemaphore(accountSemaphoreKey, {
maxConcurrency: accountSemaphoreMaxConcurrency,
signal: streamController.signal,
})
: () => {};
const releaseAccountSemaphore = await acquireConcurrencyGates(
[
{
key: "global",
maxConcurrency: resilienceSettings.requestQueue.globalConcurrentRequests,
},
{
key: `provider:${canonicalProviderKey}`,
maxConcurrency: providerConcurrency,
},
{
key: accountSemaphoreKey || "",
maxConcurrency: accountSemaphoreKey ? accountSemaphoreMaxConcurrency : null,
},
],
{
timeoutMs: resilienceSettings.requestQueue.maxWaitMs,
maxQueueSize: resilienceSettings.requestQueue.maxQueueDepth,
signal: streamController.signal,
}
);
trace("post_semaphore");
updatePendingScope(pendingScope, {
stage: "waiting_rate_limit",
Expand Down Expand Up @@ -3328,6 +3348,7 @@ export async function handleChatCore({
"ANTIGRAVITY_BYOP_ROTATION",
`BYOP 422 on connection ${String(byopFailedId).slice(0, 8)} → rotating to ${String(byopNextCreds.connectionId).slice(0, 8)}`
);
releaseAccountSemaphore();
Object.assign(credentials, byopNextCreds);
antigravityByopRotationPending = true;
continue;
Expand Down
Loading
Loading