diff --git a/docs/changelog/fragments/7778.md b/docs/changelog/fragments/7778.md new file mode 100644 index 0000000000..07ba46172b --- /dev/null +++ b/docs/changelog/fragments/7778.md @@ -0,0 +1 @@ +feat(resilience): atomically enforce cumulative global, provider, and account concurrency limits diff --git a/docs/plans/7778-hierarchical-admission.md b/docs/plans/7778-hierarchical-admission.md new file mode 100644 index 0000000000..c2f4444133 --- /dev/null +++ b/docs/plans/7778-hierarchical-admission.md @@ -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. diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 3c240717ad..d6b43b05b5 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -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"; @@ -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, @@ -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(); @@ -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, @@ -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", @@ -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; diff --git a/open-sse/services/accountSemaphore.ts b/open-sse/services/accountSemaphore.ts index ec0f06f090..4d0896ea30 100644 --- a/open-sse/services/accountSemaphore.ts +++ b/open-sse/services/accountSemaphore.ts @@ -1,20 +1,40 @@ /** - * Account Semaphore + * Hierarchical in-memory concurrency admission. * - * In-memory provider/account concurrency limiter keyed by provider and account. - * Requests beyond the configured concurrency cap wait in a FIFO queue until a slot opens, - * the gate is unblocked, or the queue timeout expires. + * `acquire()` preserves the account-semaphore API. `acquireMany()` admits one + * request only when every applicable global/provider/account gate has room. */ - export interface AccountSemaphoreKeyParts { provider: string; accountKey: string; } -interface QueuedAcquire { +export interface AcquireAccountSemaphoreOptions { + maxConcurrency?: number | null; + timeoutMs?: number; + signal?: AbortSignal | null; + maxQueueSize?: number; +} + +export interface SemaphoreRequirement { + key: string; + maxConcurrency?: number | null; +} + +export type AcquireManyOptions = Omit; + +interface AcquireRequest { + keys: string[]; resolve: (release: () => void) => void; reject: (error: Error) => void; - timer: ReturnType; + timer: ReturnType | null; + signal: AbortSignal | null; + abortListener: (() => void) | null; + settled: boolean; +} + +interface QueuedAcquire { + request: AcquireRequest; } interface AccountGate { @@ -25,13 +45,6 @@ interface AccountGate { cleanupTimer: ReturnType | null; } -export interface AcquireAccountSemaphoreOptions { - maxConcurrency?: number | null; - timeoutMs?: number; - signal?: AbortSignal | null; - maxQueueSize?: number; -} - export interface AccountSemaphoreStatsEntry { running: number; queued: number; @@ -41,12 +54,9 @@ export interface AccountSemaphoreStatsEntry { const DEFAULT_TIMEOUT_MS = 30_000; const DEFAULT_MAX_QUEUE_SIZE = 20; - const gates = new Map(); +const queuedRequests = new Set(); -/** - * Build the canonical account semaphore key. - */ export function buildAccountSemaphoreKey({ provider, accountKey, @@ -55,25 +65,23 @@ export function buildAccountSemaphoreKey({ } function isBypassed(maxConcurrency?: number | null): boolean { - return maxConcurrency == null || maxConcurrency <= 0; + return maxConcurrency == null || !Number.isFinite(maxConcurrency) || maxConcurrency <= 0; } function createNoopReleaseFn(): () => void { let released = false; - return () => { if (released) return; released = true; }; } -function ensureGate(semaphoreKey: string, maxConcurrency: number): AccountGate { - const existing = gates.get(semaphoreKey); +function ensureGate(key: string, maxConcurrency: number): AccountGate { + const existing = gates.get(key); if (existing) { existing.maxConcurrency = maxConcurrency; return existing; } - const created: AccountGate = { running: 0, maxConcurrency, @@ -81,7 +89,7 @@ function ensureGate(semaphoreKey: string, maxConcurrency: number): AccountGate { blockedUntil: null, cleanupTimer: null, }; - gates.set(semaphoreKey, created); + gates.set(key, created); return created; } @@ -100,91 +108,101 @@ function clearCleanupTimer(gate: AccountGate): void { gate.cleanupTimer = null; } -function cleanupGateIfIdle(semaphoreKey: string): void { - const gate = gates.get(semaphoreKey); - if (!gate) return; - if (gate.running > 0 || gate.queue.length > 0 || isBlocked(gate)) return; +function cleanupGateIfIdle(key: string): void { + const gate = gates.get(key); + if (!gate || gate.running > 0 || gate.queue.length > 0 || isBlocked(gate)) return; clearCleanupTimer(gate); - gates.delete(semaphoreKey); + gates.delete(key); } -function scheduleCleanup(semaphoreKey: string): void { - const gate = gates.get(semaphoreKey); +function scheduleCleanup(key: string): void { + const gate = gates.get(key); if (!gate) return; clearCleanupTimer(gate); - gate.cleanupTimer = setTimeout(() => { gate.cleanupTimer = null; - cleanupGateIfIdle(semaphoreKey); + cleanupGateIfIdle(key); }, 0); - gate.cleanupTimer.unref?.(); } -function drainQueue(semaphoreKey: string): void { - const gate = gates.get(semaphoreKey); - if (!gate) return; - - while (gate.queue.length > 0 && gate.running < gate.maxConcurrency && !isBlocked(gate)) { - const next = gate.queue.shift(); - if (!next) break; - clearTimeout(next.timer); - gate.running++; - next.resolve(createReleaseFn(semaphoreKey)); - } - - if (gate.running === 0 && gate.queue.length === 0) { - scheduleCleanup(semaphoreKey); - } -} - -function createReleaseFn(semaphoreKey: string): () => void { - let released = false; - - return () => { - if (released) return; - released = true; - - const gate = gates.get(semaphoreKey); - if (!gate) return; - if (gate.running > 0) { - gate.running--; - } - - if (gate.queue.length > 0) { - drainQueue(semaphoreKey); - return; - } - - scheduleCleanup(semaphoreKey); - }; -} - -function createSemaphoreTimeoutError( - semaphoreKey: string, - timeoutMs: number -): Error & { code: string } { - const error = new Error(`Semaphore timeout after ${timeoutMs}ms for ${semaphoreKey}`) as Error & { - code: string; - }; - error.code = "SEMAPHORE_TIMEOUT"; +function makeAbortError(signal: AbortSignal): Error { + if (signal.reason instanceof Error) return signal.reason; + const error = new Error( + typeof signal.reason === "string" ? signal.reason : "The operation was aborted" + ); + error.name = "AbortError"; return error; } -function makeAbortError(signal: AbortSignal): Error { - const reason = signal.reason; - if (reason instanceof Error) return reason; - const err = new Error(typeof reason === "string" ? reason : "The operation was aborted"); - err.name = "AbortError"; - return err; +function createSemaphoreError(code: string, message: string): Error & { code: string } { + const error = new Error(message) as Error & { code: string }; + error.code = code; + return error; +} + +function removeRequest(request: AcquireRequest): void { + queuedRequests.delete(request); + if (request.timer) clearTimeout(request.timer); + if (request.abortListener && request.signal) { + request.signal.removeEventListener("abort", request.abortListener); + } + for (const key of request.keys) { + const gate = gates.get(key); + if (!gate) continue; + const index = gate.queue.findIndex((queued) => queued.request === request); + if (index >= 0) gate.queue.splice(index, 1); + if (gate.running === 0 && gate.queue.length === 0) scheduleCleanup(key); + } +} + +function canAcquire(request: AcquireRequest): boolean { + return request.keys.every((key) => { + const gate = gates.get(key); + return ( + gate != null && + !isBlocked(gate) && + gate.running < gate.maxConcurrency && + gate.queue[0]?.request === request + ); + }); +} + +function createCompositeReleaseFn(keys: string[]): () => void { + let released = false; + return () => { + if (released) return; + released = true; + for (const key of keys) { + const gate = gates.get(key); + if (gate && gate.running > 0) gate.running--; + } + drainQueues(); + for (const key of keys) { + const gate = gates.get(key); + if (gate && gate.running === 0 && gate.queue.length === 0) scheduleCleanup(key); + } + }; +} + +function drainQueues(): void { + let progressed = true; + while (progressed) { + progressed = false; + for (const request of queuedRequests) { + if (request.settled || !canAcquire(request)) continue; + request.settled = true; + removeRequest(request); + for (const key of request.keys) gates.get(key)!.running++; + request.resolve(createCompositeReleaseFn(request.keys)); + progressed = true; + break; + } + } } -/** - * Acquire a slot for a provider/model/account tuple. - * Returns an idempotent release function that is safe to call in finally blocks. - */ export function acquire( - semaphoreKey: string, + key: string, { maxConcurrency = null, timeoutMs = DEFAULT_TIMEOUT_MS, @@ -192,146 +210,118 @@ export function acquire( maxQueueSize = DEFAULT_MAX_QUEUE_SIZE, }: AcquireAccountSemaphoreOptions = {} ): Promise<() => void> { - if (isBypassed(maxConcurrency)) { - return Promise.resolve(createNoopReleaseFn()); + return acquireMany([{ key, maxConcurrency }], { timeoutMs, signal, maxQueueSize }); +} + +/** + * Acquire all enabled requirements as one FIFO reservation. + * + * Waiting never increments any gate, preventing a saturated child gate from + * holding capacity in a parent gate. + */ +export function acquireMany( + requirements: SemaphoreRequirement[], + { + timeoutMs = DEFAULT_TIMEOUT_MS, + signal = null, + maxQueueSize = DEFAULT_MAX_QUEUE_SIZE, + }: AcquireManyOptions = {} +): Promise<() => void> { + const enabled = new Map(); + for (const requirement of requirements) { + if (isBypassed(requirement.maxConcurrency)) continue; + const limit = Math.trunc(requirement.maxConcurrency as number); + enabled.set(requirement.key, Math.min(enabled.get(requirement.key) ?? limit, limit)); + } + if (enabled.size === 0) return Promise.resolve(createNoopReleaseFn()); + if (signal?.aborted) return Promise.reject(makeAbortError(signal)); + + const keys = [...enabled.keys()].sort(); + for (const key of keys) { + const gate = ensureGate(key, enabled.get(key)!); + clearCleanupTimer(gate); + if (maxQueueSize > 0 && gate.queue.length >= maxQueueSize) { + return Promise.reject( + createSemaphoreError( + "SEMAPHORE_QUEUE_FULL", + `Semaphore queue full (${maxQueueSize}) for ${key}` + ) + ); + } } - if (signal?.aborted) { - return Promise.reject(makeAbortError(signal)); - } - - // isBypassed() above already excluded null/<=0 — ensureGate requires a plain - // number, but a boolean-returning helper isn't a type predicate TS can narrow on. - const gate = ensureGate(semaphoreKey, maxConcurrency as number); - clearCleanupTimer(gate); - - if (gate.running < gate.maxConcurrency && !isBlocked(gate)) { - gate.running++; - return Promise.resolve(createReleaseFn(semaphoreKey)); - } - - if (gate.queue.length >= maxQueueSize) { - const err = new Error(`Semaphore queue full (${maxQueueSize}) for ${semaphoreKey}`) as Error & { - code: string; - }; - err.code = "SEMAPHORE_QUEUE_FULL"; - return Promise.reject(err); + if ( + keys.every((key) => { + const gate = gates.get(key)!; + return gate.queue.length === 0 && gate.running < gate.maxConcurrency && !isBlocked(gate); + }) + ) { + for (const key of keys) gates.get(key)!.running++; + return Promise.resolve(createCompositeReleaseFn(keys)); } return new Promise((resolve, reject) => { - let abortListener: (() => void) | null = null; - - const cleanup = () => { - if (abortListener && signal) { - signal.removeEventListener("abort", abortListener); - } + const request: AcquireRequest = { + keys, + resolve, + reject, + timer: null, + signal, + abortListener: null, + settled: false, }; - - const timer = setTimeout(() => { - cleanup(); - const nextGate = gates.get(semaphoreKey); - if (!nextGate) { - reject(createSemaphoreTimeoutError(semaphoreKey, timeoutMs)); - return; - } - - const queueIndex = nextGate.queue.findIndex((item) => item.timer === timer); - if (queueIndex !== -1) { - nextGate.queue.splice(queueIndex, 1); - } - - if (nextGate.running === 0 && nextGate.queue.length === 0) { - scheduleCleanup(semaphoreKey); - } - - reject(createSemaphoreTimeoutError(semaphoreKey, timeoutMs)); + request.timer = setTimeout(() => { + if (request.settled) return; + request.settled = true; + removeRequest(request); + reject( + createSemaphoreError( + "SEMAPHORE_TIMEOUT", + `Semaphore timeout after ${timeoutMs}ms for ${keys.join(",")}` + ) + ); + drainQueues(); }, timeoutMs); - - timer.unref?.(); - - const queueItem: QueuedAcquire = { - resolve: (release) => { - cleanup(); - resolve(release); - }, - reject: (error) => { - cleanup(); - reject(error); - }, - timer, - }; - - gate.queue.push(queueItem); - + request.timer.unref?.(); if (signal) { - abortListener = () => { - cleanup(); - clearTimeout(timer); - - const nextGate = gates.get(semaphoreKey); - if (!nextGate) { - reject(makeAbortError(signal)); - return; - } - - const queueIndex = nextGate.queue.findIndex((item) => item.timer === timer); - if (queueIndex !== -1) { - nextGate.queue.splice(queueIndex, 1); - } - - if (nextGate.running === 0 && nextGate.queue.length === 0) { - scheduleCleanup(semaphoreKey); - } - + request.abortListener = () => { + if (request.settled) return; + request.settled = true; + removeRequest(request); reject(makeAbortError(signal)); + drainQueues(); }; - if (signal.aborted) { - abortListener(); - } else { - signal.addEventListener("abort", abortListener); - } + signal.addEventListener("abort", request.abortListener, { once: true }); } + queuedRequests.add(request); + for (const key of keys) gates.get(key)!.queue.push({ request }); + drainQueues(); }); } -/** - * Temporarily block new acquisitions for a key while allowing in-flight requests to finish. - */ -export function markBlocked(semaphoreKey: string, cooldownMs: number): void { - const safeCooldownMs = Number.isFinite(cooldownMs) && cooldownMs > 0 ? cooldownMs : 0; - if (safeCooldownMs <= 0) { - const gate = gates.get(semaphoreKey); - if (!gate) return; - gate.blockedUntil = null; - drainQueue(semaphoreKey); - return; - } - - const gate = gates.get(semaphoreKey) ?? ensureGate(semaphoreKey, 1); +export function markBlocked(key: string, until: Date | string | number): void { + const untilMs = + until instanceof Date + ? until.getTime() + : typeof until === "number" + ? Date.now() + Math.max(0, until) + : new Date(until).getTime(); + if (!Number.isFinite(untilMs) || untilMs <= Date.now()) return; + const gate = ensureGate(key, gates.get(key)?.maxConcurrency ?? 1); clearCleanupTimer(gate); - gate.blockedUntil = Date.now() + safeCooldownMs; + gate.blockedUntil = untilMs; +} - const timer = setTimeout(() => { - const nextGate = gates.get(semaphoreKey); - if (!nextGate) return; - if (nextGate.blockedUntil && Date.now() >= nextGate.blockedUntil) { - nextGate.blockedUntil = null; - drainQueue(semaphoreKey); - if (nextGate.running === 0 && nextGate.queue.length === 0) { - scheduleCleanup(semaphoreKey); - } - } - }, safeCooldownMs + 50); - - timer.unref?.(); +export function unblock(key: string): void { + const gate = gates.get(key); + if (!gate) return; + gate.blockedUntil = null; + drainQueues(); + cleanupGateIfIdle(key); } -/** - * Return the current in-memory semaphore snapshot. - */ export function getStats(): Record { const stats: Record = {}; - for (const [key, gate] of gates) { stats[key] = { running: gate.running, @@ -340,48 +330,50 @@ export function getStats(): Record { blockedUntil: gate.blockedUntil ? new Date(gate.blockedUntil).toISOString() : null, }; } - return stats; } -/** - * Check if an account semaphore key is currently at or over its max concurrency limit. - * Returns true if running >= maxConcurrency or blocked. - */ export function isAccountSemaphoreFull( provider: string, accountKey: string, maxConcurrency?: number | null ): boolean { if (isBypassed(maxConcurrency)) return false; - const key = buildAccountSemaphoreKey({ provider, accountKey }); - const gate = gates.get(key); + const gate = gates.get(buildAccountSemaphoreKey({ provider, accountKey })); if (!gate) return false; const effectiveCap = maxConcurrency ?? gate.maxConcurrency; - if (isBypassed(effectiveCap)) return false; - return gate.running >= effectiveCap || isBlocked(gate); + return !isBypassed(effectiveCap) && (gate.running >= effectiveCap || isBlocked(gate)); } -/** - * Reset a single key and reject queued waiters. - */ -export function reset(semaphoreKey: string): void { - const gate = gates.get(semaphoreKey); +export function reset(key: string): void { + const gate = gates.get(key); if (!gate) return; - clearCleanupTimer(gate); - for (const entry of gate.queue) { - clearTimeout(entry.timer); - entry.reject(new Error("Semaphore reset")); + const error = createSemaphoreError("SEMAPHORE_RESET", `Semaphore reset for ${key}`); + const rejections: AcquireRequest[] = []; + for (const queued of [...gate.queue]) { + const request = queued.request; + if (request.settled) continue; + request.settled = true; + removeRequest(request); + rejections.push(request); } - gates.delete(semaphoreKey); + gates.delete(key); + for (const request of rejections) request.reject(error); + drainQueues(); } -/** - * Reset all keys and reject queued waiters. - */ export function resetAll(): void { - for (const key of gates.keys()) { - reset(key); + const error = createSemaphoreError("SEMAPHORE_RESET", "Semaphore reset"); + const rejections: AcquireRequest[] = []; + for (const request of [...queuedRequests]) { + if (request.settled) continue; + request.settled = true; + removeRequest(request); + rejections.push(request); } + for (const gate of gates.values()) clearCleanupTimer(gate); + gates.clear(); + queuedRequests.clear(); + for (const request of rejections) request.reject(error); } diff --git a/src/app/(dashboard)/dashboard/settings/components/ResilienceTab.tsx b/src/app/(dashboard)/dashboard/settings/components/ResilienceTab.tsx index f2c8c7a114..723e3fd61f 100644 --- a/src/app/(dashboard)/dashboard/settings/components/ResilienceTab.tsx +++ b/src/app/(dashboard)/dashboard/settings/components/ResilienceTab.tsx @@ -13,6 +13,7 @@ type RequestQueueSettings = { requestsPerMinute: number; minTimeBetweenRequestsMs: number; concurrentRequests: number; + globalConcurrentRequests: number; maxWaitMs: number; }; @@ -240,13 +241,21 @@ function RequestQueueCard({ } /> setDraft((prev) => ({ ...prev, concurrentRequests })) } /> + + setDraft((prev) => ({ ...prev, globalConcurrentRequests })) + } + />
-
{t("resilienceConcurrentRequests")}
+
+ {t("resilienceConnectionScopeConcurrentRequests")} +
{value.concurrentRequests}
+
+
+ {t("resilienceGlobalConcurrentRequests")} +
+
+ {value.globalConcurrentRequests || t("statusDisabled")} +
+
{t("resilienceMaxQueueWait")}
diff --git a/src/i18n/messages/de.json b/src/i18n/messages/de.json index 97b4a017dc..3e35bf0160 100644 --- a/src/i18n/messages/de.json +++ b/src/i18n/messages/de.json @@ -7538,6 +7538,8 @@ "resilienceRequestsPerMinute": "Requests per minute", "resilienceMinTimeBetweenRequests": "Minimum time between requests", "resilienceConcurrentRequests": "Concurrent requests", + "resilienceConnectionScopeConcurrentRequests": "Gleichzeitige Anfragen pro Verbindung/Quota-Bereich", + "resilienceGlobalConcurrentRequests": "Globale gleichzeitige Upstream-Anfragen (0 = deaktiviert)", "resilienceMaxQueueWaitTime": "Maximale Wartezeit in der Warteschlange", "resilienceBaseCooldown": "Base cooldown", "resilienceUseUpstreamRetryHints": "Use upstream retry hints", diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index f3cab6d48c..b605d0b5fd 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -7550,6 +7550,8 @@ "resilienceRequestsPerMinute": "Requests per minute", "resilienceMinTimeBetweenRequests": "Minimum time between requests", "resilienceConcurrentRequests": "Concurrent requests", + "resilienceConnectionScopeConcurrentRequests": "Concurrent requests per connection/quota scope", + "resilienceGlobalConcurrentRequests": "Global concurrent upstream requests (0 = disabled)", "resilienceMaxQueueWaitTime": "Maximum queue wait time", "resilienceBaseCooldown": "Base cooldown", "resilienceUseUpstreamRetryHints": "Use upstream retry hints", diff --git a/src/lib/resilience/settings.ts b/src/lib/resilience/settings.ts index 3923c0fb44..f142248e70 100644 --- a/src/lib/resilience/settings.ts +++ b/src/lib/resilience/settings.ts @@ -58,6 +58,7 @@ export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = { requestsPerMinute: DEFAULT_API_LIMITS.requestsPerMinute, minTimeBetweenRequestsMs: DEFAULT_API_LIMITS.minTimeBetweenRequests, concurrentRequests: DEFAULT_API_LIMITS.concurrentRequests, + globalConcurrentRequests: 0, maxWaitMs: DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS, maxQueueDepth: DEFAULT_REQUEST_QUEUE_MAX_DEPTH, }, @@ -208,6 +209,8 @@ function buildLegacyFallback(settings: JsonRecord): ResilienceSettings { DEFAULT_RESILIENCE_SETTINGS.requestQueue.concurrentRequests, { min: 1, max: 10_000 } ), + globalConcurrentRequests: + DEFAULT_RESILIENCE_SETTINGS.requestQueue.globalConcurrentRequests, maxWaitMs: DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxWaitMs, maxQueueDepth: DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxQueueDepth, }, diff --git a/src/lib/resilience/settings/normalize.ts b/src/lib/resilience/settings/normalize.ts index 20cb87e380..65b8d3c213 100644 --- a/src/lib/resilience/settings/normalize.ts +++ b/src/lib/resilience/settings/normalize.ts @@ -124,6 +124,11 @@ export function normalizeRequestQueueSettings( min: 1, max: 10_000, }); + const globalConcurrentRequests = toInteger( + record.globalConcurrentRequests, + fallback.globalConcurrentRequests, + { min: 0, max: 100_000 } + ); const maxWaitMs = toInteger(record.maxWaitMs, fallback.maxWaitMs, { min: 1, max: 24 * 60 * 60 * 1000, @@ -141,6 +146,7 @@ export function normalizeRequestQueueSettings( requestsPerMinute, minTimeBetweenRequestsMs, concurrentRequests, + globalConcurrentRequests, maxWaitMs, maxQueueDepth, }; @@ -431,8 +437,16 @@ function normalizeProviderQuotaOverrideEntry(raw: unknown): ProviderQuotaOverrid const out: ProviderQuotaOverrideSettings = {}; const rpm = typeof record.rpm === "number" ? record.rpm : Number(record.rpm); if (Number.isFinite(rpm) && rpm > 0) out.rpm = Math.trunc(rpm); - const concurrency = typeof record.concurrency === "number" ? record.concurrency : Number(record.concurrency); + const concurrency = + typeof record.concurrency === "number" ? record.concurrency : Number(record.concurrency); if (Number.isFinite(concurrency) && concurrency > 0) out.concurrency = Math.trunc(concurrency); + const providerConcurrency = + typeof record.providerConcurrency === "number" + ? record.providerConcurrency + : Number(record.providerConcurrency); + if (Number.isFinite(providerConcurrency) && providerConcurrency >= 0) { + out.providerConcurrency = Math.trunc(providerConcurrency); + } return Object.keys(out).length > 0 ? out : null; } diff --git a/src/lib/resilience/settings/types.ts b/src/lib/resilience/settings/types.ts index 0bc16d48b2..84afc1dc5d 100644 --- a/src/lib/resilience/settings/types.ts +++ b/src/lib/resilience/settings/types.ts @@ -16,6 +16,8 @@ export interface RequestQueueSettings { requestsPerMinute: number; minTimeBetweenRequestsMs: number; concurrentRequests: number; + /** Whole-process upstream concurrency cap. Zero disables the global gate. */ + globalConcurrentRequests: number; /** * Legacy persisted key used as Bottleneck's post-dispatch execution * expiration. It does not bound time spent in Bottleneck's QUEUED state. @@ -169,6 +171,8 @@ export interface ProviderQuotaOverrideSettings { rpm?: number; /** Overrides the static per-connection concurrency cap. */ concurrency?: number; + /** Shared concurrency cap across every connection for this provider. */ + providerConcurrency?: number; } export interface StreamRecoverySettings { diff --git a/src/shared/validation/schemas/settings.ts b/src/shared/validation/schemas/settings.ts index 06ae7ebef6..467bfc1d2b 100644 --- a/src/shared/validation/schemas/settings.ts +++ b/src/shared/validation/schemas/settings.ts @@ -32,6 +32,7 @@ export const legacyResilienceDefaultsSchema = z requestsPerMinute: z.number().int().min(1).optional(), minTimeBetweenRequests: z.number().int().min(0).optional(), concurrentRequests: z.number().int().min(1).optional(), + globalConcurrentRequests: z.number().int().min(0).max(100_000).optional(), }) .strict(); @@ -170,6 +171,7 @@ export const updateResilienceSchema = z .object({ rpm: z.number().int().min(1).optional(), concurrency: z.number().int().min(1).optional(), + providerConcurrency: z.number().int().min(0).max(100_000).optional(), }) .strict() ) diff --git a/tests/e2e/resilience-plan-alignment.spec.ts b/tests/e2e/resilience-plan-alignment.spec.ts index 5744dbd239..1c7c573227 100644 --- a/tests/e2e/resilience-plan-alignment.spec.ts +++ b/tests/e2e/resilience-plan-alignment.spec.ts @@ -7,6 +7,7 @@ const resilienceSettings = { requestsPerMinute: 100, minTimeBetweenRequestsMs: 200, concurrentRequests: 10, + globalConcurrentRequests: 0, maxWaitMs: 120000, }, connectionCooldown: { diff --git a/tests/unit/accountSemaphore.test.ts b/tests/unit/accountSemaphore.test.ts index 0b8ece6c82..145fea317f 100644 --- a/tests/unit/accountSemaphore.test.ts +++ b/tests/unit/accountSemaphore.test.ts @@ -3,6 +3,7 @@ import { afterEach, describe, it } from "node:test"; import { acquire, + acquireMany, buildAccountSemaphoreKey, getStats, markBlocked, @@ -14,6 +15,108 @@ afterEach(() => { resetAll(); }); +describe("accountSemaphore acquireMany", () => { + it("atomically acquires every enabled gate and releases them once", async () => { + const release = await acquireMany([ + { key: "global", maxConcurrency: 2 }, + { key: "provider:codex", maxConcurrency: 1 }, + { key: "account:codex:one", maxConcurrency: 1 }, + { key: "disabled", maxConcurrency: 0 }, + ]); + + assert.deepEqual(getStats(), { + global: { running: 1, queued: 0, maxConcurrency: 2, blockedUntil: null }, + "provider:codex": { running: 1, queued: 0, maxConcurrency: 1, blockedUntil: null }, + "account:codex:one": { running: 1, queued: 0, maxConcurrency: 1, blockedUntil: null }, + }); + + release(); + release(); + await new Promise((resolve) => setTimeout(resolve, 10)); + assert.deepEqual(getStats(), {}); + }); + + it("queues one atomic request without partially reserving free gates", async () => { + const releaseProvider = await acquire("provider:codex", { maxConcurrency: 1 }); + const waiting = acquireMany( + [ + { key: "global", maxConcurrency: 1 }, + { key: "provider:codex", maxConcurrency: 1 }, + { key: "account:codex:two", maxConcurrency: 1 }, + ], + { timeoutMs: 200 } + ); + await new Promise((resolve) => setTimeout(resolve, 10)); + + assert.equal(getStats().global?.running ?? 0, 0); + assert.equal(getStats()["account:codex:two"]?.running ?? 0, 0); + assert.equal(getStats()["provider:codex"]?.queued, 1); + + releaseProvider(); + const release = await waiting; + assert.equal(getStats().global?.running, 1); + assert.equal(getStats()["provider:codex"]?.running, 1); + assert.equal(getStats()["account:codex:two"]?.running, 1); + release(); + }); + + it("removes an atomic waiter from every gate on abort", async () => { + const releaseGlobal = await acquire("global", { maxConcurrency: 1 }); + const controller = new AbortController(); + const waiting = acquireMany( + [ + { key: "global", maxConcurrency: 1 }, + { key: "provider:codex", maxConcurrency: 1 }, + ], + { signal: controller.signal, timeoutMs: 200 } + ); + await new Promise((resolve) => setTimeout(resolve, 10)); + controller.abort(); + await assert.rejects(waiting, { name: "AbortError" }); + assert.equal(getStats().global?.queued, 0); + assert.equal(getStats()["provider:codex"]?.queued ?? 0, 0); + releaseGlobal(); + }); + + it("times out an atomic waiter without leaking reservations", async () => { + const releaseGlobal = await acquire("global", { maxConcurrency: 1 }); + await assert.rejects( + acquireMany( + [ + { key: "global", maxConcurrency: 1 }, + { key: "provider:codex", maxConcurrency: 1 }, + ], + { timeoutMs: 10 } + ), + (error: Error & { code?: string }) => error.code === "SEMAPHORE_TIMEOUT" + ); + assert.equal(getStats().global?.queued, 0); + assert.equal(getStats()["provider:codex"]?.running ?? 0, 0); + releaseGlobal(); + }); + + it("rejects an atomic waiter when any required gate queue is full", async () => { + const releaseGlobal = await acquire("global", { maxConcurrency: 1 }); + const queued = acquire("global", { maxConcurrency: 1, maxQueueSize: 1, timeoutMs: 200 }); + await new Promise((resolve) => setTimeout(resolve, 10)); + + await assert.rejects( + acquireMany( + [ + { key: "global", maxConcurrency: 1 }, + { key: "provider:codex", maxConcurrency: 1 }, + ], + { maxQueueSize: 1, timeoutMs: 200 } + ), + (error: Error & { code?: string }) => error.code === "SEMAPHORE_QUEUE_FULL" + ); + assert.equal(getStats()["provider:codex"]?.queued ?? 0, 0); + + releaseGlobal(); + (await queued)(); + }); +}); + describe("accountSemaphore", async () => { it("queues requests beyond the account cap and drains on release", async () => { const key = buildAccountSemaphoreKey({ diff --git a/tests/unit/chatcore-hierarchical-admission.test.ts b/tests/unit/chatcore-hierarchical-admission.test.ts new file mode 100644 index 0000000000..5d8689b419 --- /dev/null +++ b/tests/unit/chatcore-hierarchical-admission.test.ts @@ -0,0 +1,37 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { test } from "node:test"; + +const source = readFileSync( + new URL("../../open-sse/handlers/chatCore.ts", import.meta.url), + "utf8" +); + +test("chatCore acquires cumulative gates immediately before withRateLimit", () => { + const acquire = source.indexOf("await acquireConcurrencyGates("); + const rateLimit = source.indexOf("await withRateLimit(", acquire); + assert.ok(acquire >= 0, "hierarchical admission must be present"); + assert.ok(rateLimit > acquire, "hierarchical admission must precede withRateLimit"); + + const admission = source.slice(acquire, rateLimit); + assert.match(admission, /key: "global"/); + assert.match(admission, /key: `provider:\$\{canonicalProviderKey\}`/); + assert.match(admission, /key: accountSemaphoreKey/); + assert.match(admission, /globalConcurrentRequests/); + assert.match(admission, /providerConcurrency/); + assert.match(admission, /maxWaitMs/); + assert.match(admission, /maxQueueDepth/); +}); + +test("each rotated account attempt acquires and releases a fresh composite slot", () => { + const attemptLoop = source.indexOf( + "while (attempts < maxAttempts || antigravityByopRotationPending)" + ); + const acquire = source.indexOf("await acquireConcurrencyGates(", attemptLoop); + const finallyRelease = source.indexOf("releaseAccountSemaphore();", acquire); + const retryContinue = source.indexOf("continue;", acquire); + + assert.ok(attemptLoop >= 0 && acquire > attemptLoop); + assert.ok(finallyRelease > acquire, "each attempt must release the composite slot"); + assert.ok(retryContinue > acquire, "rotation remains inside the per-attempt acquisition loop"); +}); diff --git a/tests/unit/resilience-settings-provider-quota-overrides.test.ts b/tests/unit/resilience-settings-provider-quota-overrides.test.ts index 374cdfd484..9124bf9b98 100644 --- a/tests/unit/resilience-settings-provider-quota-overrides.test.ts +++ b/tests/unit/resilience-settings-provider-quota-overrides.test.ts @@ -19,13 +19,26 @@ function cloneDefaults(): ResilienceSettings { test("updateResilienceSchema accepts providerQuotaOverrides entries", () => { const parsed = updateResilienceSchema.safeParse({ providerQuotaOverrides: { - minimax: { rpm: 30, concurrency: 4 }, + minimax: { rpm: 30, concurrency: 4, providerConcurrency: 8 }, nvidia: { rpm: 60 }, }, }); assert.equal(parsed.success, true, "valid override map should parse"); }); +test("provider concurrency accepts zero as disabled and survives normalization", () => { + const resolved = resolveResilienceSettings({ + resilienceSettings: { + providerQuotaOverrides: { + codex: { providerConcurrency: 3 }, + claude: { providerConcurrency: 0 }, + }, + }, + }); + assert.equal(resolved.providerQuotaOverrides.codex.providerConcurrency, 3); + assert.equal(resolved.providerQuotaOverrides.claude.providerConcurrency, 0); +}); + test("updateResilienceSchema allows a body containing only providerQuotaOverrides", () => { // The superRefine requires at least one field; a lone override map is a // legitimate update and must not trigger "Must provide resilience settings".