fix(xai,web): distinguish account cooldown from upstream failure

Selection-stage failures all collapsed into a generic "upstream service is
unavailable", so a 30s runtime cooldown was indistinguishable from a real
upstream outage. The connection test pins a single account and has no
fallback, making every retry inside the window look like a dead upstream.

Split the runtime-cooldown case out of NoEligibleSession into a new
AccountCoolingDown variant carrying the remaining wait, and attach a
structured code and error type to each selection failure. The admin
connection test now renders the reason and the remaining cooldown.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
zyycn
2026-07-26 22:03:42 +08:00
co-authored by Claude Opus 5
parent 245f87ef35
commit 69f73e8c5d
6 changed files with 136 additions and 28 deletions
@@ -102,7 +102,7 @@ impl GrokAccountSessionSelector {
let runtime_cooldowns = self.runtime_cooldowns(&catalog_eligible).await;
let mut candidates = catalog_eligible
.into_iter()
.filter(|account| !runtime_cooldowns.contains(account.id()))
.filter(|account| !runtime_cooldowns.contains_key(account.id()))
.map(|account| {
let health = self
.account_feedback
@@ -197,18 +197,30 @@ impl GrokAccountSessionSelector {
}
if capacity_denied {
Err(GrokSessionSelectorError::CapacityUnavailable { retry_after })
} else {
Err(GrokSessionSelectorError::NoEligibleSession)
return Err(GrokSessionSelectorError::CapacityUnavailable { retry_after });
}
// 钉死账号时只看该账号,否则取最早退出 cooldown 的候选。
let cooled_until = match request.required_account() {
Some(required) => runtime_cooldowns.get(required).copied(),
None => runtime_cooldowns.values().copied().min(),
};
if let Some(until) = cooled_until {
return Err(GrokSessionSelectorError::AccountCoolingDown {
retry_after: until
.duration_since(SystemTime::now())
.ok()
.filter(|remaining| !remaining.is_zero()),
});
}
Err(GrokSessionSelectorError::NoEligibleSession)
}
async fn runtime_cooldowns(
&self,
accounts: &[ProviderAccount],
) -> std::collections::BTreeSet<ProviderAccountId> {
) -> std::collections::BTreeMap<ProviderAccountId, SystemTime> {
let now = SystemTime::now();
let mut cooled = std::collections::BTreeSet::new();
let mut cooled = std::collections::BTreeMap::new();
for account in accounts {
let Ok(Some(cooldown)) = self.cooldowns.read(account.id()).await else {
continue;
@@ -220,7 +232,7 @@ impl GrokAccountSessionSelector {
continue;
}
if cooldown.until() > now {
cooled.insert(account.id().clone());
cooled.insert(account.id().clone(), cooldown.until());
}
}
cooled
+46 -12
View File
@@ -1592,21 +1592,55 @@ fn map_request_error(error: GrokRequestEncodeError) -> ProviderError {
}
}
/// 将选择阶段失败映射为带结构化 code 与 retry_after 的 Provider 错误。
fn map_selection_error(error: GrokSessionSelectorError) -> ProviderError {
match error {
GrokSessionSelectorError::CapacityUnavailable { retry_after } => {
let error = provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent);
match retry_after {
Some(retry_after) => error.with_retry_after(retry_after),
None => error,
}
}
GrokSessionSelectorError::NoEligibleSession | GrokSessionSelectorError::Unavailable => {
provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent)
}
let (retry_after, message, code) = match error {
GrokSessionSelectorError::AccountCoolingDown { retry_after } => (
retry_after,
cooling_down_message(retry_after),
"account_cooling_down",
),
GrokSessionSelectorError::CapacityUnavailable { retry_after } => (
retry_after,
"account is at its concurrency or request-interval limit".to_owned(),
"account_capacity_busy",
),
GrokSessionSelectorError::NoEligibleSession => (
None,
"no account is eligible for the requested model".to_owned(),
"no_eligible_account",
),
GrokSessionSelectorError::Unavailable => (
None,
"account scheduling state is temporarily unreadable".to_owned(),
"account_selector_unavailable",
),
GrokSessionSelectorError::InvalidSession => {
provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent)
return provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent);
}
};
let error = provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent);
let error = match retry_after {
Some(retry_after) => error.with_retry_after(retry_after),
None => error,
};
match ClientVisibleUpstreamError::new(
message,
Some(code.to_owned()),
Some("account_unavailable_error".to_owned()),
) {
Ok(detail) => error.with_client_visible_upstream_error(detail),
Err(_) => error,
}
}
fn cooling_down_message(retry_after: Option<Duration>) -> String {
match retry_after {
Some(retry_after) => format!(
"account is cooling down after an interrupted upstream stream; retry in {}s",
retry_after.as_secs().saturating_add(1)
),
None => "account is cooling down after an interrupted upstream stream".to_owned(),
}
}
@@ -311,6 +311,12 @@ pub enum GrokSessionSelectorError {
/// Optional selector-derived delay.
retry_after: Option<Duration>,
},
/// Every eligible session is inside its runtime cooldown window.
#[error("Grok Build account is cooling down")]
AccountCoolingDown {
/// Remaining cooldown of the earliest recovering session.
retry_after: Option<Duration>,
},
/// Session metadata or Provider-owned plaintext secret is invalid.
#[error("Grok Build session data is invalid")]
InvalidSession,
@@ -411,13 +411,16 @@ async fn interrupted_stream_feedback_records_runtime_cooldown_without_persisting
.cooldown(session.account_id())
.is_some_and(|cooldown| cooldown.until() > SystemTime::now())
);
assert!(matches!(
fixture
.selector
.select(fixture.request(BTreeSet::new()))
.await,
Err(GrokSessionSelectorError::NoEligibleSession)
));
let retry_after = match fixture
.selector
.select(fixture.request(BTreeSet::new()))
.await
{
Err(GrokSessionSelectorError::AccountCoolingDown { retry_after }) => retry_after,
other => panic!("expected AccountCoolingDown, got {other:?}"),
};
let retry_after = retry_after.expect("retry_after");
assert!(retry_after <= Duration::from_secs(30));
}
#[tokio::test]
@@ -1330,6 +1330,60 @@ async fn selector_capacity_failure_occurs_before_visible_upstream_send() {
assert_eq!(transport.calls.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn selection_failures_carry_a_distinguishable_client_error_code() {
let cases = [
(
GrokSessionSelectorError::AccountCoolingDown {
retry_after: Some(Duration::from_secs(12)),
},
"account_cooling_down",
),
(
GrokSessionSelectorError::CapacityUnavailable {
retry_after: Some(Duration::from_millis(20)),
},
"account_capacity_busy",
),
(
GrokSessionSelectorError::NoEligibleSession,
"no_eligible_account",
),
(
GrokSessionSelectorError::Unavailable,
"account_selector_unavailable",
),
];
for (failure, expected_code) in cases {
let transport = StubInferenceTransport::success();
let provider = provider(StubSelector::failing(failure), transport.clone()).await;
let error = match provider
.execute(
provider_request("xai"),
context(CancellationToken::new(), None),
)
.await
{
Ok(_) => panic!("selection must fail"),
Err(error) => error,
};
let detail = error
.client_visible_upstream_error()
.expect("client visible detail");
assert_eq!(error.kind(), ProviderErrorKind::Unavailable);
assert_eq!(detail.code(), Some(expected_code));
assert_eq!(detail.error_type(), Some("account_unavailable_error"));
assert_eq!(transport.calls.load(Ordering::SeqCst), 0);
if expected_code == "account_cooling_down" {
assert!(
detail.message().contains("retry in 13s"),
"{}",
detail.message()
);
}
}
}
#[tokio::test]
async fn native_previous_response_is_rejected_before_selection() {
let selector = StubSelector::success();
@@ -39,8 +39,7 @@ interface AccountConnectionTestOptions {
onAccountStatus: (accountId: string, status: string) => void
}
// 后端把"选账号阶段失败"和"上游真的挂了"拆成了不同 providerErrorCode,
// 前者是可自愈的等待,不该让人以为上游不可用。
// providerErrorCode 到测试弹窗提示文案的映射。
const CONNECTION_TEST_ERROR_HINTS: Record<string, string> = {
account_cooling_down: '账号冷却中',
account_capacity_busy: '账号并发已占满或请求间隔未到,请稍后重试',