fix(openai): rotate accounts instead of same-socket retry on WebSocket account-level rejections (#89)

WebSocket 升级或交换被上游账号级拒绝(额度耗尽、限流、封禁等)时,恢复策略
仍会把错误交给同账号传输重试,且退避直接采用上游 retry-after(额度窗口通常
是小时级)。请求在 deadline 内无法恢复,错误原样返回客户端,Core 也失去了
换号机会——而同错误在 HTTP 路径上因回放安全会立即换号重试。

现在回放安全且带账号级失败记录的 WebSocket 拒绝不进入传输恢复:保留账号失败
记录(健康度与额度标记立即生效),把换号决策交还 Core,与 HTTP 路径语义一致。
会话 WebSocket 预算与禁用标记也不被账号级拒绝消耗。
This commit is contained in:
2han9wen71an
2026-09-14 17:15:47 +08:00
committed by GitHub
parent cea4e1d8e9
commit 3f69bc861c
2 changed files with 77 additions and 0 deletions
@@ -506,6 +506,27 @@ pub(super) fn apply_websocket_recovery_policy(
failure: &mut MappedProviderFailure,
context: WebSocketRecoveryContext<'_>,
) {
// 账号级上游拒绝(额度、限流、封禁等)没有可恢复的传输语义:同账号重试只会
// 命中同一拒绝,而上游 retry-after 通常是小时级的额度窗口。保留账号失败记录
// 让健康度立即生效,并把回放安全证据随错误交给 Core 换号恢复,与 HTTP 路径
// 的调度语义保持一致。
if failure.account_failure.is_some() && failure.error.replay_is_safe() {
tracing::warn!(
request_id = context.request_id,
attempt_index = context.attempt_index,
account_id = context.account_id,
websocket_failure_kind = failure.error.kind().as_str(),
websocket_failure_code = failure
.error
.upstream_code()
.map_or("", OpaqueUpstreamValue::as_str),
upstream_status_code = failure.error.upstream_status().unwrap_or_default(),
upstream_status_code_present = failure.error.upstream_status().is_some(),
transport_requirement = context.requirement.as_str(),
"OpenAI upstream WebSocket rejected by an account-level failure; deferring to account rotation"
);
return;
}
let session_budget_exhausted = context.session_affinity_key.is_some_and(|key| {
context
.session_transport_recovery
@@ -465,6 +465,8 @@ fn contract_account_scope() -> Arc<FrozenAccountScope> {
"acct_websocket_busy_replay",
"acct_websocket_metadata_close",
"acct_websocket_turn_state",
"acct_ws_quota_a",
"acct_ws_quota_b",
]
.into_iter()
.map(|id| {
@@ -3124,6 +3126,60 @@ async fn downstream_websocket_new_chain_should_override_session_and_attempt_http
server.await.unwrap();
}
#[tokio::test]
async fn websocket_opening_account_rejection_keeps_replay_safe_without_transport_retry() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_ws_quota_a").await;
create_account(&store, "acct_ws_quota_b").await;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
let (mut opening, _) = listener.accept().await.unwrap();
let request = capture_http_request(&mut opening).await;
assert!(String::from_utf8_lossy(&request).starts_with("GET /codex/responses"));
// 额度耗尽拒绝通常携带小时级的 retry-after;同账号重试注定再次命中。
let body = r#"{"error":{"message":"You have reached your usage limit.","type":"rate_limit_error"}}"#;
opening
.write_all(
format!(
"HTTP/1.1 429 Too Many Requests\r\nretry-after: 129600\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
)
.as_bytes(),
)
.await
.unwrap();
});
let provider = provider_with_base_url(&store, base_url);
let mut stream = provider
.execute(
planned_request("openai", generate_operation()),
context("req_ws_quota_rotation", CancellationToken::new()),
)
.await
.expect("prepare quota-rejected stream");
let error = loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("quota rejection must surface as a terminal error"),
}
};
server.await.expect("upstream server");
assert_eq!(error.kind(), ProviderErrorKind::RateLimited);
assert_eq!(error.upstream_status(), Some(429));
assert_eq!(error.send_state(), UpstreamSendState::NotSent);
assert!(
error.replay_is_safe(),
"before-payload rejection is replay safe"
);
// 回放安全的账号级拒绝必须把换号决策留给 Core,不得钉死同账号传输重试
// (旧行为会携带小时级 retry-after 的同账号重试标记,请求必然超时)。
assert_eq!(error.pre_delivery_retry(), None);
}
#[tokio::test]
async fn websocket_turn_state_metadata_is_exposed_through_response_observation() {
let store = Arc::new(MemoryAccountStore::default());