diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index b0c015801..8d7b17c54 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -479,6 +479,7 @@ pub enum PreferredAccountSelection { pub struct AccountSelection<'a> { candidate: &'a AccountCandidate, preferred: PreferredAccountSelection, + policy_choice: bool, } impl<'a> AccountSelection<'a> { @@ -491,6 +492,12 @@ impl<'a> AccountSelection<'a> { pub const fn preferred(self) -> PreferredAccountSelection { self.preferred } + + /// 显式策略选号不受 Provider 的软亲和等待覆盖；委托内置选号不属于显式选择。 + #[must_use] + pub const fn is_policy_choice(self) -> bool { + self.policy_choice + } } /// 同一 target 内唯一的账号排序器。 @@ -536,6 +543,7 @@ impl AccountSelector { Some(AccountSelection { candidate, preferred, + policy_choice: true, }) } @@ -609,6 +617,7 @@ impl AccountSelector { return Some(AccountSelection { candidate, preferred: PreferredAccountSelection::Hit, + policy_choice: false, }); } } @@ -664,6 +673,7 @@ impl AccountSelector { Some(AccountSelection { candidate, preferred, + policy_choice: false, }) } diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index a48d2f209..e5a3120b3 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -609,7 +609,18 @@ impl CodexCredentialSelector { excluded_accounts: base_excluded.clone(), ..context.clone() }; - let wait_candidates = AccountSelector.wait_candidates(&candidates, &wait_context); + let mut wait_candidates = + AccountSelector.wait_candidates(&candidates, &wait_context); + let preferred_wait = preferred.as_ref().filter(|account_id| { + !diagnostic + && queue_policy.max_waiting > 0 + && (pinned_account.is_some() || !affinity.inherited) + && wait_candidates.contains(account_id) + }); + if let Some(preferred) = preferred_wait { + // 原绑定仍合格时只等待它；并发、请求间隔与已有队列均不应造成临时换号。 + wait_candidates.retain(|account_id| account_id == preferred); + } let capacity = AccountSelector.capacity_snapshot(&candidates, &context); for candidate in &candidates { if !waiting.can_try(candidate.account.id()) { @@ -632,6 +643,19 @@ impl CodexCredentialSelector { return Err(CredentialSelectionError::PolicyUnavailable); } }; + // 可等待集合使用原始排除集；这里的 Excluded 只可能来自本轮租约争用或队列让位。 + let selection = selection.filter(|selection| { + preferred_wait.is_none() + || selection.is_policy_choice() + || !matches!( + selection.preferred(), + PreferredAccountSelection::Blocked( + AccountSchedulingBlocker::ConcurrencyLimit + | AccountSchedulingBlocker::RequestInterval + | AccountSchedulingBlocker::Excluded + ) + ) + }); request.attempt.trace().account_selection( &candidates, &context, @@ -673,12 +697,14 @@ impl CodexCredentialSelector { let quota_exhausted = !diagnostic && statuses.next() == Some(AccountStatus::QuotaExhausted) && statuses.all(|status| status == AccountStatus::QuotaExhausted); - return match shortest_retry { - Some(retry_after) => Err(CredentialSelectionError::CapacityUnavailable { - retry_after: Some(retry_after), - }), - None if quota_exhausted => Err(CredentialSelectionError::QuotaExhausted), - None => Err(CredentialSelectionError::NoEligibleCredential), + return if !wait_candidates.is_empty() || shortest_retry.is_some() { + Err(CredentialSelectionError::CapacityUnavailable { + retry_after: shortest_retry, + }) + } else if quota_exhausted { + Err(CredentialSelectionError::QuotaExhausted) + } else { + Err(CredentialSelectionError::NoEligibleCredential) }; }; affinity.observe_preferred_selection(selection.preferred()); @@ -714,7 +740,9 @@ impl CodexCredentialSelector { .await? { ProviderLeaseAcquisition::Busy { retry_after } => { - affinity.observe_lease_busy(account.id()); + if preferred_wait != Some(account.id()) { + affinity.observe_lease_busy(account.id()); + } shortest_retry = minimum_duration(shortest_retry, retry_after); excluded.insert(account.id().clone()); } diff --git a/backend/crates/providers/openai/tests/credential/contract/capacity.rs b/backend/crates/providers/openai/tests/credential/contract/capacity.rs new file mode 100644 index 000000000..07cb11c18 --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/contract/capacity.rs @@ -0,0 +1,383 @@ +use futures::{FutureExt, future::BoxFuture}; +use gateway_core::account::AccountRuntimeSignals; +use gateway_core::concurrency::{ConcurrencyQueuePolicy, QueueRejection}; +use gateway_core::engine::policy::{ + AccountScheduleDecision, AccountScheduleInput, ModelRouteDecision, ModelRouteInput, + RequestPolicyContext, RequestPolicyFault, RequestPolicyPlan, +}; +use gateway_core::runtime::extensions::{ExtensionSetId, ExtensionSetLease, ExtensionSetReference}; + +use super::*; + +fn busy_signal(in_flight: u32, last_started_at: Option) -> AccountRuntimeSignals { + AccountRuntimeSignals { + in_flight, + last_started_at, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + } +} + +async fn bind_first(affinity: &MemorySessionAffinity) -> ProviderSessionAffinityKey { + let key = ProviderSessionAffinityKey::try_new("capacity-session").unwrap(); + affinity + .bind( + &ProviderKind::new("openai").unwrap(), + &key, + &ProviderAccountId::new("acct_first").unwrap(), + Duration::from_secs(60), + ) + .await + .unwrap(); + key +} + +fn capacity_attempt( + policy: AccountSelectionPolicy, + request_policy: Option, +) -> AttemptContext { + AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_capacity_contract").unwrap(), + ClientApiKeyId::new("key_codex_contract").unwrap(), + ) + .with_request_policy(request_policy), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(5), + policy, + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + None, + CancellationToken::new(), + ) +} + +#[test] +fn saturated_snapshot_is_capacity_unavailable_even_with_exhausted_accounts() { + for interval in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let exhausted = + block_on(store.get_account(&ProviderAccountId::new("acct_second").unwrap())) + .unwrap() + .unwrap(); + persist_quota_exhaustion(&store, &exhausted, None); + let leases = Arc::new(TestLeaseCoordinator::default()); + leases.signals.lock().unwrap().insert( + ProviderAccountId::new("acct_first").unwrap(), + busy_signal( + if interval { 0 } else { u32::MAX }, + interval.then(SystemTime::now), + ), + ); + let selector = selector(&store, leases.clone()); + let request_attempt = capacity_attempt( + AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(2).unwrap(), + Duration::from_secs(2), + ), + None, + ); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let error = block_on(selector.select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: None, + })) + .unwrap_err(); + assert!( + matches!(error, CredentialSelectionError::CapacityUnavailable { .. }), + "{error:?}" + ); + assert!(leases.requests.lock().unwrap().is_empty()); + } +} + +#[tokio::test] +async fn affinity_queue_waits_for_the_original_account_for_local_capacity_blockers() { + for blocker in ["lease", "concurrency", "interval"] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let leases = Arc::new(TestLeaseCoordinator::default()); + let first = ProviderAccountId::new("acct_first").unwrap(); + if blocker == "lease" { + leases.busy_accounts.lock().unwrap().insert(first.clone()); + } else { + leases.signals.lock().unwrap().insert( + first.clone(), + busy_signal( + if blocker == "concurrency" { 2 } else { 0 }, + (blocker == "interval").then(SystemTime::now), + ), + ); + } + let affinity = Arc::new(MemorySessionAffinity::default()); + let key = bind_first(&affinity).await; + let selector = selector_with_affinity(&store, leases.clone(), affinity.clone()); + let request_attempt = capacity_attempt( + AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(2).unwrap(), + Duration::from_secs(2), + ) + .with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), + None, + ); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: Some(&key), + }; + let mut pending = Box::pin(selector.select(&request)); + assert!( + pending.as_mut().now_or_never().is_none(), + "blocker={blocker}" + ); + leases.signals.lock().unwrap().clear(); + leases.busy_accounts.lock().unwrap().clear(); + let selected = pending.await.unwrap(); + assert_eq!(selected.account_id(), &first); + assert!(selected.affinity_hit()); + assert_eq!(selected.escape_reason(), None); + assert!(!selected.account_switch()); + assert!( + leases + .requests + .lock() + .unwrap() + .iter() + .all(|request| request.account_id() == &first) + ); + } +} + +#[tokio::test] +async fn affinity_queue_full_and_timeout_preserve_binding_despite_free_fallback() { + 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 leases = Arc::new(TestLeaseCoordinator::default()); + leases.busy_accounts.lock().unwrap().insert(first.clone()); + let affinity = Arc::new(MemorySessionAffinity::default()); + let key = bind_first(&affinity).await; + let selector = selector_with_affinity(&store, leases.clone(), affinity.clone()); + let head_attempt = queued_attempt(Duration::from_secs(2)); + let next_attempt = queued_attempt(Duration::from_secs(2)); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &head_attempt, + session_affinity_key: Some(&key), + }; + let mut head = Box::pin(selector.select(&request)); + assert!(head.as_mut().now_or_never().is_none()); + // 槽位已空，但新请求不能越过旧等待者，也不能转而抢占另一账号。 + leases.busy_accounts.lock().unwrap().clear(); + let rejected = selector + .select(&SelectCodexCredential { + attempt: &next_attempt, + ..request + }) + .await + .unwrap_err(); + assert!(matches!( + rejected, + CredentialSelectionError::QueueRejected(QueueRejection::Full) + )); + drop(head); + + leases.busy_accounts.lock().unwrap().insert(first.clone()); + let timeout_attempt = queued_attempt(Duration::from_millis(20)); + let timeout = selector + .select(&SelectCodexCredential { + attempt: &timeout_attempt, + ..request + }) + .await + .unwrap_err(); + assert!(matches!( + timeout, + CredentialSelectionError::QueueRejected(QueueRejection::Timeout) + )); + assert_eq!( + affinity + .load(&ProviderKind::new("openai").unwrap(), &key) + .await + .unwrap(), + Some(first.clone()) + ); + leases.busy_accounts.lock().unwrap().clear(); + let selected = selector + .select(&SelectCodexCredential { + attempt: &next_attempt, + ..request + }) + .await + .unwrap(); + assert_eq!(selected.account_id(), &first); + assert!( + leases + .requests + .lock() + .unwrap() + .iter() + .all(|request| request.account_id() == &first) + ); +} + +#[tokio::test] +async fn affinity_queue_rechecks_hard_unavailability_before_switching_accounts() { + for quota_exhausted in [false, true] { + 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 leases = Arc::new(TestLeaseCoordinator::default()); + leases.busy_accounts.lock().unwrap().insert(first.clone()); + let affinity = Arc::new(MemorySessionAffinity::default()); + let key = bind_first(&affinity).await; + let selector = selector_with_affinity(&store, leases, affinity); + let request_attempt = queued_attempt(Duration::from_secs(2)); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: Some(&key), + }; + let mut pending = Box::pin(selector.select(&request)); + assert!(pending.as_mut().now_or_never().is_none()); + let account = store.get_account(&first).await.unwrap().unwrap(); + if quota_exhausted { + persist_quota_exhaustion(&store, &account, None); + } else { + persist_credential_state(&store, &account, CredentialState::Expired); + } + let selected = pending.await.unwrap(); + assert_eq!(selected.account_id().as_str(), "acct_second"); + assert!(selected.account_switch()); + } +} + +#[tokio::test] +async fn queue_enabled_without_affinity_still_uses_a_free_account() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let leases = Arc::new(TestLeaseCoordinator::default()); + leases.signals.lock().unwrap().insert( + ProviderAccountId::new("acct_first").unwrap(), + busy_signal(2, None), + ); + let selector = selector(&store, leases); + let request_attempt = queued_attempt(Duration::from_secs(2)); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let selected = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: None, + }) + .await + .unwrap(); + assert_eq!(selected.account_id().as_str(), "acct_second"); +} + +#[derive(Debug)] +struct Scheduler { + explicit: bool, +} + +impl RequestPolicyPlan for Scheduler { + fn route_model( + &self, + _: ModelRouteInput, + ) -> BoxFuture<'static, Result> { + Box::pin(async { Ok(ModelRouteDecision::Unhandled) }) + } + + fn schedule_account( + &self, + _: AccountScheduleInput, + ) -> BoxFuture<'static, Result> { + let decision = if self.explicit { + AccountScheduleDecision::Pick(ProviderAccountId::new("acct_second").unwrap()) + } else { + AccountScheduleDecision::Delegate + }; + Box::pin(async move { Ok(decision) }) + } +} + +struct ExtensionLease; + +impl ExtensionSetLease for ExtensionLease { + fn is_ready(&self) -> bool { + true + } +} + +#[tokio::test] +async fn affinity_wait_respects_explicit_policy_choices_but_not_delegated_fallbacks() { + for explicit in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let leases = Arc::new(TestLeaseCoordinator::default()); + leases.signals.lock().unwrap().insert( + ProviderAccountId::new("acct_first").unwrap(), + busy_signal(2, None), + ); + let affinity = Arc::new(MemorySessionAffinity::default()); + let key = bind_first(&affinity).await; + let selector = selector_with_affinity(&store, leases.clone(), affinity); + let request_attempt = capacity_attempt( + account_policy().with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), + Some(RequestPolicyContext::new( + Arc::new(Scheduler { explicit }), + ExtensionSetReference::new( + ExtensionSetId::new("capacity-test".into()).unwrap(), + Arc::new(ExtensionLease), + ), + ModelRequestId::new("req_capacity_contract").unwrap(), + ClientApiKeyId::new("key_codex_contract").unwrap(), + vec![], + )), + ); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: Some(&key), + }; + let mut pending = Box::pin(selector.select(&request)); + if explicit { + let selected = pending.as_mut().now_or_never().unwrap().unwrap(); + assert_eq!(selected.account_id().as_str(), "acct_second"); + } else { + assert!(pending.as_mut().now_or_never().is_none()); + leases.signals.lock().unwrap().clear(); + let selected = pending.await.unwrap(); + assert_eq!(selected.account_id().as_str(), "acct_first"); + } + } +} diff --git a/backend/crates/providers/openai/tests/credential/contract.rs b/backend/crates/providers/openai/tests/credential/contract/mod.rs similarity index 99% rename from backend/crates/providers/openai/tests/credential/contract.rs rename to backend/crates/providers/openai/tests/credential/contract/mod.rs index 05ee8c584..8bb47dda3 100644 --- a/backend/crates/providers/openai/tests/credential/contract.rs +++ b/backend/crates/providers/openai/tests/credential/contract/mod.rs @@ -1,3 +1,5 @@ +mod capacity; + use std::collections::{BTreeMap, BTreeSet}; use std::num::NonZeroU32; use std::sync::Arc; diff --git a/backend/crates/providers/openai/tests/provider/contract/capacity.rs b/backend/crates/providers/openai/tests/provider/contract/capacity.rs new file mode 100644 index 000000000..9ebe5ba8c --- /dev/null +++ b/backend/crates/providers/openai/tests/provider/contract/capacity.rs @@ -0,0 +1,131 @@ +use futures::FutureExt; +use gateway_core::concurrency::ConcurrencyQueuePolicy; + +use super::*; + +fn queued_context(request_id: &str) -> AttemptContext { + AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new(request_id).unwrap(), + ClientApiKeyId::new("key_openai_contract").unwrap(), + ), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(5), + account_policy().with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + None, + CancellationToken::new(), + ) +} + +fn operation(thread_id: &str) -> Operation { + Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!("capacity queue test")), + ("session_id".to_owned(), json!("capacity-root")), + ("thread_id".to_owned(), json!(thread_id)), + ]), + ) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + )) +} + +#[tokio::test] +async fn queued_session_sends_only_after_capacity_recovers_while_a_new_child_can_use_another_account() + { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let leases = Arc::new(TestLeaseCoordinator::default()); + let affinity = Arc::new(MemorySessionAffinity::default()); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(3) + .mount(&server) + .await; + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + affinity, + server.uri(), + leases.clone(), + ); + let mut root = provider + .clone() + .execute( + planned_request("openai", operation("capacity-root")), + queued_context("req_capacity_root"), + ) + .await + .unwrap(); + while let Some(event) = root.next().await { + event.unwrap(); + } + drop(root); + + create_account(&store, "acct_subagent_b").await; + leases + .busy_accounts + .lock() + .unwrap() + .insert(ProviderAccountId::new("acct_subagent_a").unwrap()); + let mut waiting = Box::pin(provider.clone().execute( + planned_request("openai", operation("capacity-root")), + queued_context("req_capacity_wait"), + )); + assert!(waiting.as_mut().now_or_never().is_none()); + + // 子线程还没有自己的绑定，父账号只是默认偏好，不能被强制挤进父账号队列。 + let mut child = provider + .execute( + planned_request("openai", operation("capacity-child")), + queued_context("req_capacity_child"), + ) + .await + .unwrap(); + assert_eq!( + child.metadata().provider_account_id().as_str(), + "acct_subagent_b" + ); + while let Some(event) = child.next().await { + event.unwrap(); + } + drop(child); + assert_eq!(server.received_requests().await.unwrap().len(), 2); + + leases.busy_accounts.lock().unwrap().clear(); + let mut resumed = waiting.await.unwrap(); + assert_eq!( + resumed.metadata().provider_account_id().as_str(), + "acct_subagent_a" + ); + while let Some(event) = resumed.next().await { + event.unwrap(); + } + let requests = server.received_requests().await.unwrap(); + let accounts = requests + .iter() + .map(|request| request.headers["chatgpt-account-id"].to_str().unwrap()) + .collect::>(); + assert_eq!( + accounts, + [ + "chatgpt-acct_subagent_a", + "chatgpt-acct_subagent_b", + "chatgpt-acct_subagent_a" + ] + ); + server.verify().await; +} diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs similarity index 99% rename from backend/crates/providers/openai/tests/provider/contract.rs rename to backend/crates/providers/openai/tests/provider/contract/mod.rs index cd77ac885..6346228a3 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -1,3 +1,5 @@ +mod capacity; + use std::collections::{BTreeMap, BTreeSet}; use std::io::{self, Write}; use std::num::NonZeroU32; @@ -698,7 +700,7 @@ async fn selected_proxy_location_overrides_global_and_reloads_without_mutating_c } const OFFICIAL_FIXTURE: &[u8] = - include_bytes!("../transport/fixtures/official_models_snapshot.json"); + include_bytes!("../../transport/fixtures/official_models_snapshot.json"); #[tokio::test] async fn replay_compatibility_should_remove_only_reasoning_status_on_both_transports() { @@ -3411,30 +3413,47 @@ async fn standalone_search_returns_the_exact_upstream_error_response() { #[tokio::test] async fn capacity_selection_error_preserves_classification_and_retry_after() { - let store = Arc::new(MemoryAccountStore::default()); - create_account(&store, "acct_capacity_busy").await; - let leases = Arc::new(TestLeaseCoordinator::default()); - *leases.busy.lock().expect("lease busy lock") = true; + for snapshot in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_capacity_busy").await; + let leases = Arc::new(TestLeaseCoordinator::default()); + if snapshot { + leases.signals.lock().unwrap().insert( + ProviderAccountId::new("acct_capacity_busy").unwrap(), + gateway_core::account::AccountRuntimeSignals { + in_flight: u32::MAX, + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + }, + ); + } else { + *leases.busy.lock().unwrap() = true; + } - let error = match provider_with_leases(&store, leases) - .execute( - planned_request("openai", generate_operation()), - context("req_capacity_busy", CancellationToken::new()), - ) - .await - { - Ok(_) => panic!("busy account selection must fail"), - Err(error) => error, - }; + let error = match provider_with_leases(&store, leases) + .execute( + planned_request("openai", generate_operation()), + context("req_capacity_busy", CancellationToken::new()), + ) + .await + { + Ok(_) => panic!("busy account selection must fail"), + Err(error) => error, + }; - assert_eq!( - (error.kind(), error.send_state(), error.retry_after()), - ( - ProviderErrorKind::AccountCapacityUnavailable, - UpstreamSendState::NotSent, - Some(Duration::from_millis(25)), - ) - ); + assert_eq!( + (error.kind(), error.send_state(), error.retry_after()), + ( + ProviderErrorKind::AccountCapacityUnavailable, + UpstreamSendState::NotSent, + (!snapshot).then_some(Duration::from_millis(25)), + ) + ); + } } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index e0e1244d9..4601576fa 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -655,6 +655,7 @@ pub(crate) struct TestLeaseCoordinator { pub(crate) requests: Mutex>, pub(crate) busy: Mutex, pub(crate) busy_accounts: Mutex>, + pub(crate) signals: Mutex>, round_robin_cursor: Mutex, } @@ -666,21 +667,24 @@ impl ProviderLeasePort for TestLeaseCoordinator { accounts: &'a [ProviderAccountId], ) -> BoxFuture<'a, Result> { Box::pin(async move { + let overrides = self.signals.lock().expect("scheduling signals lock"); let signals = accounts .iter() - .cloned() .map(|account| { ( - account, - AccountRuntimeSignals { - in_flight: 0, - last_started_at: None, - quota_reset_at: None, - quota_remaining_rank: None, - cooldown: None, - failure_rate_basis_points: None, - first_output_latency_ms: None, - }, + account.clone(), + overrides + .get(account) + .cloned() + .unwrap_or(AccountRuntimeSignals { + in_flight: 0, + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + }), ) }) .collect(); diff --git a/docs/api.md b/docs/api.md index 8d6e86d83..7fac78bd9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1116,8 +1116,12 @@ HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造 `response.create`；空闲连接不占名额，内部重试不重复占用。 修改 Key 策略对既有 WebSocket 连接的下一次请求同样生效，已开始的请求保持原有快照。 -运行设置可以分别启用 Key 与账号的有界排队。Key 并发满时按 Key 等待；账号先使用其它可调度候选， -适用账号均暂时满载后按账号等待。RPM、金额限额、失效账号和上游冷却不通过排队绕过。 +运行设置可以分别启用 Key 与账号的有界排队。Key 并发满时按 Key 等待；OpenAI 已绑定的会话账号仅因 +本地并发、请求间隔或已有等待者暂忙时，开启账号排队后优先等待原账号，队列满或超时不因此迁移绑定。 +原账号失效、额度耗尽或进入上游冷却时重新选择；显式调度策略选号与智能调度高权重回切仍按各自规则执行。 +没有适用亲和时先使用其它可调度候选，适用账号均暂时满载后按账号等待。 +关闭账号排队时，OpenAI 本地容量不足返回 `503` / `account_capacity_unavailable`， +与无候选账号的 `no_available_provider` 区分。RPM、金额限额、失效账号和上游冷却不通过排队绕过。 队列满返回 `429` / `concurrency_queue_full`，排队超时返回 `429` / `concurrency_queue_timeout`； WebSocket 使用对应错误事件。等待期间不发送上游请求，取消后释放等待位置，排队重查不重复计入 RPM。 SSE 在取得有效执行前不发送保活帧，因此此阶段保留 HTTP 错误状态；已开始交付的失败沿用流内错误合同。 diff --git a/docs/architecture.md b/docs/architecture.md index e7e7b800e..f1e42cbe5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -494,6 +494,9 @@ Continuation 仍受原请求的 Client Key、账号范围、Provider 和发送/ - scope 外账号、跨 Key 复用或不明确发送结果均 fail closed。 会话亲和是优先选择提示，不是硬账号绑定；native continuation 才携带不可跨越的 owner 约束。 +OpenAI 启用账号排队时，可等待的已有亲和绑定优先保留；子线程首绑前继承的根偏好仍允许分流。 +快照满载、原子租约争用与已有等待者均进入原账号队列，等待重查仍执行完整资格校验。 +显式策略选号和内置高权重回切保留各自裁决；亲和等待不将暂忙升级成失效，也不覆盖原生续写约束。 ### 并发等待