diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 0af3e6762..afd8ac3fe 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -334,6 +334,8 @@ where attempts: 0, websocket_observation_sequence: 0, routing_attempts: 0, + last_attempt_account: None, + account_rotations: 0, candidate_index, excluded_accounts: BTreeSet::new(), credential_recovery_attempted_accounts: BTreeSet::new(), @@ -455,6 +457,15 @@ pub struct ResponseExecutionSession { websocket_observation_sequence: u64, /// 路由预算只统计正常选号/账号恢复,不被 Provider-owned 传输预算消耗 routing_attempts: u32, + /// 上一 attempt 实际选中的账号;与选号成功的 attempt 一一同步更新。 + /// 旧 CurrentAttempt 在丢弃时被 drop、excluded_accounts 无序、持久化记录异步 + /// best-effort,都不能还原「上一账号」,因此由本字段单独记账。 + last_attempt_account: Option, + /// 已发生的换号次数:选中账号与上一 attempt 不同的路由 attempt 计一次。 + /// 首个 attempt 不计;同账号钉选重试(瞬态退避、传输恢复、凭据恢复重放、 + /// continuation 精确重连)不消耗。预算耗尽后所有必然换号的重试门关闭, + /// 换号深度由 [`crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS`] 封顶。 + account_rotations: u32, candidate_index: usize, excluded_accounts: BTreeSet, credential_recovery_attempted_accounts: BTreeSet, @@ -1109,6 +1120,9 @@ where self.continuation_attempt, ContinuationAttempt::None | ContinuationAttempt::ReplayAny ) + // 跨 Provider 候选推进必然换号(账号行按 Provider 隔离), + // 预算耗尽后不再推进,交回容量类失败的原有终态语义。 + && self.account_rotations < crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS && self.advance_provider_candidate(); let retryable = self .apply_retry_policy(super::policy::RetryFacts { @@ -1275,6 +1289,18 @@ where .await?; return Err(EngineError::ContinuationPinMismatch); } + // 换号预算按实际选中账号记账:与上一 attempt 选中账号不同即消耗一次。决策门 + // (重试分类、候选推进、continuation 排除臂)已拦截预算耗尽后的必然换号, + // 这里是统一事实源,覆盖 Provider 自主换号(如 replay owner 重放)等路径; + // 首 attempt 与同账号钉选重试不计数。 + if self.last_attempt_account.as_ref() != Some(metadata.provider_account_id()) + && self + .last_attempt_account + .replace(metadata.provider_account_id().clone()) + .is_some() + { + self.account_rotations = self.account_rotations.saturating_add(1); + } if self.account_state_owner.is_none() { self.account_state_owner = Some(ProviderAccountStateOwner::new( metadata.provider().clone(), @@ -1561,12 +1587,22 @@ where && !self .credential_recovery_attempted_accounts .contains(current.metadata.provider_account_id()); + // 排除当前账号的重选必然换号:ordinary 重选与显式 AccountRotation 标记都落入 + // 同一排除分支。换号预算只在这里消耗;同账号分支(瞬态退避、传输恢复、 + // 凭据恢复重放、continuation 精确重连)不消耗。不能直接把预算门写进 + // ordinary_retry:它是同账号瞬态退避的前置条件,会被连带误伤 + let rotation_retry = !continuation_retry + && !same_account_retry + && transient_retry.is_none() + && transport_recovery.is_none() + && (ordinary_retry || account_rotation_retry) + && self.account_rotations < crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS; let retryable = !error.retry_is_prohibited() && (continuation_retry || same_account_retry - || ordinary_retry - || account_rotation_retry - || transport_recovery.is_some()); + || transient_retry.is_some() + || transport_recovery.is_some() + || rotation_retry); let retryable = match self .apply_retry_policy(super::policy::RetryFacts { @@ -1602,6 +1638,7 @@ where "sameAccountRetry": same_account_retry, "accountRotationRetry": account_rotation_retry, "ordinaryRetry": ordinary_retry, "transportRecovery": transport_recovery.is_some(), "transientRetry": transient_retry.is_some(), + "rotationRetry": rotation_retry, "accountRotations": self.account_rotations, "connectionRetry": connection_retry_requested, "connectionRetries": self.connection_retries, "connectionBudgetRemainingMs": self.connection_budget.remaining().map(duration_ms), @@ -1799,6 +1836,13 @@ where { return false; } + // 两个排除臂都必然换号;预算耗尽后不再排除当前账号做跨账号续写重放, + // 落回不可重试路径以原始上游错误终态。 + ContinuationAttempt::ReplayOwner | ContinuationAttempt::ReplayAny + if self.account_rotations >= crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS => + { + return false; + } ContinuationAttempt::ReplayOwner => { self.continuation_attempt = ContinuationAttempt::ReplayAny; self.excluded_accounts @@ -1846,6 +1890,11 @@ where { return false; } + // Native 期间所有 attempt 都在 pin 账号上,到这里预算必未耗尽; + // 与其余换号路径保持同一预算门,防止状态机演化后破坏换号上限。 + if self.account_rotations >= crate::routing::MAX_ACCOUNT_ROTATION_ATTEMPTS { + return false; + } self.continuation_attempt = ContinuationAttempt::ReplayAny; self.excluded_accounts.insert(pin.account().clone()); diff --git a/backend/crates/gateway-core/src/provider_ports.rs b/backend/crates/gateway-core/src/provider_ports.rs index 65443588d..508655612 100644 --- a/backend/crates/gateway-core/src/provider_ports.rs +++ b/backend/crates/gateway-core/src/provider_ports.rs @@ -886,6 +886,49 @@ impl ProviderRefreshPolicy { Err(_) => true, } } + + /// 按账号派生 `[0, margin]` 内的整秒错峰偏移。 + /// + /// 网关以固定周期扫描到期账号,到期时刻相同的账号会在同一轮齐刷, + /// 在 auth.openai.com 侧形成单 IP 批量刷新特征。偏移从账号 ID 的 + /// 稳定哈希派生(与退避扰动同源、独立 salt),跨扫描与重启保持不变; + /// 只会增大有效提前量,不会比配置的 margin 更贴近过期时刻。 + #[must_use] + pub fn refresh_stagger(self, account_id: &ProviderAccountId) -> Duration { + // margin 以整秒配置;秒级粒度在默认 300s 下对应约 10 个扫描槽位。 + let bound = u32::try_from(self.margin.as_secs()).unwrap_or(u32::MAX); + Duration::from_secs(u64::from(stable_factor( + account_id.as_str(), + "refresh-stagger", + 0, + bound, + ))) + } + + /// 含账号错峰偏移的到期判定;扫描路径先按 [`Self::staggered_refresh_bound`] + /// 取回候选超集,再用它在内存中收窄到本轮真正到期的账号。 + #[must_use] + pub fn is_refresh_due_staggered( + self, + account_id: &ProviderAccountId, + access_token_expires_at: SystemTime, + observed_at: SystemTime, + ) -> bool { + let staggered_margin = self.margin.saturating_add(self.refresh_stagger(account_id)); + match access_token_expires_at.duration_since(observed_at) { + Ok(remaining) => remaining <= staggered_margin, + Err(_) => true, + } + } + + /// 错峰候选窗口上界;覆盖 margin 与最大偏移之和。 + /// + /// 与 [`Self::refresh_stagger`] 的值域同址维护:偏移上限为 margin, + /// 因此 `2 × margin` 必然覆盖最大有效提前量。 + #[must_use] + pub fn staggered_refresh_bound(self) -> Duration { + self.margin.saturating_mul(2) + } } /// 指数退避基准延迟;attempt=1 即为该值 diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index 9c8fef052..c063f4a64 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -26,6 +26,10 @@ use crate::validation::{IdentifierError, RoutingError, validate_text}; const MAX_REQUEST_ATTEMPTS: u32 = 32; +/// 单请求内允许的换号次数上限,选中账号与上一 attempt 不同的路由 attempt 记一次 +/// 初始账号和同账号重试不计入换号次数,所有尝试仍受总路由预算限制 +pub(crate) const MAX_ACCOUNT_ROTATION_ATTEMPTS: u32 = 3; + /// 客户端请求中的模型名称 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize)] #[serde(transparent)] diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index dfb9a79e0..a119ca2f9 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -292,6 +292,7 @@ enum Script { } struct ScriptedProvider { + provider_name: &'static str, profile_generation: AtomicUsize, default_profile_calls: AtomicUsize, default_profile: Mutex>, @@ -311,7 +312,13 @@ impl Drop for TrackedLease { impl ScriptedProvider { fn new(scripts: Vec