feat(scheduling): split queue timeouts and allow account fallback (#625)

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
This commit is contained in:
Zyy
2026-11-06 23:09:29 +08:00
committed by GitHub
co-authored by Zyy
parent 1240f22580
commit ee7e837d1f
43 changed files with 622 additions and 167 deletions
@@ -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,
)?;
@@ -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,
@@ -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,
@@ -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,
@@ -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,
@@ -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<gateway_admin::model::settings::RuntimeSettingsValues> 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",
@@ -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,
@@ -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::<UpdateRuntimeSettingsRequest>(body)
.unwrap()
.validate()
@@ -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;
}),
@@ -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<QueueRejection>,
F: Future<Output = Result<T, E>>,
>(
self,
rejection: QueueRejection,
remaining: Option<Duration>,
select: impl FnOnce(ConcurrencyQueuePolicy) -> F,
no_account_available: impl FnOnce(&E) -> bool,
) -> Result<T, E> {
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;
@@ -19,7 +19,7 @@ pub struct ConcurrencyQueuePolicy {
pub timeout: Duration,
}
/// 从首次入队开始计时,密钥、账号与后续重试共享同一个截止时刻
/// 从首次入队开始计时,同一层的重新排队与重试共享截止时刻
#[derive(Debug, Clone, Default)]
pub struct ConcurrencyWaitBudget {
deadline: Arc<OnceLock<Instant>>,
@@ -600,12 +600,6 @@ impl<S: ?Sized> ResponseExecutionSession<S>
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()
@@ -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<ActiveRequestLease>,
pub(super) concurrency_wait_budget: ConcurrencyWaitBudget,
pub(super) admission_decision_ms: Option<u64>,
}
@@ -724,7 +724,6 @@ impl DefaultExecutionService {
request_id: &ModelRequestId,
authorization: &mut AuthorizedExecution,
) -> Result<ExecutionStartGuard, GatewayError> {
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<AdmissionLease, GatewayError> {
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(),
@@ -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
@@ -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
@@ -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,
@@ -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 =
@@ -4390,10 +4390,7 @@ impl Provider for QueuedAccountProvider {
) -> Result<ProviderStream, ProviderError> {
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());
@@ -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<RuntimeSettings> 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,
@@ -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(),
@@ -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<Utc>,
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<RuntimeSett
min_codex_cli_version: row.min_codex_cli_version,
max_waiting_per_key: to_u32(row.max_waiting_per_key)?,
max_waiting_per_account: to_u32(row.max_waiting_per_account)?,
concurrency_wait_timeout_seconds: to_u32(row.concurrency_wait_timeout_seconds)?,
key_queue_timeout_seconds: to_u32(row.key_queue_timeout_seconds)?,
account_queue_timeout_seconds: to_u32(row.account_queue_timeout_seconds)?,
switch_account_on_queue_failure: row.switch_account_on_queue_failure,
max_retry_after_seconds: to_u32(row.max_retry_after_seconds)?,
openai_guardian_reserved_concurrency: to_u32(row.openai_guardian_reserved_concurrency)?,
openai_account_affinity: gateway_core::account::AccountAffinity::parse(
@@ -227,7 +227,9 @@ struct SnapshotSettingsRow {
min_codex_cli_version: Option<String>,
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,
),
))
}
@@ -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;
@@ -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<CodexCredentialLease, CredentialSelectionError> {
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<CodexCredentialLease, CredentialSelectionError> {
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 {
@@ -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 [
@@ -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,
@@ -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<SelectedGrokSession, GrokSessionSelectorError> {
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<SelectedGrokSession, GrokSessionSelectorError> {
let mut waiting = CapacityWait::new(
&self.waiting,
queue_policy,
@@ -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;
+1
View File
@@ -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
@@ -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;
+9 -6
View File
@@ -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 时,
+2 -2
View File
@@ -848,8 +848,8 @@ OpenAI 内置调度仅在排队目标与固定归属或严格会话绑定一致
Key 的 RPM 在成功准入时才计数,金额限制在入队前及成功准入后检查。
普通账号容量等待只包含并发;严格亲和下的后代线程还可等待根线程恢复或迁移绑定,但等待不使权限、额度或冷却检查失效。
密钥准入、账号选择与后续重试共享请求级等待预算,从首次入队开始计时;显式设置请求总时限时也受其约束。
切换等待层、账号或 Provider 不重置等待时限;没有发生排队时不启动该计时,也不据此中断已开始的上游生成。
密钥准入与账号选择各自从首次入队开始计时,账号重选、Provider 切换和后续重试共享账号等待预算,两层均受请求总期限约束。
账号队满或超时释放队位后,可按设置对允许换号的请求再选一次可立即准入的账号。各 Provider 仍检查原有资格、租约和会话归属;OpenAI 严格亲和在取得租约后以当前绑定版本提交迁移。
队列满/超时属于本地容量拒绝,不作为上游限流反馈或 Provider 故障熔断证据
#### Guardian 独立并发
+3 -1
View File
@@ -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
+16 -7
View File
@@ -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",
@@ -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": "冷却时长秒数",
@@ -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": "冷卻時長秒數",
@@ -1,8 +1,8 @@
<script setup lang="ts">
import { ZCard, ZFormItem, ZIconButton, ZInput, ZPopover } from '@codex-proxy/ui'
import { ZCard, ZFormItem, ZIconButton, ZInput, ZPopover, ZSegmented } from '@codex-proxy/ui'
import { CircleAlert } from '@lucide/vue'
import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
const { t } = useI18n({ useScope: 'global' })
@@ -10,9 +10,21 @@ const { t } = useI18n({ useScope: 'global' })
const maxConcurrentPerAccount = defineModel<string>('maxConcurrentPerAccount', { required: true })
const maxWaitingPerKey = defineModel<string>('maxWaitingPerKey', { required: true })
const maxWaitingPerAccount = defineModel<string>('maxWaitingPerAccount', { required: true })
const concurrencyWaitTimeoutSeconds = defineModel<string>('concurrencyWaitTimeoutSeconds', { required: true })
const keyQueueTimeoutSeconds = defineModel<string>('keyQueueTimeoutSeconds', { required: true })
const accountQueueTimeoutSeconds = defineModel<string>('accountQueueTimeoutSeconds', { required: true })
const switchAccountOnQueueFailure = defineModel<boolean>('switchAccountOnQueueFailure', { required: true })
const openaiGuardianReservedConcurrency = defineModel<string>('openaiGuardianReservedConcurrency', { required: true })
const maxRetryAfterSeconds = defineModel<string>('maxRetryAfterSeconds', { required: true })
const queueFailureOptions = computed(() => [
{
value: 'reject',
label: t('settings.concurrency.returnError'),
},
{
value: 'switch',
label: t('settings.concurrency.trySwitch'),
},
])
</script>
<template>
@@ -34,27 +46,6 @@ const maxRetryAfterSeconds = defineModel<string>('maxRetryAfterSeconds', { requi
/>
</ZFormItem>
<ZFormItem :show-message="Boolean(maxWaitingPerKey.trim())" prop="maxWaitingPerKey" :label="t('settings.concurrency.keyQueueCapacity')" :description="t('settings.concurrency.keyQueueHelp')">
<ZInput v-model="maxWaitingPerKey" :aria-label="t('settings.concurrency.keyQueueCapacity')" type="number" min="0" max="1000" step="1" />
</ZFormItem>
<ZFormItem :show-message="Boolean(maxWaitingPerAccount.trim())" prop="maxWaitingPerAccount" :label="t('settings.concurrency.accountQueueCapacity')" :description="t('settings.concurrency.accountQueueHelp')">
<ZInput v-model="maxWaitingPerAccount" :aria-label="t('settings.concurrency.accountQueueCapacity')" type="number" min="0" max="1000" step="1" />
</ZFormItem>
<ZFormItem :show-message="Boolean(concurrencyWaitTimeoutSeconds.trim())" prop="concurrencyWaitTimeoutSeconds" :label="t('settings.concurrency.queueTimeoutLabel')" :description="t('settings.concurrency.queueTimeoutHelp')">
<template #label-extra>
<ZPopover class="-my-1" trigger="hover-click" placement="top-start">
<template #reference="{ open }">
<ZIconButton variant="link" :aria-label="t('settings.concurrency.queueTimeoutHelpTitle')" :title="undefined" :aria-expanded="open" class="size-6!">
<CircleAlert class="size-3.5" aria-hidden="true" />
</ZIconButton>
</template>
<p class="max-w-72 px-3 py-2 text-z-sm leading-relaxed text-z-text-secondary">
{{ t('settings.concurrency.queueTimeoutDetails') }}
</p>
</ZPopover>
</template>
<ZInput v-model="concurrencyWaitTimeoutSeconds" :aria-label="t('settings.concurrency.queueTimeoutLabel')" type="number" min="1" max="120" step="1" />
</ZFormItem>
<ZFormItem :show-message="Boolean(openaiGuardianReservedConcurrency.trim())" prop="openaiGuardianReservedConcurrency" :label="t('settings.concurrency.autoReview.label')" :description="t('settings.concurrency.autoReview.description')">
<template #label-extra>
<ZPopover class="-my-1" trigger="hover-click" placement="top-start">
@@ -71,6 +62,19 @@ const maxRetryAfterSeconds = defineModel<string>('maxRetryAfterSeconds', { requi
</template>
<ZInput v-model="openaiGuardianReservedConcurrency" :aria-label="t('settings.concurrency.autoReview.label')" type="number" min="0" max="4294967295" step="1" />
</ZFormItem>
<ZFormItem :show-message="Boolean(maxWaitingPerKey.trim())" prop="maxWaitingPerKey" :label="t('settings.concurrency.keyQueueCapacity')" :description="t('settings.concurrency.keyQueueHelp')">
<ZInput v-model="maxWaitingPerKey" :aria-label="t('settings.concurrency.keyQueueCapacity')" type="number" min="0" max="1000" step="1" />
</ZFormItem>
<ZFormItem :show-message="Boolean(keyQueueTimeoutSeconds.trim())" prop="keyQueueTimeoutSeconds" :label="t('settings.concurrency.keyTimeout')" :description="t('settings.concurrency.keyTimeoutHelp')">
<ZInput v-model="keyQueueTimeoutSeconds" :aria-label="t('settings.units.inSeconds', { label: t('settings.concurrency.keyTimeout') })" type="number" min="1" max="120" step="1">
<template #suffix>
{{ t('settings.units.seconds') }}
</template>
</ZInput>
</ZFormItem>
<ZFormItem :show-message="Boolean(maxWaitingPerAccount.trim())" prop="maxWaitingPerAccount" :label="t('settings.concurrency.accountQueueCapacity')" :description="t('settings.concurrency.accountQueueHelp')">
<ZInput v-model="maxWaitingPerAccount" :aria-label="t('settings.concurrency.accountQueueCapacity')" type="number" min="0" max="1000" step="1" />
</ZFormItem>
<ZFormItem :show-message="Boolean(maxRetryAfterSeconds.trim())" prop="maxRetryAfterSeconds" :label="t('settings.concurrency.retryAfter.label')" :description="t('settings.concurrency.retryAfter.help')">
<template #label-extra>
<ZPopover class="-my-1" trigger="hover-click" placement="top-start">
@@ -84,7 +88,38 @@ const maxRetryAfterSeconds = defineModel<string>('maxRetryAfterSeconds', { requi
</p>
</ZPopover>
</template>
<ZInput v-model="maxRetryAfterSeconds" :aria-label="t('settings.concurrency.retryAfter.label')" type="number" min="0" max="4294967295" step="1" />
<ZInput v-model="maxRetryAfterSeconds" :aria-label="t('settings.units.inSeconds', { label: t('settings.concurrency.retryAfter.label') })" type="number" min="0" max="4294967295" step="1">
<template #suffix>
{{ t('settings.units.seconds') }}
</template>
</ZInput>
</ZFormItem>
<ZFormItem :show-message="Boolean(accountQueueTimeoutSeconds.trim())" prop="accountQueueTimeoutSeconds" :label="t('settings.concurrency.accountTimeout')" :description="t('settings.concurrency.accountTimeoutHelp')">
<ZInput v-model="accountQueueTimeoutSeconds" :aria-label="t('settings.units.inSeconds', { label: t('settings.concurrency.accountTimeout') })" type="number" min="1" max="120" step="1">
<template #suffix>
{{ t('settings.units.seconds') }}
</template>
</ZInput>
</ZFormItem>
<ZFormItem :label="t('settings.concurrency.queueFailure')" :description="t('settings.concurrency.queueFailureHelp')">
<template #label-extra>
<ZPopover class="-my-1" trigger="hover-click" placement="top-start">
<template #reference="{ open }">
<ZIconButton variant="link" :aria-label="t('settings.concurrency.queueFailureHelpTitle')" :title="undefined" :aria-expanded="open" class="size-6!">
<CircleAlert class="size-3.5" aria-hidden="true" />
</ZIconButton>
</template>
<p class="max-w-72 px-3 py-2 text-z-sm leading-relaxed text-z-text-secondary">
{{ t('settings.concurrency.queueFailureDetails') }}
</p>
</ZPopover>
</template>
<ZSegmented
:model-value="switchAccountOnQueueFailure ? 'switch' : 'reject'"
:aria-label="t('settings.concurrency.queueFailure')"
:options="queueFailureOptions"
@update:model-value="switchAccountOnQueueFailure = $event === 'switch'"
/>
</ZFormItem>
</div>
</ZCard>
@@ -14,10 +14,13 @@ const refreshConcurrency = defineModel<string>('refreshConcurrency', { required:
<ZCard :title="t('settings.tokenRefresh.title')">
<div class="grid min-w-0 gap-4.5 max-w-6xl sm:grid-cols-2">
<ZFormItem :show-message="Boolean(refreshMarginSeconds.trim())" prop="refreshMarginSeconds" :label="t('settings.tokenRefresh.leadTimeLabel')" :description="t('settings.tokenRefresh.leadTimeHelp')">
<ZInput v-model="refreshMarginSeconds" :aria-label="t('settings.tokenRefresh.leadTimeLabel')" type="number">
<ZInput v-model="refreshMarginSeconds" :aria-label="t('settings.units.inSeconds', { label: t('settings.tokenRefresh.leadTimeLabel') })" type="number">
<template #prefix>
<Timer class="size-4" />
</template>
<template #suffix>
{{ t('settings.units.seconds') }}
</template>
</ZInput>
</ZFormItem>
<ZFormItem :show-message="Boolean(refreshConcurrency.trim())" prop="refreshConcurrency" :label="t('settings.tokenRefresh.concurrencyLabel')" :description="t('settings.tokenRefresh.concurrencyHelp')">
@@ -193,7 +193,7 @@ function removeRule(id: string) {
<Timer class="size-4" />
</template>
<template #suffix>
<span class="text-z-sm">{{ t('settings.errorProtection.seconds') }}</span>
<span class="text-z-sm">{{ t('settings.units.seconds') }}</span>
</template>
</ZInput>
</ZFormItem>
@@ -216,7 +216,7 @@ function removeRule(id: string) {
<Snowflake class="size-4" />
</template>
<template #suffix>
<span class="text-z-sm">{{ t('settings.errorProtection.seconds') }}</span>
<span class="text-z-sm">{{ t('settings.units.seconds') }}</span>
</template>
</ZInput>
</ZFormItem>
@@ -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<ReturnType<typeof getSettings>>) {
@@ -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,
@@ -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')),
+5 -2
View File
@@ -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"
/>