From ee7e837d1fef63e4dea70c8a364d97479564e9e2 Mon Sep 17 00:00:00 2001 From: Zyy <0x@ainz.cc> Date: Fri, 6 Nov 2026 23:09:29 +0800 Subject: [PATCH] feat(scheduling): split queue timeouts and allow account fallback (#625) Co-authored-by: Zyy --- .../gateway-admin/src/model/settings.rs | 7 +- .../tests/use_case/accounts/mod.rs | 4 +- .../tests/use_case/freeze_recovery.rs | 4 +- .../tests/use_case/observability/mod.rs | 4 +- .../gateway-admin/tests/use_case/settings.rs | 4 +- .../crates/gateway-api/src/admin/settings.rs | 15 +- backend/crates/gateway-api/tests/admin/mod.rs | 8 +- .../gateway-api/tests/admin/settings.rs | 12 +- .../crates/gateway-api/tests/openai/models.rs | 4 +- .../gateway-core/src/account/selection.rs | 43 ++++++ .../crates/gateway-core/src/concurrency.rs | 2 +- .../gateway-core/src/engine/coordinator.rs | 6 - .../src/engine/execution/scope.rs | 2 - .../src/engine/execution/service.rs | 10 +- backend/crates/gateway-core/src/engine/mod.rs | 2 +- .../gateway-core/src/settings/compiled.rs | 12 +- .../gateway-core/src/settings/values.rs | 27 +++- .../crates/gateway-core/tests/concurrency.rs | 11 +- .../tests/engine/execution/mod.rs | 23 +-- .../sdk/src/call/services/settings.rs | 12 +- .../crates/gateway-store/src/admin_adapter.rs | 4 +- .../src/postgres/runtime_settings.rs | 20 ++- .../gateway-store/src/postgres/snapshot.rs | 10 +- .../tests/postgres/runtime_settings.rs | 30 ++-- .../openai/src/credential/selector.rs | 60 ++++++-- .../contract/scheduling_projection.rs | 135 ++++++++++++++++++ .../tests/provider/contract/capacity.rs | 10 +- .../providers/xai/src/credential/selector.rs | 35 ++++- .../xai/tests/credential/selector.rs | 44 ++++++ backend/migrations/.frozen-sha256 | 1 + backend/migrations/0039_queue_timeouts.sql | 10 ++ docs/api.md | 15 +- docs/architecture.md | 4 +- frontend/src/api/modules/settings/index.ts | 4 +- frontend/src/i18n/locales/en/settings.json | 23 ++- .../src/i18n/locales/zh-Hans/settings.json | 23 ++- .../src/i18n/locales/zh-Hant/settings.json | 23 ++- .../components/ConcurrencyPolicyCard.vue | 85 +++++++---- .../settings/components/TokenRefreshCard.vue | 5 +- .../AccountErrorProtectionCard.vue | 4 +- .../settings/composables/useSettingsForm.ts | 22 ++- .../composables/useSettingsFormRules.ts | 3 +- frontend/src/views/settings/index.vue | 7 +- 43 files changed, 622 insertions(+), 167 deletions(-) create mode 100644 backend/migrations/0039_queue_timeouts.sql diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index a86629a0b..257111015 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -53,7 +53,9 @@ pub struct RuntimeSettingsValues { pub max_concurrent_per_account: u32, pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, - pub concurrency_wait_timeout_seconds: u32, + pub key_queue_timeout_seconds: u32, + pub account_queue_timeout_seconds: u32, + pub switch_account_on_queue_failure: bool, pub max_retry_after_seconds: u32, pub openai_guardian_reserved_concurrency: u32, pub openai_account_affinity: gateway_core::account::AccountAffinity, @@ -201,7 +203,8 @@ impl RuntimeSettingsValues { gateway_core::settings::validate_request_limits( self.max_waiting_per_key, self.max_waiting_per_account, - self.concurrency_wait_timeout_seconds, + self.key_queue_timeout_seconds, + self.account_queue_timeout_seconds, self.max_account_rotations, self.openai_session_affinity_ttl_hours, )?; diff --git a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs index 8bd27b700..aeb538cfd 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs @@ -1166,7 +1166,9 @@ impl SettingsStore for StaticSettingsStore { max_concurrent_per_account: 1, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, diff --git a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs index a5d7a46d4..7565435f3 100644 --- a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs +++ b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs @@ -41,7 +41,9 @@ fn runtime_settings(enabled: bool, probe_enabled: bool) -> RuntimeSettings { max_concurrent_per_account: 5, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, diff --git a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs index 349803cd4..5e59e2561 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs @@ -1420,7 +1420,9 @@ impl SettingsStore for FixtureSettingsStore { max_concurrent_per_account: self.max_concurrent_per_account, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, diff --git a/backend/crates/gateway-admin/tests/use_case/settings.rs b/backend/crates/gateway-admin/tests/use_case/settings.rs index c25ccc53f..a269f6886 100644 --- a/backend/crates/gateway-admin/tests/use_case/settings.rs +++ b/backend/crates/gateway-admin/tests/use_case/settings.rs @@ -333,7 +333,9 @@ async fn settings_should_reject_zero_refresh_margin_before_store_call() { max_concurrent_per_account: 1, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index 8cbb71109..32515b2a6 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -44,7 +44,9 @@ pub struct RuntimeSettingsFields { pub max_concurrent_per_account: u64, pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, - pub concurrency_wait_timeout_seconds: u32, + pub key_queue_timeout_seconds: u32, + pub account_queue_timeout_seconds: u32, + pub switch_account_on_queue_failure: bool, pub max_retry_after_seconds: u32, pub openai_guardian_reserved_concurrency: u32, pub openai_account_affinity: gateway_core::account::AccountAffinity, @@ -751,7 +753,9 @@ impl RuntimeSettingsFields { .map_err(|_| WireValidationError::new("settingsMaxConcurrencyOverflow"))?, max_waiting_per_key: self.max_waiting_per_key, max_waiting_per_account: self.max_waiting_per_account, - concurrency_wait_timeout_seconds: self.concurrency_wait_timeout_seconds, + key_queue_timeout_seconds: self.key_queue_timeout_seconds, + account_queue_timeout_seconds: self.account_queue_timeout_seconds, + switch_account_on_queue_failure: self.switch_account_on_queue_failure, max_retry_after_seconds: self.max_retry_after_seconds, openai_guardian_reserved_concurrency: self.openai_guardian_reserved_concurrency, openai_account_affinity: self.openai_account_affinity, @@ -791,7 +795,9 @@ impl From for RuntimeSett max_concurrent_per_account: u64::from(settings.max_concurrent_per_account), max_waiting_per_key: settings.max_waiting_per_key, max_waiting_per_account: settings.max_waiting_per_account, - concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, + key_queue_timeout_seconds: settings.key_queue_timeout_seconds, + account_queue_timeout_seconds: settings.account_queue_timeout_seconds, + switch_account_on_queue_failure: settings.switch_account_on_queue_failure, max_retry_after_seconds: settings.max_retry_after_seconds, openai_guardian_reserved_concurrency: settings.openai_guardian_reserved_concurrency, openai_account_affinity: settings.openai_account_affinity, @@ -824,7 +830,8 @@ fn settings_validation_error(field: &'static str) -> WireValidationError { "reset_credit_max_weekly_remaining_percent" => "resetCreditMaxWeeklyRemainingPercent", "account_warmup_schedule_time" => "accountWarmupScheduleTime", "audit_retention_days" => "auditRetentionDays", - "concurrency_wait_timeout_seconds" => "concurrencyWaitTimeoutSeconds", + "key_queue_timeout_seconds" => "keyQueueTimeoutSeconds", + "account_queue_timeout_seconds" => "accountQueueTimeoutSeconds", "max_account_rotations" => "maxAccountRotations", "max_concurrent_per_account" => "maxConcurrentPerAccount", "max_waiting_per_account" => "maxWaitingPerAccount", diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 7a777ac6a..c2a5dbebb 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -650,7 +650,9 @@ impl SettingsStore for MemorySettingsStore { max_concurrent_per_account: command.values.max_concurrent_per_account, max_waiting_per_key: command.values.max_waiting_per_key, max_waiting_per_account: command.values.max_waiting_per_account, - concurrency_wait_timeout_seconds: command.values.concurrency_wait_timeout_seconds, + key_queue_timeout_seconds: command.values.key_queue_timeout_seconds, + account_queue_timeout_seconds: command.values.account_queue_timeout_seconds, + switch_account_on_queue_failure: command.values.switch_account_on_queue_failure, max_retry_after_seconds: command.values.max_retry_after_seconds, openai_guardian_reserved_concurrency: command .values @@ -1895,7 +1897,9 @@ fn test_runtime_settings() -> RuntimeSettings { max_concurrent_per_account: 3, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index dd101b184..49b526dfa 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -56,7 +56,9 @@ fn update_body() -> Value { "openaiAccountAffinity": "relaxed", "maxAccountRotations": 3, "openaiSessionAffinityTtlHours": 24, - "concurrencyWaitTimeoutSeconds": 30, + "keyQueueTimeoutSeconds": 30, + "accountQueueTimeoutSeconds": 30, + "switchAccountOnQueueFailure": false, "maxRetryAfterSeconds": 240, "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", @@ -659,8 +661,10 @@ fn concurrency_queue_fields_validate_bounds_and_accept_disabled_queues() { for (field, invalid) in [ ("maxWaitingPerKey", 1001), ("maxWaitingPerAccount", 1001), - ("concurrencyWaitTimeoutSeconds", 0), - ("concurrencyWaitTimeoutSeconds", 121), + ("keyQueueTimeoutSeconds", 0), + ("accountQueueTimeoutSeconds", 0), + ("keyQueueTimeoutSeconds", 121), + ("accountQueueTimeoutSeconds", 121), ] { let mut body = update_body(); body[field] = json!(invalid); @@ -670,7 +674,7 @@ fn concurrency_queue_fields_validate_bounds_and_accept_disabled_queues() { let mut body = update_body(); body["maxWaitingPerKey"] = json!(0); body["maxWaitingPerAccount"] = json!(1000); - body["concurrencyWaitTimeoutSeconds"] = json!(120); + body["accountQueueTimeoutSeconds"] = json!(120); serde_json::from_value::(body) .unwrap() .validate() diff --git a/backend/crates/gateway-api/tests/openai/models.rs b/backend/crates/gateway-api/tests/openai/models.rs index 183be4e02..cdf1731c6 100644 --- a/backend/crates/gateway-api/tests/openai/models.rs +++ b/backend/crates/gateway-api/tests/openai/models.rs @@ -208,7 +208,9 @@ async fn request_settings_reach_query_terminal_without_mutating_authenticated_ba settings: Some(|settings| { let mut runtime = serde_json::to_value(&settings.runtime).unwrap(); runtime["model_mappings"] = serde_json::json!({"model-a":"plugin-model"}); - runtime["concurrency_wait_timeout_seconds"] = serde_json::json!(30); + runtime["key_queue_timeout_seconds"] = serde_json::json!(30); + runtime["account_queue_timeout_seconds"] = serde_json::json!(30); + runtime["switch_account_on_queue_failure"] = serde_json::json!(false); settings.runtime = serde_json::from_value(runtime).unwrap(); settings.fast_mode = FastMode::Disabled; }), diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 318dacce8..32fc34b12 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -6,6 +6,8 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Mutex, RwLock}; use std::time::{Duration, Instant, SystemTime}; +use futures::FutureExt; + use crate::concurrency::{CapacityWait, ConcurrencyQueuePolicy, QueueRejection}; use crate::identity::ProviderKind; @@ -92,6 +94,7 @@ pub struct AccountSelectionPolicy { queue_policy: ConcurrencyQueuePolicy, openai_guardian_reserved_concurrency: u32, openai_account_affinity: AccountAffinity, + switch_account_on_queue_failure: bool, max_account_rotations: u32, openai_session_affinity_ttl: Duration, } @@ -108,6 +111,7 @@ impl AccountSelectionPolicy { max_concurrent_per_account: max_concurrent_per_account.into(), openai_guardian_reserved_concurrency: 0, openai_account_affinity: AccountAffinity::default(), + switch_account_on_queue_failure: false, max_account_rotations: 3, openai_session_affinity_ttl: Duration::from_secs(24 * 3600), queue_policy: ConcurrencyQueuePolicy { @@ -128,6 +132,45 @@ impl AccountSelectionPolicy { self.openai_account_affinity } + #[must_use] + pub const fn with_switch_account_on_queue_failure(mut self, enabled: bool) -> Self { + self.switch_account_on_queue_failure = enabled; + self + } + + /// 原队位由首次选号释放。仅对允许换号的请求调用，重选不排队，也不延长请求总期限 + pub async fn reselect_after_queue_rejection< + T, + E: From, + F: Future>, + >( + self, + rejection: QueueRejection, + remaining: Option, + select: impl FnOnce(ConcurrencyQueuePolicy) -> F, + no_account_available: impl FnOnce(&E) -> bool, + ) -> Result { + if !self.switch_account_on_queue_failure || remaining.is_some_and(|time| time.is_zero()) { + return Err(rejection.into()); + } + let queue = ConcurrencyQueuePolicy { + max_waiting: 0, + ..self.queue_policy + }; + let selection = select(queue); + let result = match remaining { + Some(limit) => futures::select_biased! { + () = futures_timer::Delay::new(limit).fuse() => return Err(rejection.into()), + result = selection.fuse() => result, + }, + None => selection.await, + }; + match result { + Err(error) if no_account_available(&error) => Err(rejection.into()), + result => result, + } + } + #[must_use] pub const fn with_max_account_rotations(mut self, rotations: u32) -> Self { self.max_account_rotations = rotations; diff --git a/backend/crates/gateway-core/src/concurrency.rs b/backend/crates/gateway-core/src/concurrency.rs index ec879289b..8e8e09204 100644 --- a/backend/crates/gateway-core/src/concurrency.rs +++ b/backend/crates/gateway-core/src/concurrency.rs @@ -19,7 +19,7 @@ pub struct ConcurrencyQueuePolicy { pub timeout: Duration, } -/// 从首次入队开始计时，密钥、账号与后续重试共享同一个截止时刻 +/// 从首次入队开始计时，同一层的重新排队与重试共享截止时刻 #[derive(Debug, Clone, Default)] pub struct ConcurrencyWaitBudget { deadline: Arc>, diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 544cc31ad..6e05f7122 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -600,12 +600,6 @@ impl ResponseExecutionSession where S: ExecutionStore + 'static, { - /// 首次驱动前接入密钥准入使用的预算，所有账号尝试随后共享它 - pub(crate) fn with_concurrency_wait_budget(mut self, budget: ConcurrencyWaitBudget) -> Self { - self.concurrency_wait_budget = budget; - self - } - /// 传输首字节事实可在 next_event 等待期间由协议层提交 pub fn response_commit(&self) -> super::execution::ResponseCommit { self.response_commit.clone() diff --git a/backend/crates/gateway-core/src/engine/execution/scope.rs b/backend/crates/gateway-core/src/engine/execution/scope.rs index e8113e7bd..18fd22d93 100644 --- a/backend/crates/gateway-core/src/engine/execution/scope.rs +++ b/backend/crates/gateway-core/src/engine/execution/scope.rs @@ -5,7 +5,6 @@ use super::contract::{AuthenticatedClient, ExecutionRequestMetadata}; use super::{MAX_CONCURRENT_NESTED_EXECUTIONS, MAX_NESTED_EXECUTIONS}; use crate::engine::budget::{ClientBudgetCharge, ClientBudgetPort}; use crate::{ - concurrency::ConcurrencyWaitBudget, engine::{ ModelRequestId, ProviderAccountId, coordinator::CoordinationExtensions, extensions::ExtensionCallScope, nested::ExecutionEffects, @@ -75,7 +74,6 @@ pub(super) struct AuthorizedExecution { pub(super) struct ExecutionStartGuard { pub(super) admission: ExecutionAdmission, pub(super) active_request: Option, - pub(super) concurrency_wait_budget: ConcurrencyWaitBudget, pub(super) admission_decision_ms: Option, } diff --git a/backend/crates/gateway-core/src/engine/execution/service.rs b/backend/crates/gateway-core/src/engine/execution/service.rs index 042bb861c..58fa3fa25 100644 --- a/backend/crates/gateway-core/src/engine/execution/service.rs +++ b/backend/crates/gateway-core/src/engine/execution/service.rs @@ -724,7 +724,6 @@ impl DefaultExecutionService { request_id: &ModelRequestId, authorization: &mut AuthorizedExecution, ) -> Result { - let concurrency_wait_budget = ConcurrencyWaitBudget::default(); let (admission, admission_decision_ms) = if authorization.nested.is_some() { let permit = authorization.nested_permit.take().ok_or_else(|| { GatewayError::new( @@ -741,7 +740,6 @@ impl DefaultExecutionService { request_id, authorization.deadline_at, authorization.cancellation.clone(), - &concurrency_wait_budget, ) .await?; let admission = if let Some(permit) = authorization.nested_permit.take() { @@ -774,7 +772,6 @@ impl DefaultExecutionService { Ok(ExecutionStartGuard { admission, active_request, - concurrency_wait_budget, admission_decision_ms, }) } @@ -802,7 +799,6 @@ impl DefaultExecutionService { let ExecutionStartGuard { admission, active_request, - concurrency_wait_budget, admission_decision_ms, } = start_guard; let observation = plan @@ -884,7 +880,7 @@ impl DefaultExecutionService { ) .await { - Ok(core) => core.with_concurrency_wait_budget(concurrency_wait_budget), + Ok(core) => core, Err(error) => { if let Some(budget) = &self.budget { settle_budget( @@ -1245,13 +1241,13 @@ impl DefaultExecutionService { request_id: &ModelRequestId, deadline_at: Deadline, cancellation: CancellationToken, - budget: &ConcurrencyWaitBudget, ) -> Result { let policy = client.snapshot.client_queue_policy(); let limits = client.policy.limits(); let key = client.policy.key_id(); + let budget = ConcurrencyWaitBudget::default(); let mut waiting = - CapacityWait::new(&self.admission_waiting, policy, deadline_at.at(), budget); + CapacityWait::new(&self.admission_waiting, policy, deadline_at.at(), &budget); let mut admission = AdmissionLease { port: Arc::clone(&self.admissions), diagnostics: self.diagnostics.clone(), diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index 732d5d497..4731759d6 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -880,7 +880,7 @@ impl AttemptContext { self.deadline } - /// 返回密钥准入、账号选择与重试共用的排队预算 + /// 返回账号选择与后续重试共用的排队预算 #[must_use] pub const fn concurrency_wait_budget(&self) -> &crate::concurrency::ConcurrencyWaitBudget { &self.request.concurrency_wait_budget diff --git a/backend/crates/gateway-core/src/settings/compiled.rs b/backend/crates/gateway-core/src/settings/compiled.rs index 6d6ebcbc0..986de5c76 100644 --- a/backend/crates/gateway-core/src/settings/compiled.rs +++ b/backend/crates/gateway-core/src/settings/compiled.rs @@ -74,13 +74,12 @@ impl CompiledSettings { super::validate_request_limits( settings.max_waiting_per_key, settings.max_waiting_per_account, - settings.concurrency_wait_timeout_seconds, + settings.key_queue_timeout_seconds, + settings.account_queue_timeout_seconds, settings.max_account_rotations, settings.openai_session_affinity_ttl_hours, ) .map_err(|_| InvalidSettings)?; - let queue_timeout = - Duration::from_secs(u64::from(settings.concurrency_wait_timeout_seconds)); let responses_max_decompressed_body_bytes = super::response_body_limit(settings.responses_max_decompressed_body_bytes) .map_err(|_| InvalidSettings)?; @@ -104,11 +103,12 @@ impl CompiledSettings { .with_smart_scheduling(settings.smart_scheduling) .with_queue(ConcurrencyQueuePolicy { max_waiting: settings.max_waiting_per_account, - timeout: queue_timeout, - }); + timeout: Duration::from_secs(u64::from(settings.account_queue_timeout_seconds)), + }) + .with_switch_account_on_queue_failure(settings.switch_account_on_queue_failure); let client_queue_policy = ConcurrencyQueuePolicy { max_waiting: settings.max_waiting_per_key, - timeout: queue_timeout, + timeout: Duration::from_secs(u64::from(settings.key_queue_timeout_seconds)), }; let request_location = settings .request_location_enabled diff --git a/backend/crates/gateway-core/src/settings/values.rs b/backend/crates/gateway-core/src/settings/values.rs index f2b0ad80b..9634a0525 100644 --- a/backend/crates/gateway-core/src/settings/values.rs +++ b/backend/crates/gateway-core/src/settings/values.rs @@ -20,7 +20,9 @@ pub struct SettingsValues { pub(crate) max_concurrent_per_account: u32, pub(crate) max_waiting_per_key: u32, pub(crate) max_waiting_per_account: u32, - pub(crate) concurrency_wait_timeout_seconds: u32, + pub(crate) key_queue_timeout_seconds: u32, + pub(crate) account_queue_timeout_seconds: u32, + pub(crate) switch_account_on_queue_failure: bool, pub(crate) max_retry_after_seconds: u32, pub(crate) openai_guardian_reserved_concurrency: u32, pub(crate) openai_account_affinity: crate::account::AccountAffinity, @@ -158,11 +160,15 @@ impl SettingsValues { mut self, max_waiting_per_key: u32, max_waiting_per_account: u32, - timeout_seconds: u32, + key_timeout_seconds: u32, + account_timeout_seconds: u32, + switch_on_failure: bool, ) -> Self { self.max_waiting_per_key = max_waiting_per_key; self.max_waiting_per_account = max_waiting_per_account; - self.concurrency_wait_timeout_seconds = timeout_seconds; + self.key_queue_timeout_seconds = key_timeout_seconds; + self.account_queue_timeout_seconds = account_timeout_seconds; + self.switch_account_on_queue_failure = switch_on_failure; self } @@ -185,7 +191,9 @@ impl SettingsValues { max_concurrent_per_account, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: crate::account::AccountAffinity::default(), @@ -221,7 +229,8 @@ impl SettingsValues { pub fn validate_request_limits( max_waiting_per_key: u32, max_waiting_per_account: u32, - timeout_seconds: u32, + key_timeout_seconds: u32, + account_timeout_seconds: u32, max_account_rotations: u32, affinity_ttl_hours: u32, ) -> Result<(), &'static str> { @@ -229,8 +238,12 @@ pub fn validate_request_limits( (max_waiting_per_key <= 1_000, "max_waiting_per_key"), (max_waiting_per_account <= 1_000, "max_waiting_per_account"), ( - (1..=120).contains(&timeout_seconds), - "concurrency_wait_timeout_seconds", + (1..=120).contains(&key_timeout_seconds), + "key_queue_timeout_seconds", + ), + ( + (1..=120).contains(&account_timeout_seconds), + "account_queue_timeout_seconds", ), ( max_account_rotations <= crate::account::MAX_ACCOUNT_ROTATIONS, diff --git a/backend/crates/gateway-core/tests/concurrency.rs b/backend/crates/gateway-core/tests/concurrency.rs index 3a6afa293..48b8b6fa4 100644 --- a/backend/crates/gateway-core/tests/concurrency.rs +++ b/backend/crates/gateway-core/tests/concurrency.rs @@ -172,9 +172,8 @@ fn request_deadline_bounds_waiting_and_dropped_future_releases_its_ticket() { } #[test] -fn new_layers_and_retries_cannot_restart_a_requests_wait_budget() { +fn account_reselection_and_retries_share_the_account_wait_budget() { block_on(async { - let keys = ConcurrencyWaitQueue::default(); let accounts = ConcurrencyWaitQueue::default(); let policy = ConcurrencyQueuePolicy { max_waiting: 1, @@ -182,11 +181,11 @@ fn new_layers_and_retries_cannot_restart_a_requests_wait_budget() { }; let request_deadline = SystemTime::now() + Duration::from_secs(5); let budget = ConcurrencyWaitBudget::default(); - let mut key_wait = CapacityWait::new(&keys, policy, request_deadline, &budget); - key_wait.wait(&["key"]).await.unwrap(); - drop(key_wait); + let mut first = CapacityWait::new(&accounts, policy, request_deadline, &budget); + first.wait(&["first_account"]).await.unwrap(); + drop(first); - // 第一层已取得重试机会，后续准备跨过总时限后不能在账号层重新排队 + // 第一次账号等待已启动时限，重试或改选其他账号都不能重新计时 futures_timer::Delay::new(policy.timeout).await; let next_attempt_budget = budget.clone(); let mut account_wait = diff --git a/backend/crates/gateway-core/tests/engine/execution/mod.rs b/backend/crates/gateway-core/tests/engine/execution/mod.rs index d655eab12..3d213002b 100644 --- a/backend/crates/gateway-core/tests/engine/execution/mod.rs +++ b/backend/crates/gateway-core/tests/engine/execution/mod.rs @@ -4390,10 +4390,7 @@ impl Provider for QueuedAccountProvider { ) -> Result { let mut waiting = gateway_core::concurrency::CapacityWait::new( &self.waiting, - gateway_core::concurrency::ConcurrencyQueuePolicy { - max_waiting: 1, - timeout: Duration::from_secs(5), - }, + context.account_selection_policy().queue_policy(), context.deadline().at(), context.concurrency_wait_budget(), ); @@ -4406,7 +4403,7 @@ impl Provider for QueuedAccountProvider { } #[test] -fn account_wait_inherits_the_budget_spent_during_client_admission() { +fn account_wait_starts_its_own_budget_after_client_admission() { use futures::FutureExt; block_on(async { let admissions = Arc::new(BoundedAdmissions::default()); @@ -4419,7 +4416,10 @@ fn account_wait_inherits_the_budget_spent_during_client_admission() { }, FastMode::Default, ); - let settings = snapshot.settings().clone().with_concurrency_queues(1, 0, 1); + let settings = snapshot + .settings() + .clone() + .with_concurrency_queues(1, 1, 1, 2, false); let snapshot = snapshot.with_settings(&settings).unwrap(); let service = DefaultExecutionService::new( RuntimeSnapshotHandle::new(snapshot), @@ -4437,12 +4437,12 @@ fn account_wait_inherits_the_budget_spent_during_client_admission() { .start(request(&service, ClientTransport::HttpSse)) .await .unwrap(); - let started_at = Instant::now(); let mut queued = service.start(request(&service, ClientTransport::HttpSse)); assert!(queued.as_mut().now_or_never().is_none()); futures_timer::Delay::new(Duration::from_millis(200)).await; drop(running); let mut started = queued.await.unwrap(); + let account_started = Instant::now(); let error = started.session.collect_uncommitted().await.unwrap_err(); assert!( matches!( @@ -4453,9 +4453,8 @@ fn account_wait_inherits_the_budget_spent_during_client_admission() { "{error:?}" ); assert!( - started_at.elapsed() < Duration::from_secs(3), - "账号层不能重新获得自己的 5 秒预算，实际耗时 {:?}", - started_at.elapsed() + account_started.elapsed() >= Duration::from_secs(2), + "账号排队应获得独立的两秒预算" ); assert!(admissions.active.lock().unwrap().is_empty()); }); @@ -4532,6 +4531,8 @@ fn queue_service( max_waiting, 0, u32::try_from(timeout.as_secs()).unwrap(), + 30, + false, ); let snapshot = snapshot.with_settings(&settings).unwrap(); let service = DefaultExecutionService::new( @@ -4704,7 +4705,7 @@ fn request_settings_recompute_routing_and_admission_without_changing_sibling_or_ let settings = snapshot .settings() .clone() - .with_concurrency_queues(0, 0, 30); + .with_concurrency_queues(0, 0, 30, 30, false); let snapshot = snapshot.with_settings(&settings).unwrap(); let snapshots = RuntimeSnapshotHandle::new(snapshot); let admissions = Arc::new(Admissions::default()); diff --git a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs index bef8d2035..1f18f3554 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs @@ -144,7 +144,9 @@ pub struct RuntimeSettings { pub max_concurrent_per_account: u32, pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, - pub concurrency_wait_timeout_seconds: u32, + pub key_queue_timeout_seconds: u32, + pub account_queue_timeout_seconds: u32, + pub switch_account_on_queue_failure: bool, pub max_retry_after_seconds: u32, pub openai_guardian_reserved_concurrency: u32, pub openai_account_affinity: AccountAffinity, @@ -180,7 +182,9 @@ pub struct ReplaceRuntimeSettings { pub max_concurrent_per_account: u32, pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, - pub concurrency_wait_timeout_seconds: u32, + pub key_queue_timeout_seconds: u32, + pub account_queue_timeout_seconds: u32, + pub switch_account_on_queue_failure: bool, pub max_retry_after_seconds: u32, pub openai_guardian_reserved_concurrency: u32, pub openai_account_affinity: AccountAffinity, @@ -378,7 +382,9 @@ impl From for ReplaceRuntimeSettings { max_concurrent_per_account: settings.max_concurrent_per_account, max_waiting_per_key: settings.max_waiting_per_key, max_waiting_per_account: settings.max_waiting_per_account, - concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, + key_queue_timeout_seconds: settings.key_queue_timeout_seconds, + account_queue_timeout_seconds: settings.account_queue_timeout_seconds, + switch_account_on_queue_failure: settings.switch_account_on_queue_failure, max_retry_after_seconds: settings.max_retry_after_seconds, openai_guardian_reserved_concurrency: settings.openai_guardian_reserved_concurrency, openai_account_affinity: settings.openai_account_affinity, diff --git a/backend/crates/gateway-store/src/admin_adapter.rs b/backend/crates/gateway-store/src/admin_adapter.rs index f8fb81c8e..4c0787d2d 100644 --- a/backend/crates/gateway-store/src/admin_adapter.rs +++ b/backend/crates/gateway-store/src/admin_adapter.rs @@ -103,7 +103,9 @@ impl SettingsStore for AdminSettingsStoreAdapter { "max_concurrent_per_account".to_owned(), "max_waiting_per_key".to_owned(), "max_waiting_per_account".to_owned(), - "concurrency_wait_timeout_seconds".to_owned(), + "key_queue_timeout_seconds".to_owned(), + "account_queue_timeout_seconds".to_owned(), + "switch_account_on_queue_failure".to_owned(), "max_retry_after_seconds".to_owned(), "openai_guardian_reserved_concurrency".to_owned(), "openai_account_affinity".to_owned(), diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 7216cb23e..cda4b001c 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -155,7 +155,7 @@ pub(crate) async fn load_runtime_settings_from_pool(pool: &PgPool) -> StoreResul refresh_concurrency, max_concurrent_per_account, rotation_strategy, smart_scheduling_json, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, - min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, max_retry_after_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, + min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, key_queue_timeout_seconds, account_queue_timeout_seconds, switch_account_on_queue_failure, max_retry_after_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, account_error_protection_json, account_error_protection_revision, account_warmup_enabled, account_warmup_schedule_time, account_warmup_model, reset_credit_lead_minutes, reset_credit_max_weekly_remaining_percent from runtime_settings where id = 1", @@ -311,7 +311,7 @@ pub(crate) async fn load_runtime_settings_in_transaction( refresh_concurrency, max_concurrent_per_account, rotation_strategy, smart_scheduling_json, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, - min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, max_retry_after_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, + min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, key_queue_timeout_seconds, account_queue_timeout_seconds, switch_account_on_queue_failure, max_retry_after_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, account_error_protection_json, account_error_protection_revision, account_warmup_enabled, account_warmup_schedule_time, account_warmup_model, reset_credit_lead_minutes, reset_credit_max_weekly_remaining_percent from runtime_settings where id = 1", @@ -373,7 +373,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( min_codex_cli_version = $10, max_waiting_per_key = $11, max_waiting_per_account = $12, - concurrency_wait_timeout_seconds = $13, + key_queue_timeout_seconds = $13, account_error_protection_revision = case when account_error_protection_json is distinct from $14::jsonb then config_revision + 1 else account_error_protection_revision end, account_error_protection_json = $14, request_location_json = $15, @@ -392,6 +392,8 @@ pub(crate) async fn update_runtime_settings_in_transaction( reset_credit_lead_minutes = $29, reset_credit_max_weekly_remaining_percent = $30, max_retry_after_seconds = $31, + account_queue_timeout_seconds = $32, + switch_account_on_queue_failure = $33, updated_at = now() where id = 1 returning config_revision", @@ -408,7 +410,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( .bind(update.values.min_codex_cli_version.as_deref()) .bind(i64::from(update.values.max_waiting_per_key)) .bind(i64::from(update.values.max_waiting_per_account)) - .bind(i64::from(update.values.concurrency_wait_timeout_seconds)) + .bind(i64::from(update.values.key_queue_timeout_seconds)) .bind(sqlx::types::Json(&update.values.account_error_protection)) .bind(sqlx::types::Json( update.values.request_location @@ -435,6 +437,8 @@ pub(crate) async fn update_runtime_settings_in_transaction( .bind(i64::from(update.values.reset_credit_lead_minutes)) .bind(update.values.reset_credit_max_weekly_remaining_percent.map(i64::from)) .bind(i64::from(update.values.max_retry_after_seconds)) + .bind(i64::from(update.values.account_queue_timeout_seconds)) + .bind(update.values.switch_account_on_queue_failure) .fetch_optional(&mut **transaction) .await .map_err(|source| postgres_unavailable("update runtime settings in transaction", source))? @@ -521,7 +525,9 @@ struct RuntimeSettingsRow { updated_at: DateTime, max_waiting_per_key: i64, max_waiting_per_account: i64, - concurrency_wait_timeout_seconds: i64, + key_queue_timeout_seconds: i64, + account_queue_timeout_seconds: i64, + switch_account_on_queue_failure: bool, max_retry_after_seconds: i64, openai_guardian_reserved_concurrency: i64, openai_account_affinity: String, @@ -580,7 +586,9 @@ fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult, max_waiting_per_key: i64, max_waiting_per_account: i64, - concurrency_wait_timeout_seconds: i64, + key_queue_timeout_seconds: i64, + account_queue_timeout_seconds: i64, + switch_account_on_queue_failure: bool, max_retry_after_seconds: i64, openai_guardian_reserved_concurrency: i64, openai_account_affinity: String, @@ -246,7 +248,7 @@ async fn load_settings( transaction: &mut Transaction<'_, Postgres>, ) -> StoreResult<(Revision, SettingsValues)> { let row = sqlx::query_as::<_, SnapshotSettingsRow>( - "select config_revision, account_error_protection_revision, account_error_protection_json, max_concurrent_per_account, rotation_strategy, smart_scheduling_json, model_mappings_json, min_codex_desktop_version, min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, max_retry_after_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, request_location_json, request_location_enabled, codex_privacy_policy_json, responses_max_decompressed_body_bytes, provider_request_profiles_json, pricing_overrides_json, pricing_synced_json from runtime_settings where id = 1", + "select config_revision, account_error_protection_revision, account_error_protection_json, max_concurrent_per_account, rotation_strategy, smart_scheduling_json, model_mappings_json, min_codex_desktop_version, min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, key_queue_timeout_seconds, account_queue_timeout_seconds, switch_account_on_queue_failure, max_retry_after_seconds, openai_guardian_reserved_concurrency, openai_account_affinity, max_account_rotations, openai_session_affinity_ttl_hours, request_location_json, request_location_enabled, codex_privacy_policy_json, responses_max_decompressed_body_bytes, provider_request_profiles_json, pricing_overrides_json, pricing_synced_json from runtime_settings where id = 1", ) .fetch_optional(&mut **transaction) .await @@ -297,7 +299,9 @@ async fn load_settings( .with_concurrency_queues( to_u32(row.max_waiting_per_key)?, to_u32(row.max_waiting_per_account)?, - to_u32(row.concurrency_wait_timeout_seconds)?, + to_u32(row.key_queue_timeout_seconds)?, + to_u32(row.account_queue_timeout_seconds)?, + row.switch_account_on_queue_failure, ), )) } diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index 58fa04ff7..fb2f42d27 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -30,7 +30,9 @@ pub(super) fn settings_with_margin(refresh_margin_seconds: u64) -> RuntimeSettin max_concurrent_per_account: 3, max_waiting_per_key: 0, max_waiting_per_account: 0, - concurrency_wait_timeout_seconds: 30, + key_queue_timeout_seconds: 30, + account_queue_timeout_seconds: 30, + switch_account_on_queue_failure: false, max_retry_after_seconds: 240, openai_guardian_reserved_concurrency: 0, openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed, @@ -568,17 +570,21 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() { before.values.max_concurrent_per_account, before.values.max_waiting_per_key, before.values.max_waiting_per_account, - before.values.concurrency_wait_timeout_seconds, + before.values.key_queue_timeout_seconds, + before.values.account_queue_timeout_seconds, + before.values.switch_account_on_queue_failure, before.values.openai_guardian_reserved_concurrency, ), - (3, 10, 10, 30, 5) + (3, 10, 10, 30, 30, false, 5) ); let mut update = settings_with_margin(1234); update.values.max_concurrent_per_account = 7; update.values.refresh_concurrency = 12; update.values.max_waiting_per_key = 5; update.values.max_waiting_per_account = 7; - update.values.concurrency_wait_timeout_seconds = 12; + update.values.key_queue_timeout_seconds = 12; + update.values.account_queue_timeout_seconds = 5; + update.values.switch_account_on_queue_failure = true; update.values.openai_guardian_reserved_concurrency = 1; repository.update_runtime_settings(update).await.unwrap(); let settings = repository.load_runtime_settings().await.unwrap(); @@ -589,10 +595,12 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() { settings.values.refresh_margin_seconds, settings.values.max_waiting_per_key, settings.values.max_waiting_per_account, - settings.values.concurrency_wait_timeout_seconds, + settings.values.key_queue_timeout_seconds, + settings.values.account_queue_timeout_seconds, + settings.values.switch_account_on_queue_failure, settings.values.openai_guardian_reserved_concurrency, ), - (7, 12, 1234, 5, 7, 12, 1) + (7, 12, 1234, 5, 7, 12, 5, true, 1) ); let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone()) .load_runtime_snapshot() @@ -606,14 +614,20 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() { .unwrap(), snapshot_values["max_waiting_per_key"].as_u64().unwrap(), snapshot_values["max_waiting_per_account"].as_u64().unwrap(), - snapshot_values["concurrency_wait_timeout_seconds"] + snapshot_values["key_queue_timeout_seconds"] + .as_u64() + .unwrap(), + snapshot_values["switch_account_on_queue_failure"] + .as_bool() + .unwrap(), + snapshot_values["account_queue_timeout_seconds"] .as_u64() .unwrap(), snapshot_values["openai_guardian_reserved_concurrency"] .as_u64() .unwrap(), ), - (7, 5, 7, 12, 1) + (7, 5, 7, 12, true, 5, 1) ); assert!(snapshot.config_revision > before.config_revision); database.close().await; diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 9026fbd8f..14066f10b 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -14,7 +14,9 @@ use gateway_core::account::{ AccountStatus, CredentialState, PreferredAccountSelection, ProviderAccount, ProviderAccountId, QuotaEvidence, RotationStrategy, }; -use gateway_core::concurrency::{CapacityWait, ConcurrencyWaitQueue, QueueRejection, WaitPriority}; +use gateway_core::concurrency::{ + CapacityWait, ConcurrencyQueuePolicy, ConcurrencyWaitQueue, QueueRejection, WaitPriority, +}; use gateway_core::engine::{AttemptContext, ContinuationAttempt, policy::AccountPolicyError}; use gateway_core::provider_ports::{ ProviderConcurrencyPool, ProviderCooldownPort, ProviderCooldownScope, ProviderLeaseAcquisition, @@ -472,19 +474,61 @@ impl CodexCredentialSelector { cyber_policy_session_key: Option<&ProviderSessionAffinityKey>, model: CodexSelectionModel<'_>, ) -> Result { + let policy = request.attempt.account_selection_policy(); let follow_only = !request.attempt.is_diagnostic_required_account() - && request - .attempt - .account_selection_policy() - .openai_account_affinity() - == AccountAffinity::Strict + && policy.openai_account_affinity() == AccountAffinity::Strict && request .session_affinity_observation .is_some_and(CodexSessionAffinity::follow_only); + let select = |queue_policy| { + self.select_inner_loop( + request, + cyber_policy_session_key, + model, + follow_only, + queue_policy, + ) + }; + let selection = async { + let rejection = match select(policy.queue_policy()).await { + Err(CredentialSelectionError::QueueRejected(reason)) + if !follow_only + && request.attempt.required_account().is_none() + && matches!( + request.attempt.continuation_attempt(), + ContinuationAttempt::None | ContinuationAttempt::ReplayAny + ) => + { + reason + } + result => return result, + }; + policy + .reselect_after_queue_rejection( + rejection, + request.attempt.deadline().remaining(), + |queue| { + request.attempt.trace().record( + "account.queue.reselect", + serde_json::json!({"reason": rejection.to_string()}), + ); + select(queue) + }, + |err| { + matches!( + err, + CredentialSelectionError::CapacityUnavailable { .. } + | CredentialSelectionError::NoEligibleCredential + | CredentialSelectionError::QuotaExhausted + ) + }, + ) + .await + }; let result = tokio::select! { biased; () = request.attempt.cancellation().cancelled() => Err(CredentialSelectionError::Cancelled), - result = self.select_inner_loop(request, cyber_policy_session_key, model, follow_only) => result, + result = selection => result, }; result.map_err(|error| { if follow_only @@ -510,6 +554,7 @@ impl CodexCredentialSelector { cyber_policy_session_key: Option<&ProviderSessionAffinityKey>, model: CodexSelectionModel<'_>, follow_only: bool, + mut queue_policy: ConcurrencyQueuePolicy, ) -> Result { let adapter_model = match model { CodexSelectionModel::Requested(model) if request.managed_upstream => Some( @@ -528,7 +573,6 @@ impl CodexCredentialSelector { .attempt .account_selection_policy() .openai_account_affinity(); - let mut queue_policy = request.attempt.account_selection_policy().queue_policy(); if follow_only { // 关闭普通账号排队不能放开后代线程换号，沿用队列时间设置并保留有界人数 queue_policy.max_waiting = if queue_policy.max_waiting == 0 { diff --git a/backend/crates/providers/openai/tests/credential/contract/scheduling_projection.rs b/backend/crates/providers/openai/tests/credential/contract/scheduling_projection.rs index 7068d7db7..a00a146d1 100644 --- a/backend/crates/providers/openai/tests/credential/contract/scheduling_projection.rs +++ b/backend/crates/providers/openai/tests/credential/contract/scheduling_projection.rs @@ -358,6 +358,141 @@ async fn strict_bound_waiters_recheck_only_their_account_until_capacity_opens() } } +#[tokio::test] +async fn strict_queue_failure_reselects_once_and_only_rebinds_after_admission() { + use gateway_core::concurrency::{ConcurrencyQueuePolicy, QueueRejection}; + for (full, spare, enabled) in [ + (false, true, true), + (true, true, true), + (false, false, true), + (true, false, true), + (false, true, false), + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let first = ProviderAccountId::new("acct_first").unwrap(); + let second = ProviderAccountId::new("acct_second").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + let selector = selector_with_runtime( + &store, + leases.clone(), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + Arc::new(MemoryCooldownPort::new()), + ); + let policy = account_policy() + .with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_millis(150), + }) + .with_switch_account_on_queue_failure(enabled); + let context = || projection_attempt("key_codex_contract", contract_account_scope(), policy); + let key = ProviderSessionAffinityKey::try_new("strict-queue-failure").unwrap(); + assert_eq!( + select(&selector, &context(), &key, "gpt-5.4") + .await + .unwrap() + .account_id(), + &first + ); + leases + .signals + .lock() + .unwrap() + .insert(first.clone(), signal(2)); + if !spare { + leases + .signals + .lock() + .unwrap() + .insert(second.clone(), signal(2)); + } + let head_ctx = context(); + let mut head = Box::pin(select(&selector, &head_ctx, &key, "gpt-5.4")); + if full { + assert!(head.as_mut().now_or_never().is_none()); + } + let ctx = context(); + let mut pending = Box::pin(select(&selector, &ctx, &key, "gpt-5.4")); + if !full { + // B 即使空闲也不能在 A 的等待期内直接分流 + assert!(pending.as_mut().now_or_never().is_none()); + } + let result = pending.await; + let expected = if spare && enabled { + assert_eq!(result.unwrap().account_id(), &second); + &second + } else { + let reason = if full { + QueueRejection::Full + } else { + QueueRejection::Timeout + }; + assert!( + matches!(result, Err(CredentialSelectionError::QueueRejected(actual)) if actual == reason) + ); + &first + }; + drop(head); + leases.signals.lock().unwrap().clear(); + assert_eq!( + select(&selector, &context(), &key, "gpt-5.4") + .await + .unwrap() + .account_id(), + expected + ); + } +} + +#[tokio::test] +async fn account_queue_timeout_can_reselect_for_preferred_and_sessionless_requests() { + for mode in [AccountAffinity::Preferred, AccountAffinity::Relaxed] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let first = ProviderAccountId::new("acct_first").unwrap(); + let second = ProviderAccountId::new("acct_second").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + let selector = selector(&store, leases.clone()); + let policy = account_policy() + .with_openai_account_affinity(mode) + .with_switch_account_on_queue_failure(true) + .with_queue(gateway_core::concurrency::ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_millis(30), + }); + let ctx = projection_attempt("key_codex_contract", contract_account_scope(), policy); + let key = ProviderSessionAffinityKey::try_new("queue-fallback").unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(), + attempt: &ctx, + session_affinity_key: (mode == AccountAffinity::Preferred).then_some(&key), + }; + assert_eq!( + selector.select(&request).await.unwrap().account_id(), + &first + ); + for id in [&first, &second] { + leases.signals.lock().unwrap().insert(id.clone(), signal(2)); + } + let mut pending = Box::pin(selector.select(&request)); + assert!(pending.as_mut().now_or_never().is_none()); + // B 在排队后恢复，等待时限短于队首重查周期，只有最后一次立即选号能取到 B + leases.signals.lock().unwrap().remove(&second); + assert_eq!(pending.await.unwrap().account_id(), &second); + if mode == AccountAffinity::Preferred { + leases.signals.lock().unwrap().remove(&first); + assert_eq!( + selector.select(&request).await.unwrap().account_id(), + &first + ); + } + } +} + #[tokio::test] async fn movable_waiters_use_other_accounts_even_before_queue_promotion() { for (mode, has_session, bind_session) in [ diff --git a/backend/crates/providers/openai/tests/provider/contract/capacity.rs b/backend/crates/providers/openai/tests/provider/contract/capacity.rs index 480c142f7..b3503c894 100644 --- a/backend/crates/providers/openai/tests/provider/contract/capacity.rs +++ b/backend/crates/providers/openai/tests/provider/contract/capacity.rs @@ -14,10 +14,12 @@ fn queued_context(request_id: &str) -> AttemptContext { ), NonZeroU32::new(1).unwrap(), SystemTime::now() + Duration::from_secs(5), - account_policy().with_queue(ConcurrencyQueuePolicy { - max_waiting: 1, - timeout: Duration::from_secs(2), - }), + account_policy() + .with_switch_account_on_queue_failure(true) + .with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), AccountAttemptContext::new(BTreeSet::new(), None, None) .with_account_scope(contract_account_scope()), None, diff --git a/backend/crates/providers/xai/src/credential/selector.rs b/backend/crates/providers/xai/src/credential/selector.rs index 13a223552..eb69778dc 100644 --- a/backend/crates/providers/xai/src/credential/selector.rs +++ b/backend/crates/providers/xai/src/credential/selector.rs @@ -10,7 +10,7 @@ use gateway_core::account::{ AccountSelector, CredentialState, ProviderAccount, ProviderAccountId, QuotaAccessState, QuotaEvidence, QuotaState, }; -use gateway_core::concurrency::{CapacityWait, ConcurrencyWaitQueue}; +use gateway_core::concurrency::{CapacityWait, ConcurrencyQueuePolicy, ConcurrencyWaitQueue}; use gateway_core::engine::policy::AccountPolicyError; use gateway_core::provider_ports::{ ProviderCooldown, ProviderCooldownPort, ProviderCooldownScope, ProviderLeaseAcquisition, @@ -93,7 +93,38 @@ impl GrokAccountSessionSelector { &self, request: GrokSessionSelection, ) -> Result { - let queue_policy = request.account_selection_policy().queue_policy(); + let policy = request.account_selection_policy(); + let select = |queue| self.select_with_queue(&request, queue); + match select(policy.queue_policy()).await { + Err(GrokSessionSelectorError::QueueRejected(reason)) + if request.required_account().is_none() => + { + policy + .reselect_after_queue_rejection( + reason, + request.deadline().remaining(), + select, + |err| { + matches!( + err, + GrokSessionSelectorError::CapacityUnavailable { .. } + | GrokSessionSelectorError::NoEligibleSession + | GrokSessionSelectorError::AccountCoolingDown { .. } + | GrokSessionSelectorError::ModelCoolingDown { .. } + ) + }, + ) + .await + } + result => result, + } + } + + async fn select_with_queue( + &self, + request: &GrokSessionSelection, + queue_policy: ConcurrencyQueuePolicy, + ) -> Result { let mut waiting = CapacityWait::new( &self.waiting, queue_policy, diff --git a/backend/crates/providers/xai/tests/credential/selector.rs b/backend/crates/providers/xai/tests/credential/selector.rs index 4a5ae2154..36713c757 100644 --- a/backend/crates/providers/xai/tests/credential/selector.rs +++ b/backend/crates/providers/xai/tests/credential/selector.rs @@ -1326,6 +1326,50 @@ async fn saturated_account_snapshot_queues_and_rechecks_live_capacity() { assert_eq!(selected.account_id(), &id); } +#[tokio::test] +async fn account_queue_timeout_reselects_only_if_enabled_and_not_pinned() { + use futures::FutureExt; + let fixture = SelectorFixture::new(&["first", "spare"]).await; + let first = account_id("first"); + let spare = account_id("spare"); + for (enabled, pinned) in [(true, false), (true, true), (false, false)] { + let base = queue_request(&fixture, Duration::from_millis(30)); + let req = GrokSessionSelection::new( + base.upstream_model().clone(), + BTreeSet::new(), + pinned.then_some(first.clone()), + base.account_selection_policy() + .with_switch_account_on_queue_failure(enabled), + base.deadline(), + base.account_scope().clone(), + base.client_api_key_id().clone(), + ); + for signal in fixture.coordinator.signals.lock().unwrap().values_mut() { + signal.in_flight = 2; + } + let mut pending = fixture.selector.select(req); + assert!(pending.as_mut().now_or_never().is_none()); + fixture + .coordinator + .signals + .lock() + .unwrap() + .get_mut(&spare) + .unwrap() + .in_flight = 0; + if enabled && !pinned { + assert_eq!(pending.await.unwrap().account_id(), &spare); + } else { + assert!(matches!( + pending.await, + Err(GrokSessionSelectorError::QueueRejected( + gateway_core::concurrency::QueueRejection::Timeout + )) + )); + } + } +} + #[tokio::test] async fn account_queue_cancel_reclaims_capacity_and_does_not_wait_for_upstream_cooldown() { use futures::FutureExt; diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 40dd3d391..6df68a2ef 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -36,3 +36,4 @@ bd73379456839e8a12d7591ad2a839901af0f5ba27871248569abdfa26bc83ef 0035_cpr_link_ d05442ee4cb85db8131e015dbed9c6ad988f93306abcda34647317674857ee44 0036_account_display_names.sql b215ea90d5c8753aeb8192b17ffe65f7a95c1d620aaeaf9fc4337d8cf6f2634e 0037_retry_after_wait_limit.sql 321f4a09367717eb13c3cf5dbfd1f4b014e31c615a74bf6e3b09b8e810d87198 0038_preserve_client_key_ledger.sql +80ba91cad23667ace80a43daf905ca78fd0f6de5fb57f444c9ebbfa2fab90c34 0039_queue_timeouts.sql diff --git a/backend/migrations/0039_queue_timeouts.sql b/backend/migrations/0039_queue_timeouts.sql new file mode 100644 index 000000000..e3e804359 --- /dev/null +++ b/backend/migrations/0039_queue_timeouts.sql @@ -0,0 +1,10 @@ +-- 两层排队分别沿用实例已配置的时限，账号队满或超时默认返回错误 +ALTER TABLE runtime_settings + RENAME COLUMN concurrency_wait_timeout_seconds TO key_queue_timeout_seconds; + +ALTER TABLE runtime_settings + ADD COLUMN account_queue_timeout_seconds BIGINT NOT NULL DEFAULT 30 + CHECK (account_queue_timeout_seconds BETWEEN 1 AND 120), + ADD COLUMN switch_account_on_queue_failure BOOLEAN NOT NULL DEFAULT FALSE; + +UPDATE runtime_settings SET account_queue_timeout_seconds = key_queue_timeout_seconds; diff --git a/docs/api.md b/docs/api.md index d4a713bca..3f3486d31 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1912,7 +1912,7 @@ OpenAI Key 可设置 `installationId`（带连字符的 UUID，大小写均可 繁忙时先按调度策略选择可用账号，临时分流不改写会话绑定；没有可立即使用的账号时才按配置排队。严格亲和下已有绑定的后代线程只等待当前会话账号， 根线程迁移后跟随新账号。绑定缺失时，任何有可靠会话身份的请求都可按调度策略原子首绑，后续请求沿用该绑定 -严格亲和下的后代线程始终启用账号排队，沿用 `concurrencyWaitTimeoutSeconds` 和请求截止时间； +严格亲和下的后代线程始终启用账号排队，沿用 `accountQueueTimeoutSeconds` 和请求截止时间； `maxWaitingPerAccount` 为 0 时使用每队列 1,000 人上限，否则沿用配置的上限。 可识别来源的 Search、Images、Live 创建请求采用同样规则，身份与轮次关联见[会话绑定](architecture.md#6-路由账号范围与-continuation)。 插件显式选号须通过账号资格与租约检查，见[调度合同](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#路由调度与观察)。普通请求关闭账号排队且没有可用容量时返回 `503` / `account_capacity_unavailable`， @@ -1928,7 +1928,7 @@ xAI 由本地账号/模型冷却、无可选账号、并发容量不足及选号 保留 `account_unavailable_error` 类型和各自的机器码；可用备用 Provider 仍按路由范围与换号预算尝试。 同次请求选号耗尽时保留最近一次上游失败，新的基础设施故障仍按自身原因报告 -队列满返回 `429` / `concurrency_queue_full`，排队超时返回 `429` / `concurrency_queue_timeout`； +默认队列满返回 `429` / `concurrency_queue_full`，排队超时返回 `429` / `concurrency_queue_timeout`； WebSocket 使用对应错误事件。等待期间不发送上游请求，取消后释放等待位置，排队重查不重复计入 RPM。 SSE 在 Key 准入完成并取得执行会话前不发送保活帧，此阶段保留 HTTP 错误状态； 取得会话后的账号等待受 SSE 保活覆盖，已开始交付的失败沿用流内错误合同 @@ -1977,7 +1977,7 @@ HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly | [隐私策略](#隐私策略) | `codexPrivacyPolicy` | | 模型映射 | `modelMappings` | | 凭据刷新 | `refreshMarginSeconds`、`refreshConcurrency` | -| [并发与排队](#并发与排队) | `maxConcurrentPerAccount`、`maxWaitingPerKey`、`maxWaitingPerAccount`、`openaiGuardianReservedConcurrency`、`concurrencyWaitTimeoutSeconds` | +| [并发与排队](#并发与排队) | `maxConcurrentPerAccount`、`maxWaitingPerKey`、`maxWaitingPerAccount`、`openaiGuardianReservedConcurrency`、`keyQueueTimeoutSeconds`、`accountQueueTimeoutSeconds`、`switchAccountOnQueueFailure` | | [上游重试等待](#容量拒绝与重试) | `maxRetryAfterSeconds` | | [亲和与换号](#账号亲和与换号) | `openaiAccountAffinity`、`openaiSessionAffinityTtlHours`、`maxAccountRotations` | | [调度策略](#智能调度) | `rotationStrategy`、`smartScheduling` | @@ -2090,9 +2090,12 @@ UA、平台参数及自动 Git 上下文版本由 [OpenAI 上游客户端身份] 每个 Key、每个账号各自独立计数，没有单对象覆盖字段。执行并发为 5、最大排队数为 5 时， 该对象最多容纳 5 个执行请求与 5 个等待请求。Key 并发为 0（不限）时跳过 Key 排队。 OpenAI 严格亲和下后代线程的会话账号等待不随普通账号排队关闭，见 [Client Key 等待规则](#7-client-key)。 -`concurrencyWaitTimeoutSeconds` 取值 1～120，默认 30，从首次入队开始计时，密钥与账号两层共享该等待时限； -切换账号或内部重试不重新计时；若插件设置了请求总时限，等待也计入该时限。排队超时不用于中断已开始的上游生成。 -设置更新请求须包含这三个字段，新请求使用更新后的快照 +`keyQueueTimeoutSeconds` 和 `accountQueueTimeoutSeconds` 分别限制密钥与账号排队，各取 1～120 秒，默认 30 秒。 +每层从首次入队开始独立计时，账号重选、切换 Provider 或内部重试不重置账号时限。请求总期限仍优先，排队超时不用于中断已开始的上游生成 + +`switchAccountOnQueueFailure` 默认 `false`，账号队满或超时返回原排队错误。开启后，允许换号的请求会在发送前再选一次可立即准入的账号，不重新排队。 +没有空位时返回原排队错误。OpenAI 会话沿用其亲和策略：严格模式成功换号后迁移绑定，优先模式临时分流时保留主绑定。密钥排队失败、必须跟随根线程的后代、固定账号与原生续写不触发换号。 +设置更新须提交两层队列容量、两个超时值和换号开关，对新请求生效 `openaiGuardianReservedConcurrency`（默认 5，取值 0～4,294,967,295）设置 Codex Guardian 自动审批的每账号独立并发，保存后对新请求生效。 Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardian` 识别。取值 R 大于 0 时， diff --git a/docs/architecture.md b/docs/architecture.md index a160fb23b..81d217e5f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -848,8 +848,8 @@ OpenAI 内置调度仅在排队目标与固定归属或严格会话绑定一致 Key 的 RPM 在成功准入时才计数，金额限制在入队前及成功准入后检查。 普通账号容量等待只包含并发；严格亲和下的后代线程还可等待根线程恢复或迁移绑定，但等待不使权限、额度或冷却检查失效。 -密钥准入、账号选择与后续重试共享请求级等待预算，从首次入队开始计时；显式设置请求总时限时也受其约束。 -切换等待层、账号或 Provider 不重置等待时限；没有发生排队时不启动该计时，也不据此中断已开始的上游生成。 +密钥准入与账号选择各自从首次入队开始计时，账号重选、Provider 切换和后续重试共享账号等待预算，两层均受请求总期限约束。 +账号队满或超时释放队位后，可按设置对允许换号的请求再选一次可立即准入的账号。各 Provider 仍检查原有资格、租约和会话归属；OpenAI 严格亲和在取得租约后以当前绑定版本提交迁移。 队列满/超时属于本地容量拒绝，不作为上游限流反馈或 Provider 故障熔断证据 #### Guardian 独立并发 diff --git a/frontend/src/api/modules/settings/index.ts b/frontend/src/api/modules/settings/index.ts index a0f78c63c..fcb869ee1 100644 --- a/frontend/src/api/modules/settings/index.ts +++ b/frontend/src/api/modules/settings/index.ts @@ -53,7 +53,9 @@ export interface RuntimeSettings { openaiSessionAffinityTtlHours: number maxWaitingPerKey: number maxWaitingPerAccount: number - concurrencyWaitTimeoutSeconds: number + keyQueueTimeoutSeconds: number + accountQueueTimeoutSeconds: number + switchAccountOnQueueFailure: boolean maxRetryAfterSeconds: number responsesMaxDecompressedBodyBytes: number rotationStrategy: RotationStrategy diff --git a/frontend/src/i18n/locales/en/settings.json b/frontend/src/i18n/locales/en/settings.json index 937e1a927..1e2423915 100644 --- a/frontend/src/i18n/locales/en/settings.json +++ b/frontend/src/i18n/locales/en/settings.json @@ -245,6 +245,10 @@ "enterNewPasswordAgain": "Enter new password again", "saveAndLogInAgain": "Save and log in again" }, + "units": { + "seconds": "seconds", + "inSeconds": "{label} (seconds)" + }, "concurrency": { "title": "Concurrency and retries", "accountLimit": { @@ -255,17 +259,23 @@ "keyQueueHelp": "Per API key, 0 disables queuing", "accountQueueCapacity": "Account queue capacity", "accountQueueHelp": "Per account, 0 disables queuing", - "queueTimeoutLabel": "Queue timeout (s)", - "queueTimeoutHelp": "Shared by both queues, 1–120 seconds", + "keyTimeout": "API key queue timeout", + "keyTimeoutHelp": "Maximum wait for a key slot before returning an error", + "accountTimeout": "Account queue timeout", + "accountTimeoutHelp": "Starts on first account queue entry and continues across retries", + "queueFailure": "When the account queue is full or times out", + "queueFailureHelp": "Try another account with room. If switching is not possible, return the queue error.", + "queueFailureHelpTitle": "Account switching rules", + "queueFailureDetails": "Check once without queuing again. Requests pinned to an account, native continuations, and requests bound to a parent session cannot switch on their own.", + "returnError": "Return error", + "trySwitch": "Try another account", "autoReview": { "label": "Reserved auto review slots", "description": "Extra slots per Codex account, 0 shares regular concurrency", "helpTitle": "About auto review concurrency" }, - "queueTimeoutHelpTitle": "About queue timeout", - "queueTimeoutDetails": "The API key and account queues share this timeout, measured from the first queue entry", "retryAfter": { - "label": "Upstream retry wait limit (s)", + "label": "Upstream retry wait limit", "help": "Try another account above the limit", "helpTitle": "About upstream retry waits", "details": "Defaults to 240 seconds. Limits each Retry-After wait, not the total request duration. Above the limit, try another account; return the error if none is available. Set to 0 to reject positive waits. Account rotation still follows the rotation limit and request replay conditions." @@ -333,7 +343,7 @@ }, "tokenRefresh": { "title": "Credential refresh", - "leadTimeLabel": "Refresh lead time (s)", + "leadTimeLabel": "Refresh lead time", "leadTimeHelp": "Refresh access tokens before they expire", "concurrencyLabel": "Refresh concurrency limit", "concurrencyHelp": "Concurrent refresh limit, lower to reduce rate limiting" @@ -641,7 +651,6 @@ "countValidity": "Count validity", "windowHelp": "Each matching failure renews the count, a successful call clears it", "countValiditySeconds": "Count validity in seconds", - "seconds": "seconds", "cooldownTime": "Cooldown duration", "cooldownHelp": "A failed probe starts another cooldown", "cooldownTimeInSeconds": "Cooldown duration in seconds", diff --git a/frontend/src/i18n/locales/zh-Hans/settings.json b/frontend/src/i18n/locales/zh-Hans/settings.json index 47019c2f1..e6453151a 100644 --- a/frontend/src/i18n/locales/zh-Hans/settings.json +++ b/frontend/src/i18n/locales/zh-Hans/settings.json @@ -245,6 +245,10 @@ "enterNewPasswordAgain": "再次输入新密码", "saveAndLogInAgain": "保存并重新登录" }, + "units": { + "seconds": "秒", + "inSeconds": "{label}（秒）" + }, "concurrency": { "title": "并发与重试", "accountLimit": { @@ -255,17 +259,23 @@ "keyQueueHelp": "每个密钥的等待上限，0 不排队", "accountQueueCapacity": "账号队列容量", "accountQueueHelp": "每个账号的等待上限，0 不排队", - "queueTimeoutLabel": "排队超时（秒）", - "queueTimeoutHelp": "两个队列共用，1–120 秒", + "keyTimeout": "密钥排队超时", + "keyTimeoutHelp": "等待密钥并发名额的时限，超时返回错误", + "accountTimeout": "账号排队超时", + "accountTimeoutHelp": "从首次进入账号队列起计时，重试不重置", + "queueFailure": "账号队满或超时后", + "queueFailureHelp": "尝试换号时，改用有空位的账号；无法换号则返回排队错误", + "queueFailureHelpTitle": "账号换号规则", + "queueFailureDetails": "只重新选择一次，不再排队。指定账号、需要用原账号续写，或必须跟随主会话的请求不能自行换号。", + "returnError": "返回错误", + "trySwitch": "尝试换号", "autoReview": { "label": "自动审批独立并发", "description": "每个 Codex 账号的额外名额，0 共用普通并发", "helpTitle": "自动审批独立并发说明" }, - "queueTimeoutHelpTitle": "排队超时说明", - "queueTimeoutDetails": "密钥与账号队列共用此时限，从首次入队起计时", "retryAfter": { - "label": "上游重试等待上限（秒）", + "label": "上游重试等待上限", "help": "超限时尝试换号", "helpTitle": "上游重试等待说明", "details": "默认 240 秒，限制单次 Retry-After 等待，不是请求总时长。超限时尝试换号，无可用账号则返回错误。0 表示不接受正数等待。换号仍受最大换号次数和请求可重试条件限制。" @@ -333,7 +343,7 @@ }, "tokenRefresh": { "title": "凭据刷新", - "leadTimeLabel": "提前刷新（秒）", + "leadTimeLabel": "提前刷新", "leadTimeHelp": "在访问令牌到期前触发刷新", "concurrencyLabel": "刷新并发上限", "concurrencyHelp": "同时刷新的上限，调低可减少限流" @@ -641,7 +651,6 @@ "countValidity": "计数有效期", "windowHelp": "每次匹配失败后续期，普通调用成功后清零", "countValiditySeconds": "计数有效期秒数", - "seconds": "秒", "cooldownTime": "冷却时长", "cooldownHelp": "探测失败时，按此时长延后恢复", "cooldownTimeInSeconds": "冷却时长秒数", diff --git a/frontend/src/i18n/locales/zh-Hant/settings.json b/frontend/src/i18n/locales/zh-Hant/settings.json index 3410018bf..4cdbe04ab 100644 --- a/frontend/src/i18n/locales/zh-Hant/settings.json +++ b/frontend/src/i18n/locales/zh-Hant/settings.json @@ -245,6 +245,10 @@ "enterNewPasswordAgain": "再次輸入新密碼", "saveAndLogInAgain": "保存並重新登錄" }, + "units": { + "seconds": "秒", + "inSeconds": "{label}（秒）" + }, "concurrency": { "title": "並發與重試", "accountLimit": { @@ -255,17 +259,23 @@ "keyQueueHelp": "每個密鑰的等待上限，0 不排隊", "accountQueueCapacity": "賬號隊列容量", "accountQueueHelp": "每個賬號的等待上限，0 不排隊", - "queueTimeoutLabel": "排隊超時（秒）", - "queueTimeoutHelp": "兩個隊列共用，1–120 秒", + "keyTimeout": "密鑰排隊超時", + "keyTimeoutHelp": "等待密鑰併發名額的時限，超時返回錯誤", + "accountTimeout": "賬號排隊超時", + "accountTimeoutHelp": "從首次進入賬號隊列起計時，重試不重置", + "queueFailure": "賬號隊滿或超時後", + "queueFailureHelp": "嘗試換號時，改用有空位的賬號；無法換號則返回排隊錯誤", + "queueFailureHelpTitle": "賬號換號規則", + "queueFailureDetails": "只重新選擇一次，不再排隊。指定賬號、需要用原賬號續寫，或必須跟隨主會話的請求不能自行換號。", + "returnError": "返回錯誤", + "trySwitch": "嘗試換號", "autoReview": { "label": "自動審批獨立併發", "description": "每個 Codex 賬號的額外名額，0 共用普通併發", "helpTitle": "自動審批獨立併發說明" }, - "queueTimeoutHelpTitle": "排隊超時說明", - "queueTimeoutDetails": "密鑰與賬號隊列共用此時限，從首次入隊起計時", "retryAfter": { - "label": "上游重試等待上限（秒）", + "label": "上游重試等待上限", "help": "超限時嘗試換號", "helpTitle": "上游重試等待說明", "details": "預設 240 秒，限制單次 Retry-After 等待，並非請求總時長。超限時嘗試換號，無可用賬號則返回錯誤。0 表示不接受正數等待。換號仍受最大換號次數和請求可重試條件限制。" @@ -333,7 +343,7 @@ }, "tokenRefresh": { "title": "憑據刷新", - "leadTimeLabel": "提前刷新（秒）", + "leadTimeLabel": "提前刷新", "leadTimeHelp": "在訪問令牌到期前觸發刷新", "concurrencyLabel": "刷新併發上限", "concurrencyHelp": "同時刷新的上限，調低可減少限流" @@ -641,7 +651,6 @@ "countValidity": "計數有效期", "windowHelp": "每次匹配失敗後續期，普通調用成功後清零", "countValiditySeconds": "計數有效期秒數", - "seconds": "秒", "cooldownTime": "冷卻時長", "cooldownHelp": "探測失敗時，按此時長延後恢復", "cooldownTimeInSeconds": "冷卻時長秒數", diff --git a/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue b/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue index 3cb8e6a05..235be865b 100644 --- a/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue +++ b/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue @@ -1,8 +1,8 @@ @@ -216,7 +216,7 @@ function removeRule(id: string) { diff --git a/frontend/src/views/settings/composables/useSettingsForm.ts b/frontend/src/views/settings/composables/useSettingsForm.ts index b4c0d3473..0b7f9a601 100644 --- a/frontend/src/views/settings/composables/useSettingsForm.ts +++ b/frontend/src/views/settings/composables/useSettingsForm.ts @@ -48,7 +48,9 @@ export function useSettingsForm() { openaiSessionAffinityTtlHours: null as number | null, maxWaitingPerKey: null as number | null, maxWaitingPerAccount: null as number | null, - concurrencyWaitTimeoutSeconds: null as number | null, + keyQueueTimeoutSeconds: null as number | null, + accountQueueTimeoutSeconds: null as number | null, + switchAccountOnQueueFailure: false, maxRetryAfterSeconds: null as number | null, responsesMaxDecompressedBodyMiB: null as number | null, @@ -91,7 +93,7 @@ export function useSettingsForm() { mappings.value = initial.mappings } - function numericModel(key: 'resetCreditMaxWeeklyRemainingPercent' | 'resetCreditLeadMinutes' | 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'maxAccountRotations' | 'openaiSessionAffinityTtlHours' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'maxRetryAfterSeconds' | 'responsesMaxDecompressedBodyMiB' | 'usageRetentionDays') { + function numericModel(key: 'resetCreditMaxWeeklyRemainingPercent' | 'resetCreditLeadMinutes' | 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'maxAccountRotations' | 'openaiSessionAffinityTtlHours' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'keyQueueTimeoutSeconds' | 'accountQueueTimeoutSeconds' | 'maxRetryAfterSeconds' | 'responsesMaxDecompressedBodyMiB' | 'usageRetentionDays') { return computed({ get: () => (form[key] === null ? '' @@ -121,7 +123,8 @@ export function useSettingsForm() { const resetCreditLeadMinutesValue = numericModel('resetCreditLeadMinutes') const resetCreditMaxWeeklyRemainingPercentValue = numericModel('resetCreditMaxWeeklyRemainingPercent') const usageRetentionDaysValue = numericModel('usageRetentionDays') - const concurrencyWaitTimeoutSecondsValue = numericModel('concurrencyWaitTimeoutSeconds') + const keyQueueTimeoutSecondsValue = numericModel('keyQueueTimeoutSeconds') + const accountQueueTimeoutSecondsValue = numericModel('accountQueueTimeoutSeconds') const maxRetryAfterSecondsValue = numericModel('maxRetryAfterSeconds') function applySettings(data: Awaited>) { @@ -138,7 +141,9 @@ export function useSettingsForm() { form.openaiSessionAffinityTtlHours = data.openaiSessionAffinityTtlHours form.maxWaitingPerKey = data.maxWaitingPerKey form.maxWaitingPerAccount = data.maxWaitingPerAccount - form.concurrencyWaitTimeoutSeconds = data.concurrencyWaitTimeoutSeconds + form.keyQueueTimeoutSeconds = data.keyQueueTimeoutSeconds + form.accountQueueTimeoutSeconds = data.accountQueueTimeoutSeconds + form.switchAccountOnQueueFailure = data.switchAccountOnQueueFailure form.maxRetryAfterSeconds = data.maxRetryAfterSeconds form.responsesMaxDecompressedBodyMiB = data.responsesMaxDecompressedBodyBytes / MIB @@ -238,7 +243,7 @@ export function useSettingsForm() { const savedSettings = saved.value if (saving.value || loading.value || detectingLocation.value || !savedSettings) return - const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, openaiGuardianReservedConcurrency, openaiAccountAffinity, maxAccountRotations, openaiSessionAffinityTtlHours, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, maxRetryAfterSeconds, responsesMaxDecompressedBodyMiB } = form + const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, openaiGuardianReservedConcurrency, openaiAccountAffinity, maxAccountRotations, openaiSessionAffinityTtlHours, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, keyQueueTimeoutSeconds, accountQueueTimeoutSeconds, maxRetryAfterSeconds, responsesMaxDecompressedBodyMiB } = form const { resetCreditMaxWeeklyRemainingPercent, resetCreditLeadMinutes, usageRetentionDays } = form // 关闭时保留已保存的自定义值，未完成的草稿不阻止停止覆盖。 const requestLocation = form.requestLocationEnabled @@ -267,7 +272,9 @@ export function useSettingsForm() { openaiSessionAffinityTtlHours: openaiSessionAffinityTtlHours!, maxWaitingPerKey: maxWaitingPerKey!, maxWaitingPerAccount: maxWaitingPerAccount!, - concurrencyWaitTimeoutSeconds: concurrencyWaitTimeoutSeconds!, + keyQueueTimeoutSeconds: keyQueueTimeoutSeconds!, + accountQueueTimeoutSeconds: accountQueueTimeoutSeconds!, + switchAccountOnQueueFailure: form.switchAccountOnQueueFailure, maxRetryAfterSeconds: maxRetryAfterSeconds!, responsesMaxDecompressedBodyBytes: responsesMaxDecompressedBodyMiB! * MIB, rotationStrategy: rotationStrategy as RotationStrategy, @@ -327,7 +334,8 @@ export function useSettingsForm() { openaiSessionAffinityTtlHoursValue, maxWaitingPerKeyValue, maxWaitingPerAccountValue, - concurrencyWaitTimeoutSecondsValue, + keyQueueTimeoutSecondsValue, + accountQueueTimeoutSecondsValue, maxRetryAfterSecondsValue, responsesMaxDecompressedBodyMiBValue, usageRetentionDaysValue, diff --git a/frontend/src/views/settings/composables/useSettingsFormRules.ts b/frontend/src/views/settings/composables/useSettingsFormRules.ts index fcd83c844..39e150ed7 100644 --- a/frontend/src/views/settings/composables/useSettingsFormRules.ts +++ b/frontend/src/views/settings/composables/useSettingsFormRules.ts @@ -38,7 +38,8 @@ export function useSettingsFormRules(form: () => SettingsForm) { 'openaiSessionAffinityTtlHours': integer(1, 720, t('settings.form.invalidAffinityDuration')), 'maxWaitingPerKey': integer(0, 1000, t('settings.form.invalidQueueSettings')), 'maxWaitingPerAccount': integer(0, 1000, t('settings.form.invalidQueueSettings')), - 'concurrencyWaitTimeoutSeconds': integer(1, 120, t('settings.form.invalidQueueSettings')), + 'keyQueueTimeoutSeconds': integer(1, 120, t('settings.form.invalidQueueSettings')), + 'accountQueueTimeoutSeconds': integer(1, 120, t('settings.form.invalidQueueSettings')), 'maxRetryAfterSeconds': integer(0, 4_294_967_295, t('settings.form.invalidRetryAfterLimit')), 'responsesMaxDecompressedBodyMiB': integer(1, Math.floor(Number.MAX_SAFE_INTEGER / 1024 / 1024), t('settings.form.invalidBodyLimit')), 'resetCreditLeadMinutes': integer(1, 1440, t('settings.form.invalidResetLeadTime')), diff --git a/frontend/src/views/settings/index.vue b/frontend/src/views/settings/index.vue index 045dd7769..10f5d3ca8 100644 --- a/frontend/src/views/settings/index.vue +++ b/frontend/src/views/settings/index.vue @@ -94,7 +94,8 @@ const { openaiSessionAffinityTtlHoursValue, maxWaitingPerKeyValue, maxWaitingPerAccountValue, - concurrencyWaitTimeoutSecondsValue, + keyQueueTimeoutSecondsValue, + accountQueueTimeoutSecondsValue, maxRetryAfterSecondsValue, responsesMaxDecompressedBodyMiBValue, usageRetentionDaysValue, @@ -210,7 +211,9 @@ watch(section, (value) => { v-model:max-concurrent-per-account="maxConcurrentPerAccountValue" v-model:max-waiting-per-key="maxWaitingPerKeyValue" v-model:max-waiting-per-account="maxWaitingPerAccountValue" - v-model:concurrency-wait-timeout-seconds="concurrencyWaitTimeoutSecondsValue" + v-model:key-queue-timeout-seconds="keyQueueTimeoutSecondsValue" + v-model:account-queue-timeout-seconds="accountQueueTimeoutSecondsValue" + v-model:switch-account-on-queue-failure="form.switchAccountOnQueueFailure" v-model:max-retry-after-seconds="maxRetryAfterSecondsValue" v-model:openai-guardian-reserved-concurrency="openaiGuardianReservedConcurrencyValue" />