diff --git a/backend/crates/providers/openai/src/credential/repository.rs b/backend/crates/providers/openai/src/credential/repository.rs index 7b690f8e5..f4a554ba1 100644 --- a/backend/crates/providers/openai/src/credential/repository.rs +++ b/backend/crates/providers/openai/src/credential/repository.rs @@ -180,6 +180,28 @@ impl CodexCredentialRepository { self.decode_runtime_credential(&loaded) } + /// 选号专用的凭据重读,返回最新账号快照 + /// + /// 成功推理会持续刷新额度观测时刻;未耗尽账号的访问结论不变时调度资格不变, + /// 沿用最新快照,避免繁忙账号在选号阶段反复冲突 + pub async fn reload_selected_credential( + &self, + account: &ProviderAccount, + ) -> Result<(ProviderAccount, CodexRuntimeCredential), CredentialRepositoryError> { + if account.provider().as_str() != PROVIDER_NAME { + return Err(CredentialRepositoryError::InvalidCredentialData(None)); + } + let loaded = self + .store + .load_credential(account.id(), account.revision()) + .await?; + if !only_quota_observation_advanced(account, &loaded.account) { + return Err(CredentialRepositoryError::RevisionConflict(None)); + } + let runtime = self.decode_runtime_credential(&loaded)?; + Ok((loaded.account, runtime)) + } + pub async fn load_complete_data( &self, account: &ProviderAccount, @@ -320,3 +342,27 @@ fn oauth_account_state(account: &ProviderAccount, observed: CredentialState) -> observed } } + +fn only_quota_observation_advanced(selected: &ProviderAccount, current: &ProviderAccount) -> bool { + if selected == current { + return true; + } + let (before, after) = (selected.quota(), current.quota()); + // 耗尽状态的复核时间依赖观测时刻,必须按冲突处理 + if before.is_exhausted() + || before.access() != after.access() + || before.evidence() != after.evidence() + || before.reset_at() != after.reset_at() + { + return false; + } + // 只替换额度观测,其余账号事实必须与选号时完全一致 + let normalized = current.clone().with_account_facts( + current.enabled(), + current.credential_state(), + before, + current.last_error_reason(), + current.last_error_message().map(str::to_owned), + ); + normalized == *selected +} diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index f70de56f0..c40036023 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -624,8 +624,8 @@ impl CodexCredentialSelector { if request.requires_websocket && pinned_account.as_ref().is_none_or(|id| id == account.id()) { - let runtime = match self.repository.load_runtime_credential(&account).await { - Ok(runtime) => runtime, + let runtime = match self.repository.reload_selected_credential(&account).await { + Ok((_, runtime)) => runtime, Err(CredentialRepositoryError::RevisionConflict(_)) => { retry_account_snapshot( request.attempt, @@ -907,10 +907,15 @@ impl CodexCredentialSelector { .find(|candidate| candidate.account.id() == selected.account.id()) .map(|candidate| candidate.account.clone()) .ok_or(CredentialSelectionError::InvalidCredential(None))?; - // 额度观测等并发更新会使整个账号快照失效,必须重新选号并校验资格 + // 凭据、状态或额度结论的并发更新会使账号快照失效,必须重新选号并校验资格; + // 仅额度观测时刻前进时沿用最新快照 // 在占用租约和请求间隔前完成校验,避免重读被自己的异步释放挡住 - let runtime = match self.repository.load_runtime_credential(&account).await { - Ok(runtime) => runtime, + let (account, runtime) = match self + .repository + .reload_selected_credential(&account) + .await + { + Ok(reloaded) => reloaded, Err(CredentialRepositoryError::RevisionConflict(_)) => { retry_account_snapshot(request.attempt, &account, &mut snapshot_retries)?; continue 'capacity; diff --git a/backend/crates/providers/openai/tests/credential/contract/mod.rs b/backend/crates/providers/openai/tests/credential/contract/mod.rs index 28bfabc25..b2c0a123d 100644 --- a/backend/crates/providers/openai/tests/credential/contract/mod.rs +++ b/backend/crates/providers/openai/tests/credential/contract/mod.rs @@ -537,7 +537,11 @@ fn selector_reloads_quota_snapshot_before_acquiring_the_only_lease() { selected.account(), &store.account("acct_primary").expect("current account") ); - assert_eq!(store.credential_loads(), 2); + assert_eq!( + store.credential_loads(), + 1, + "observation-only quota updates must not restart selection" + ); let requests = leases.requests.lock().expect("lease requests"); assert_eq!( requests.len(), @@ -548,6 +552,52 @@ fn selector_reloads_quota_snapshot_before_acquiring_the_only_lease() { assert_eq!(requests[0].request_interval(), Duration::from_millis(10)); } +#[test] +fn selector_rejects_quota_exhaustion_observed_after_selection() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_primary", "at-primary"); + let original = store.account("acct_primary").expect("account"); + block_on(store.apply_quota_access(QuotaAccessChange { + account_id: original.id().clone(), + expected_revision: original.revision(), + state: QuotaState::allowed(SystemTime::now()), + })) + .expect("initial quota observation"); + store.on_credential_load(Arc::new(|store, id, count| { + Box::pin(async move { + if count == 1 { + let account = store.account(id.as_str()).expect("account"); + store + .apply_quota_access(QuotaAccessChange { + account_id: id.clone(), + expected_revision: account.revision(), + state: QuotaState::exhausted( + QuotaEvidence::UsageLimitReached, + SystemTime::now(), + None, + ), + }) + .await?; + } + Ok(()) + }) + })); + let leases = Arc::new(TestLeaseCoordinator::default()); + let selector = selector(&store, Arc::clone(&leases)); + let attempt = attempt(BTreeSet::new()); + let error = block_on(selector.select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &Url::parse("https://chatgpt.com/backend-api/codex/responses").expect("URL"), + attempt: &attempt, + session_affinity_key: None, + })) + .expect_err("quota exhaustion must still invalidate the selected snapshot"); + + assert!(matches!(error, CredentialSelectionError::QuotaExhausted)); + assert_eq!(store.credential_loads(), 1); + assert!(leases.requests.lock().expect("lease requests").is_empty()); +} + #[test] fn selector_uses_rotated_credentials_after_a_snapshot_conflict() { use gateway_core::account::{CredentialCasOutcome, CredentialCasUpdate, ProviderAccountUpdate}; diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 069fbc55c..3a69eb618 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -2329,7 +2329,11 @@ async fn quota_snapshot_race_reloads_and_sends_the_request_once() { .await .expect("concurrent quota observation must not fail prepare"); assert_eq!(stream.metadata().provider_account_id(), original.id()); - assert_eq!(store.credential_loads(), 2); + assert_eq!( + store.credential_loads(), + 1, + "observation-only quota updates must not restart selection" + ); assert_eq!(leases.requests.lock().expect("leases").len(), 1); let mut completed = false; while let Some(event) = stream.next().await { @@ -2476,19 +2480,7 @@ async fn repeated_snapshot_conflicts_are_bounded_and_report_the_selection_stage( store.on_credential_load(Arc::new(|store, id, count| { Box::pin(async move { assert!(count <= 4, "snapshot conflict retries must be bounded"); - let account = store.account(id.as_str()).expect("account"); - let observed_at = account - .quota() - .observed_at() - .unwrap_or_else(SystemTime::now) - + Duration::from_millis(1); - store - .apply_quota_access(QuotaAccessChange { - account_id: id.clone(), - expected_revision: account.revision(), - state: QuotaState::allowed(observed_at), - }) - .await?; + bump_account_weight(store, id, count); Ok(()) }) })); @@ -2547,19 +2539,7 @@ async fn api_websocket_precheck_and_selection_share_the_snapshot_retry_budget() ); // 第 1、4 次在传输预检冲突,第 3、6 次在最终候选校验冲突 if matches!(count, 1 | 3 | 4 | 6) { - let account = store.account(id.as_str()).expect("account"); - let observed_at = account - .quota() - .observed_at() - .unwrap_or_else(SystemTime::now) - + Duration::from_millis(1); - store - .apply_quota_access(QuotaAccessChange { - account_id: id.clone(), - expected_revision: account.revision(), - state: QuotaState::allowed(observed_at), - }) - .await?; + bump_account_weight(store, id, count); } Ok(()) }) @@ -2609,6 +2589,15 @@ async fn api_websocket_precheck_and_selection_share_the_snapshot_retry_budget() ); } +// 调度权重属于选号事实,每次变化都必须让已选快照失效 +fn bump_account_weight(store: &MemoryAccountStore, id: &ProviderAccountId, count: usize) { + store.set_scheduling( + id.as_str(), + None, + AccountWeight::new(u16::try_from(count + 1).expect("small count")).expect("weight"), + ); +} + async fn exhaust_account_quota(store: &Arc, account_id: &str) { let account = store.account(account_id).expect("test account"); store