From 20358bc8bf51d80e894a4f104f3f87596b444406 Mon Sep 17 00:00:00 2001 From: Paijo <14921983+oyi77@users.noreply.github.com> Date: Fri, 1 May 2026 22:37:09 +0700 Subject: [PATCH] fix(combo): stabilize provider routing at 500+ connections (Issue #1846) (#1854) Integrated into release/v3.7.8 --- open-sse/config/constants.ts | 12 +++++------ open-sse/services/accountFallback.ts | 30 ++++++++++++++++++++++++--- open-sse/services/accountSemaphore.ts | 11 ++++++++++ open-sse/services/combo.ts | 9 +++++--- open-sse/services/quotaPreflight.ts | 2 +- 5 files changed, 51 insertions(+), 13 deletions(-) diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts index 7fb1d46392..462dcd481c 100644 --- a/open-sse/config/constants.ts +++ b/open-sse/config/constants.ts @@ -139,22 +139,22 @@ export const PROVIDER_PROFILES = { transientCooldown: 5000, // 5s (session tokens — short recovery) rateLimitCooldown: 60000, // 60s default when no retry-after header maxBackoffLevel: 8, // Higher ceiling (sessions may stay bad longer) - circuitBreakerThreshold: 3, // Opens fast (low limit providers) + circuitBreakerThreshold: 8, // Scaled for 500+ connections (was 3) circuitBreakerReset: 60000, // 1min reset // Provider-level circuit breaker (entire provider cooldown after repeated failures) - providerFailureThreshold: 3, // 3 transient failures trigger provider cooldown - providerFailureWindowMs: 600000, // 10min window for counting failures + providerFailureThreshold: 10, // Scaled for 500+ connections (was 3) + providerFailureWindowMs: 900000, // 15min window (was 10min) providerCooldownMs: 300000, // 5min cooldown when threshold reached }, apikey: { transientCooldown: 3000, // 3s (API providers recover faster) rateLimitCooldown: 0, // 0 = respect retry-after header from provider maxBackoffLevel: 5, // Lower ceiling (API quotas reset at known intervals) - circuitBreakerThreshold: 5, // More tolerant (occasional 502 is normal) + circuitBreakerThreshold: 12, // Scaled for 500+ connections (was 5) circuitBreakerReset: 30000, // 30s reset // Provider-level circuit breaker (entire provider cooldown after repeated failures) - providerFailureThreshold: 5, // 5 transient failures trigger provider cooldown - providerFailureWindowMs: 1200000, // 20min window for counting failures + providerFailureThreshold: 15, // Scaled for 500+ connections (was 5) + providerFailureWindowMs: 1800000, // 30min window (was 20min) providerCooldownMs: 600000, // 10min cooldown when threshold reached }, // Local providers (localhost inference backends like Ollama, LM Studio, oMLX). diff --git a/open-sse/services/accountFallback.ts b/open-sse/services/accountFallback.ts index 62b32c0d30..6b75933e44 100644 --- a/open-sse/services/accountFallback.ts +++ b/open-sse/services/accountFallback.ts @@ -56,7 +56,16 @@ type ModelFailureState = { // Provider-level failure tracking for circuit breaker behavior // Error codes that count toward provider-level failure threshold -const PROVIDER_FAILURE_ERROR_CODES = new Set([408, 429, 500, 502, 503, 504]); +// 429 (rate limit) is intentionally excluded: rate limits are connection-scoped +// and handled via Connection Cooldown, not provider-wide circuit breaker. +// Counting 429 toward provider failure causes cascading provider trips at scale +// when many connections hit rate limits simultaneously (Issue #1846). +const PROVIDER_FAILURE_ERROR_CODES = new Set([408, 500, 502, 503, 504]); + +// Per-connection failure deduplication: prevents rapid-fire failures from the +// same connection from counting multiple times toward the provider breaker. +const CONNECTION_FAILURE_DEDUP_MS = 5000; +const lastConnectionFailure = new Map(); // T06 (sub2api PR #1037): Signals that indicate permanent account deactivation. // When a 401 body contains these strings, the account is permanently dead @@ -541,14 +550,29 @@ export function getProviderCooldownRemainingMs(provider: string | null | undefin */ export function recordProviderFailure( provider: string | null | undefined, - log?: { warn?: (...args: unknown[]) => void } + log?: { warn?: (...args: unknown[]) => void }, + connectionId?: string | null ): void { if (!provider) return; + // Deduplicate rapid-fire failures from the same connection + if (connectionId) { + const dedupKey = `${provider}:${connectionId}`; + const now = Date.now(); + const lastFailure = lastConnectionFailure.get(dedupKey); + if (lastFailure && now - lastFailure < CONNECTION_FAILURE_DEDUP_MS) { + return; + } + // Prevent memory leak by clearing map if it grows too large + if (lastConnectionFailure.size > 10000) { + lastConnectionFailure.clear(); + } + lastConnectionFailure.set(dedupKey, now); + } + const breaker = getProviderBreaker(provider); if (!breaker) return; - // Skip if already in cooldown to prevent timer reset (indefinite lockout bug) if (!breaker.canExecute()) return; breaker._onFailure(); diff --git a/open-sse/services/accountSemaphore.ts b/open-sse/services/accountSemaphore.ts index d210dedf06..affb06a41f 100644 --- a/open-sse/services/accountSemaphore.ts +++ b/open-sse/services/accountSemaphore.ts @@ -29,6 +29,7 @@ export interface AcquireAccountSemaphoreOptions { maxConcurrency?: number | null; timeoutMs?: number; signal?: AbortSignal | null; + maxQueueSize?: number; } export interface AccountSemaphoreStatsEntry { @@ -39,6 +40,7 @@ export interface AccountSemaphoreStatsEntry { } const DEFAULT_TIMEOUT_MS = 30_000; +const DEFAULT_MAX_QUEUE_SIZE = 20; const gates = new Map(); @@ -187,6 +189,7 @@ export function acquire( maxConcurrency = null, timeoutMs = DEFAULT_TIMEOUT_MS, signal = null, + maxQueueSize = DEFAULT_MAX_QUEUE_SIZE, }: AcquireAccountSemaphoreOptions = {} ): Promise<() => void> { if (isBypassed(maxConcurrency)) { @@ -205,6 +208,14 @@ export function acquire( 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); + } + return new Promise((resolve, reject) => { let abortListener: (() => void) | null = null; diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 05830f16ac..93a84f4741 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -1678,7 +1678,7 @@ export async function handleComboChat({ // Trigger shared provider circuit breaker for 5xx errors and connection failures if (isProviderFailureCode(result.status)) { - recordProviderFailure(provider, log); + recordProviderFailure(provider, log, target.connectionId); } // Check if this is a transient error worth retrying on same model @@ -1841,8 +1841,11 @@ async function handleRoundRobinCombo({ timeoutMs: queueTimeout, }); } catch (err) { - if (err.code === "SEMAPHORE_TIMEOUT") { - log.warn("COMBO-RR", `Semaphore timeout for ${modelStr}, trying next model`); + if (err.code === "SEMAPHORE_TIMEOUT" || err.code === "SEMAPHORE_QUEUE_FULL") { + log.warn( + "COMBO-RR", + `Semaphore ${err.code === "SEMAPHORE_QUEUE_FULL" ? "queue full" : "timeout"} for ${modelStr}, trying next model` + ); if (offset > 0) fallbackCount++; continue; } diff --git a/open-sse/services/quotaPreflight.ts b/open-sse/services/quotaPreflight.ts index 1d1386f4bf..f600fa36a7 100644 --- a/open-sse/services/quotaPreflight.ts +++ b/open-sse/services/quotaPreflight.ts @@ -26,7 +26,7 @@ export type QuotaFetcher = ( connection?: Record ) => Promise; -const EXHAUSTION_THRESHOLD = 0.95; +const EXHAUSTION_THRESHOLD = 0.98; const WARN_THRESHOLD = 0.8; const quotaFetcherRegistry = new Map();