starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: unify account failures and usage diagnostics

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-15 10:15:10 +08:00
co-authored by Codex
parent 096e48621a
commit a4f1843876
34 changed files with 766 additions and 512 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/bootstrap/tasks/coordinator.rs
+1 -1
View File
@@ -82,6 +82,7 @@ impl TaskCoordinator {
stores.accounts.clone(),
stores.account_usage.clone(),
services.codex.clone(),
services.account_pool.clone(),
services.account_pseudonymizer.clone(),
config
.quota
@@ -91,7 +92,6 @@ impl TaskCoordinator {
30 * 60,
)
.with_cookie_store(stores.cookies.clone())
.with_account_pool(services.account_pool.clone())
.start(),
);
coordinator.push("fingerprint_update", {
backend/src/bootstrap/tasks/quota_refresh.rs
+2 -6
View File
@@ -35,6 +35,7 @@ impl QuotaRefreshTask {
store: PgAccountStore,
usage_store: PgAccountUsageStore,
codex: Arc<CodexBackendClient>,
account_pool: Arc<AccountPoolService>,
account_pseudonymizer: Arc<AccountPseudonymizer>,
interval_secs: u64,
min_refresh_interval_secs: u64,
@@ -44,6 +45,7 @@ impl QuotaRefreshTask {
store,
usage_store,
codex,
account_pool,
account_pseudonymizer,
min_refresh_interval_secs,
),
@@ -58,12 +60,6 @@ impl QuotaRefreshTask {
self
}
/// 设置运行时账号池，用于刷新后同步内存调度状态。
pub fn with_account_pool(mut self, account_pool: Arc<AccountPoolService>) -> Self {
self.service = self.service.with_account_pool(account_pool);
self
}
/// 启动后台刷新任务。
pub fn start(self) -> SchedulerHandle {
let config = PeriodicTaskConfig::new(
backend/src/dispatch/controllers/account_failure.rs
+45 -171
View File
@@ -2,19 +2,25 @@
use crate::{
dispatch::{
controllers::{
ControllerFailureFact,
account_state::{AccountStateEffect, AccountStateEffects},
},
controllers::ControllerFailureFact,
errors::ClientFailure,
failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccountKind},
lifecycle::contract::{
AttemptDecision, AttemptObservation, AttemptObservationKind, CompleteResponseFacts,
},
transport::observation::{UpstreamFailureFacts, UpstreamFailureKind},
},
fleet::{account::AccountStatus, pool::AccountPoolService},
upstream::openai::{protocol::responses::ResponsesSseFailure, transport::CodexBackendClient},
fleet::{
account_failure::{
AccountFailureKind, AccountStateEffect, apply_account_state_effect_immediately,
classify_response_failure, classify_upstream_failure,
},
pool::AccountPoolService,
},
upstream::openai::{
failure::{UpstreamFailureFacts, UpstreamFailureKind},
protocol::responses::ResponsesSseFailure,
transport::CodexBackendClient,
},
};
pub(super) struct AccountFailureController;
@@ -46,7 +52,7 @@ impl AccountFailureController {
return;
};
if let Some(effect) = &classified.effect {
AccountStateEffects::apply(account_pool, codex, account_id, effect).await;
apply_account_state_effect_immediately(account_pool, codex, account_id, effect).await;
}
}
@@ -110,15 +116,7 @@ impl AccountFailureController {
.as_deref()
.unwrap_or_default()
.to_ascii_lowercase();
let error_type = failure
.upstream_type
.as_deref()
.unwrap_or_default()
.to_ascii_lowercase();
let message = failure.message.to_ascii_lowercase();
if is_model_unsupported(&code)
|| is_model_unsupported(&message)
|| code.contains("invalid_request")
if code.contains("invalid_request")
|| code.contains("not_found")
|| code.contains("context_window")
|| code.contains("invalid_prompt")
@@ -126,17 +124,18 @@ impl AccountFailureController {
{
return Some(ClientFailure::new(failure.clone(), 400, false));
}
if is_auth_failure(&code, &error_type, &message) {
return Some(ClientFailure::new(failure.clone(), 401, false));
}
if code.contains("forbidden") || code.contains("banned") || error_type == "permission_error"
{
return Some(ClientFailure::new(failure.clone(), 403, false));
}
if code.contains("server_overloaded") {
return Some(ClientFailure::new(failure.clone(), 503, true));
}
None
classify_response_failure(failure).map(|classified| {
let status = match classified.kind {
AccountFailureKind::ModelUnsupported => 400,
AccountFailureKind::Expired => 401,
AccountFailureKind::Disabled | AccountFailureKind::Banned => 403,
AccountFailureKind::QuotaExhausted | AccountFailureKind::RateLimited => 429,
};
ClientFailure::new(failure.clone(), status, false)
})
}
}
@@ -168,167 +167,42 @@ fn classify_failure(
}
fn classify_upstream(account_id: &str, facts: &UpstreamFailureFacts) -> Option<ClassifiedFailure> {
let status = facts.status_code;
let body = facts.body.as_str();
let code = facts.code.as_deref().unwrap_or_default();
let error_type = facts.error_type.as_deref().unwrap_or_default();
let message = facts.message.as_str();
if is_model_unsupported(code) || is_model_unsupported(message) || is_model_unsupported(body) {
return Some(model_unsupported_failure(account_id, body));
}
if let Some(account_status) = explicit_account_status(code, error_type, body) {
return Some(account_status_failure(
account_id,
account_status,
body,
status,
));
}
if status == Some(403) && !is_html(body) {
return Some(account_status_failure(
account_id,
AccountStatus::Banned,
body,
status,
));
}
if status == Some(401) {
return Some(account_status_failure(
account_id,
AccountStatus::Expired,
body,
status,
));
}
None
classified_failure(
account_id,
&facts.body,
facts.status_code,
classify_upstream_failure(facts)?,
)
}
fn classify_sse(account_id: &str, failure: &ResponsesSseFailure) -> Option<ClassifiedFailure> {
let code = failure.upstream_code.as_deref().unwrap_or_default();
let error_type = failure.upstream_type.as_deref().unwrap_or_default();
if is_model_unsupported(code) || is_model_unsupported(&failure.message) {
return Some(model_unsupported_failure(
account_id,
&crate::dispatch::failure::sse::sse_failure_error_body(failure),
));
}
let account_status =
explicit_account_status(code, error_type, &failure.message).or_else(|| {
(code.eq_ignore_ascii_case("forbidden") || error_type == "permission_error")
.then_some(AccountStatus::Banned)
})?;
Some(account_status_failure(
classified_failure(
account_id,
account_status,
&crate::dispatch::failure::sse::sse_failure_error_body(failure),
failure.explicit_status_code,
))
classify_response_failure(failure)?,
)
}
fn model_unsupported_failure(account_id: &str, message: &str) -> ClassifiedFailure {
ClassifiedFailure {
exhaustion: AccountExhaustionRecord::new(
account_id,
ExhaustedAccountKind::ModelUnsupported,
message,
),
effect: None,
}
}
fn account_status_failure(
fn classified_failure(
account_id: &str,
status: AccountStatus,
message: &str,
status_code: Option<u16>,
) -> ClassifiedFailure {
let kind = match status {
AccountStatus::Expired => ExhaustedAccountKind::Expired,
AccountStatus::Disabled => ExhaustedAccountKind::Disabled,
AccountStatus::Banned => ExhaustedAccountKind::Banned,
AccountStatus::Active | AccountStatus::QuotaExhausted => {
unreachable!("account failure controller only invalidates unavailable statuses")
}
classified: crate::fleet::account_failure::ClassifiedAccountFailure,
) -> Option<ClassifiedFailure> {
let kind = match classified.kind {
AccountFailureKind::ModelUnsupported => ExhaustedAccountKind::ModelUnsupported,
AccountFailureKind::Expired => ExhaustedAccountKind::Expired,
AccountFailureKind::Disabled => ExhaustedAccountKind::Disabled,
AccountFailureKind::Banned => ExhaustedAccountKind::Banned,
AccountFailureKind::QuotaExhausted | AccountFailureKind::RateLimited => return None,
};
let mut exhaustion = AccountExhaustionRecord::new(account_id, kind, message);
if let Some(status_code) = status_code {
exhaustion = exhaustion.with_status_code(status_code);
}
ClassifiedFailure {
Some(ClassifiedFailure {
exhaustion,
effect: Some(AccountStateEffect::SetStatus(status)),
}
}
fn explicit_account_status(code: &str, error_type: &str, message: &str) -> Option<AccountStatus> {
let code = code.trim().to_ascii_lowercase();
let error_type = error_type.trim().to_ascii_lowercase();
let message = message.to_ascii_lowercase();
if matches!(
code.as_str(),
"identity_verification_required" | "verification_required"
) || message.contains("identity verification is required")
{
return Some(AccountStatus::Disabled);
}
if matches!(
code.as_str(),
"account_banned"
| "account_deactivated"
| "account_disabled"
| "account_suspended"
| "deactivated_workspace"
| "organization_disabled"
| "workspace_deactivated"
) || message.contains("account is banned")
|| message.contains("account has been banned")
|| message.contains("account deactivated")
|| message.contains("account has been deactivated")
|| message.contains("account disabled")
|| message.contains("account has been disabled")
|| message.contains("account suspended")
|| message.contains("organization has been disabled")
|| message.contains("workspace has been deactivated")
|| message.contains("deactivated_workspace")
{
return Some(AccountStatus::Banned);
}
is_auth_failure(&code, &error_type, &message).then_some(AccountStatus::Expired)
}
fn is_auth_failure(code: &str, error_type: &str, message: &str) -> bool {
matches!(
code,
"token_invalid"
| "token_invalidated"
| "token_expired"
| "token_revoked"
| "refresh_token_invalidated"
| "unauthorized"
| "invalid_api_key"
| "authentication_error"
) || error_type == "authentication_error"
|| message.contains("token revoked")
|| message.contains("token invalidated")
|| message.contains("token invalid")
|| message.contains("token expired")
|| message.contains("unauthorized")
|| message.contains("invalid api key")
}
fn is_model_unsupported(value: &str) -> bool {
let value = value.to_ascii_lowercase();
value.contains("model_not_supported")
|| value.contains("model_not_available")
|| (value.contains("model")
&& (value.contains("not supported")
|| value.contains("not available")
|| value.contains("not_supported")
|| value.contains("not_available")))
}
fn is_html(value: &str) -> bool {
let value = value.trim_start().to_ascii_lowercase();
value.starts_with("<!doctype") || value.starts_with("<html") || value.contains("<html")
effect: classified.effect,
})
}
backend/src/dispatch/controllers/account_state.rs
-38
View File
@@ -1,38 +0,0 @@
//! 明确账号事实触发的运行时失效与连接驱逐。
use chrono::{DateTime, Utc};
use crate::{
fleet::{account::AccountStatus, pool::AccountPoolService},
upstream::openai::transport::CodexBackendClient,
};
pub(super) enum AccountStateEffect {
SetStatus(AccountStatus),
MarkQuotaLimitedUntil(DateTime<Utc>),
}
pub(super) struct AccountStateEffects;
impl AccountStateEffects {
pub(super) async fn apply(
account_pool: &AccountPoolService,
codex: &CodexBackendClient,
account_id: &str,
effect: &AccountStateEffect,
) {
match effect {
AccountStateEffect::SetStatus(status) => {
account_pool
.set_status_immediately(account_id, *status)
.await;
}
AccountStateEffect::MarkQuotaLimitedUntil(until) => {
account_pool
.mark_quota_limited_until_immediately(account_id, *until)
.await;
}
}
codex.evict_websocket_account(account_id).await;
}
}
backend/src/dispatch/controllers/cloudflare.rs
+1 -1
View File
@@ -9,13 +9,13 @@ use crate::{
AttemptDecision, AttemptObservation, AttemptObservationKind, AttemptReturnKind,
},
lifecycle::stream::StreamTerminal,
transport::observation::UpstreamFailureFacts,
},
fleet::{
account::AccountStatus,
cookies::{CloudflareChallengeCooldownTracker, CloudflarePathBlockTracker, PgCookieStore},
pool::AccountPoolService,
},
upstream::openai::failure::UpstreamFailureFacts,
};
pub(super) struct CloudflareController;
backend/src/dispatch/controllers/cyber_policy/mod.rs
+4 -2
View File
@@ -15,9 +15,11 @@ use crate::{
controllers::ControllerFailureFact,
errors::ClientFailure,
lifecycle::contract::{AttemptDecision, AttemptObservation, AttemptReturnKind},
transport::observation::UpstreamFailureFacts,
},
upstream::openai::protocol::responses::{CodexResponsesRequest, ResponsesSseFailure},
upstream::openai::{
failure::UpstreamFailureFacts,
protocol::responses::{CodexResponsesRequest, ResponsesSseFailure},
},
};
use self::{store::Store, types::SessionState};
backend/src/dispatch/controllers/history.rs
+1 -1
View File
@@ -13,9 +13,9 @@ use crate::{
AttemptDecision, AttemptObservation, AttemptObservationKind, AttemptReturnKind,
AttemptRoutingFacts, CompleteResponseFacts, PinnedCandidateAcquireFailureKind,
},
transport::observation::{UpstreamFailureFacts, UpstreamFailureKind},
},
upstream::openai::{
failure::{UpstreamFailureFacts, UpstreamFailureKind},
protocol::responses::{CodexResponsesRequest, PreviousResponseScope, StreamCommitPolicy},
transport::CodexBackendTransport,
transport::websocket::PreviousResponseUnavailableReason,
backend/src/dispatch/controllers/mod.rs
+4 -3
View File
@@ -49,14 +49,15 @@ use crate::{
},
stream::StreamTerminal,
},
transport::observation::UpstreamFailureFacts,
},
fleet::pool::AccountPoolService,
upstream::openai::protocol::responses::{CodexResponsesRequest, ResponsesSseFailure},
upstream::openai::{
failure::UpstreamFailureFacts,
protocol::responses::{CodexResponsesRequest, ResponsesSseFailure},
},
};
mod account_failure;
mod account_state;
mod affinity;
pub(crate) mod cloudflare;
pub(crate) mod cyber_policy;
backend/src/dispatch/controllers/quota.rs
+25 -123
View File
@@ -1,23 +1,22 @@
//! 请求前配额验证规则 owner。
use chrono::{Duration, Utc};
use serde_json::Value;
use crate::{
dispatch::{
affinity::AccountIdentityService,
controllers::{
ControllerFailureFact,
account_state::{AccountStateEffect, AccountStateEffects},
},
controllers::ControllerFailureFact,
errors::ClientFailure,
failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccountKind},
lifecycle::contract::{
AttemptDecision, AttemptObservation, AttemptObservationKind, AttemptReturnKind,
},
transport::observation::UpstreamFailureFacts,
},
fleet::{
account_failure::{
AccountFailureKind, AccountStateEffect, apply_account_state_effect_immediately,
classify_response_failure, classify_upstream_failure,
},
pool::{AccountLease, AccountPoolService},
quota::{
quota_from_usage, quota_snapshot_limit_reached, quota_snapshot_limit_window_seconds,
@@ -25,13 +24,13 @@ use crate::{
},
},
upstream::openai::{
failure::UpstreamFailureFacts,
protocol::responses::ResponsesSseFailure,
transport::{CodexBackendClient, CodexRequestContext},
},
};
const LIMIT_REACHED_MESSAGE: &str = "Upstream usage quota still reports limit_reached";
const DEFAULT_RATE_LIMIT_RETRY_SECONDS: u64 = 60;
pub(super) enum QuotaEnterOutcome {
Ready(Box<AccountLease>),
@@ -167,7 +166,7 @@ impl QuotaController {
return;
};
if let Some(effect) = &classified.effect {
AccountStateEffects::apply(account_pool, codex, account_id, effect).await;
apply_account_state_effect_immediately(account_pool, codex, account_id, effect).await;
}
}
@@ -201,131 +200,34 @@ fn classify_upstream(
account_id: &str,
facts: &UpstreamFailureFacts,
) -> Option<ClassifiedQuotaFailure> {
let signaled_kind = facts
.code
.as_deref()
.and_then(classify_quota_signal)
.or_else(|| facts.error_type.as_deref().and_then(classify_quota_signal))
.or_else(|| classify_quota_message(&facts.message));
match signaled_kind.or(match facts.status_code {
Some(429) => Some(StreamQuotaKind::RateLimited),
Some(402) => Some(StreamQuotaKind::QuotaExhausted),
_ => None,
})? {
StreamQuotaKind::RateLimited => Some(rate_limited_failure(
account_id,
facts.body.clone(),
facts.retry_after_seconds,
)),
StreamQuotaKind::QuotaExhausted => {
Some(quota_exhausted_failure(account_id, facts.body.clone()))
}
}
classified_quota_failure(account_id, &facts.body, classify_upstream_failure(facts)?)
}
fn classify_stream(
account_id: &str,
failure: &ResponsesSseFailure,
) -> Option<ClassifiedQuotaFailure> {
let kind = classify_stream_signal(failure).or(match failure.explicit_status_code {
Some(429) => Some(StreamQuotaKind::RateLimited),
Some(402) => Some(StreamQuotaKind::QuotaExhausted),
_ => None,
})?;
let body = crate::dispatch::failure::sse::sse_failure_error_body(failure);
Some(match kind {
StreamQuotaKind::RateLimited => {
rate_limited_failure(account_id, body, failure.retry_after_seconds)
}
StreamQuotaKind::QuotaExhausted => quota_exhausted_failure(account_id, body),
})
classified_quota_failure(account_id, &body, classify_response_failure(failure)?)
}
#[derive(Clone, Copy)]
enum StreamQuotaKind {
RateLimited,
QuotaExhausted,
}
fn classify_stream_signal(failure: &ResponsesSseFailure) -> Option<StreamQuotaKind> {
failure
.upstream_code
.as_deref()
.and_then(classify_quota_signal)
.or_else(|| {
failure
.upstream_type
.as_deref()
.and_then(classify_quota_signal)
})
.or_else(|| classify_quota_message(&failure.message))
}
fn classify_quota_signal(signal: &str) -> Option<StreamQuotaKind> {
let signal = signal.trim().to_ascii_lowercase();
match signal.as_str() {
"usage_limit_reached"
| "rate_limit_exceeded"
| "rate_limit_reached"
| "rate_limit_error"
| "workspace_owner_usage_limit_reached"
| "workspace_member_usage_limit_reached" => Some(StreamQuotaKind::RateLimited),
"quota_exhausted"
| "quota_exceeded"
| "payment_required"
| "insufficient_quota"
| "workspace_owner_credits_depleted"
| "workspace_member_credits_depleted" => Some(StreamQuotaKind::QuotaExhausted),
signal if signal.starts_with("billing_limit") => Some(StreamQuotaKind::QuotaExhausted),
_ => None,
}
}
fn classify_quota_message(message: &str) -> Option<StreamQuotaKind> {
let message = message.to_ascii_lowercase();
if message.contains("rate limit") || message.contains("usage limit") {
return Some(StreamQuotaKind::RateLimited);
}
if message.contains("quota")
|| message.contains("payment required")
|| message.contains("billing limit")
{
return Some(StreamQuotaKind::QuotaExhausted);
}
None
}
fn rate_limited_failure(
fn classified_quota_failure(
account_id: &str,
message: String,
retry_after_seconds: Option<u64>,
) -> ClassifiedQuotaFailure {
let seconds = retry_after_seconds
.unwrap_or(DEFAULT_RATE_LIMIT_RETRY_SECONDS)
.min(i64::MAX as u64) as i64;
ClassifiedQuotaFailure {
exhaustion: AccountExhaustionRecord::new(
account_id,
ExhaustedAccountKind::RateLimited,
message,
),
effect: Some(AccountStateEffect::MarkQuotaLimitedUntil(
Utc::now() + Duration::seconds(seconds),
)),
}
}
fn quota_exhausted_failure(account_id: &str, message: String) -> ClassifiedQuotaFailure {
ClassifiedQuotaFailure {
exhaustion: AccountExhaustionRecord::new(
account_id,
ExhaustedAccountKind::QuotaExhausted,
message,
),
effect: Some(AccountStateEffect::SetStatus(
crate::fleet::account::AccountStatus::QuotaExhausted,
)),
}
message: &str,
classified: crate::fleet::account_failure::ClassifiedAccountFailure,
) -> Option<ClassifiedQuotaFailure> {
let kind = match classified.kind {
AccountFailureKind::RateLimited => ExhaustedAccountKind::RateLimited,
AccountFailureKind::QuotaExhausted => ExhaustedAccountKind::QuotaExhausted,
AccountFailureKind::ModelUnsupported
| AccountFailureKind::Expired
| AccountFailureKind::Disabled
| AccountFailureKind::Banned => return None,
};
Some(ClassifiedQuotaFailure {
exhaustion: AccountExhaustionRecord::new(account_id, kind, message),
effect: classified.effect,
})
}
fn acquired_with_verified_quota(mut acquired: AccountLease, quota: &Value) -> AccountLease {
backend/src/dispatch/controllers/telemetry/events.rs
+16
View File
@@ -546,6 +546,9 @@ pub(super) async fn record_live_response_stream_event(
status_code,
body,
);
if !is_success_usage_event(level, status_code) {
ensure_stream_failure_class(&mut metadata);
}
log_live_response_stream_finalized(context, status_code, level, message, &metadata);
enrich_usage_record_identity(
&mut metadata,
@@ -601,6 +604,19 @@ fn is_success_usage_event(level: UsageRecordLevel, status_code: i64) -> bool {
level != UsageRecordLevel::Error && status_code < 400
}
fn ensure_stream_failure_class(metadata: &mut Value) {
let failure_class = ["failureClass", "upstreamCode", "terminal"]
.into_iter()
.find_map(|field| metadata_string_field(metadata, field))
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string);
let (Some(failure_class), Some(object)) = (failure_class, metadata.as_object_mut()) else {
return;
};
object.insert("failureClass".to_string(), Value::String(failure_class));
}
fn log_live_response_stream_finalized(
context: &StreamContext<'_>,
status_code: i64,
backend/src/dispatch/lifecycle/attempt.rs
+1 -1
View File
@@ -21,7 +21,6 @@ use crate::{
create_response_with_account, prepare_response_transport_with_account,
},
canonical::{CanonicalStreamDecoder, normalize_complete_response},
observation::upstream_failure_facts,
prefetch::{PrefetchedStream, StreamPrefetchError, prefetch_until_commit},
},
},
@@ -30,6 +29,7 @@ use crate::{
pool::{AccountCandidateLease, AccountPoolService},
},
upstream::openai::{
failure::upstream_failure_facts,
protocol::responses::{
CodexResponsesRequest, CollectedResponse, TransportRequirement, transport_requirement,
},
backend/src/dispatch/lifecycle/contract.rs
+2 -4
View File
@@ -6,13 +6,11 @@ use crate::{
dispatch::{
errors::{ClientFailure, ResponseDispatchError},
failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccount},
transport::{
canonical::{CanonicalStreamBatch, CanonicalStreamDecoder},
observation::UpstreamFailureFacts,
},
transport::canonical::{CanonicalStreamBatch, CanonicalStreamDecoder},
},
fleet::{account::Account, pool::AccountLease},
upstream::openai::{
failure::UpstreamFailureFacts,
protocol::{
responses::{CodexResponsesRequest, CollectedResponse, ResponsesSseFailure},
sse::SseError,
backend/src/dispatch/transport/mod.rs
+1 -2
View File
@@ -1,6 +1,5 @@
//! Attempt 阶段的账号作用域上游调用与事实规范化。
//! Attempt 阶段的账号作用域上游调用与流规范化。
pub(in crate::dispatch) mod account;
pub(crate) mod canonical;
pub(in crate::dispatch) mod observation;
pub(in crate::dispatch) mod prefetch;
backend/src/fleet/account_failure.rs
+317
View File
@@ -0,0 +1,317 @@
//! 上游失败到账号状态 effect 的唯一领域分类。
use chrono::{Duration, Utc};
use crate::{
fleet::{account::AccountStatus, pool::AccountPoolService},
upstream::openai::{
failure::{UpstreamFailureFacts, upstream_failure_facts},
protocol::responses::ResponsesSseFailure,
transport::{CodexBackendClient, CodexClientError},
},
};
const DEFAULT_RATE_LIMIT_RETRY_SECONDS: u64 = 60;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AccountFailureKind {
ModelUnsupported,
Expired,
Disabled,
Banned,
QuotaExhausted,
RateLimited,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AccountStateEffect {
SetStatus(AccountStatus),
MarkQuotaLimitedUntil(chrono::DateTime<Utc>),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClassifiedAccountFailure {
pub kind: AccountFailureKind,
pub effect: Option<AccountStateEffect>,
}
impl ClassifiedAccountFailure {
pub fn account_status(&self) -> Option<AccountStatus> {
match self.effect.as_ref()? {
AccountStateEffect::SetStatus(status) => Some(*status),
AccountStateEffect::MarkQuotaLimitedUntil(_) => None,
}
}
}
pub fn classify_client_failure(error: &CodexClientError) -> Option<ClassifiedAccountFailure> {
classify_upstream_failure(&upstream_failure_facts(error))
}
pub fn classify_upstream_failure(facts: &UpstreamFailureFacts) -> Option<ClassifiedAccountFailure> {
classify_failure(FailureFields {
status_code: facts.status_code,
code: facts.code.as_deref(),
error_type: facts.error_type.as_deref(),
message: &facts.message,
body: &facts.body,
retry_after_seconds: facts.retry_after_seconds,
})
}
pub fn classify_response_failure(
failure: &ResponsesSseFailure,
) -> Option<ClassifiedAccountFailure> {
classify_failure(FailureFields {
status_code: failure.explicit_status_code,
code: failure.upstream_code.as_deref(),
error_type: failure.upstream_type.as_deref(),
message: &failure.message,
body: &failure.message,
retry_after_seconds: failure.retry_after_seconds,
})
}
pub async fn apply_account_state_effect_immediately(
account_pool: &AccountPoolService,
codex: &CodexBackendClient,
account_id: &str,
effect: &AccountStateEffect,
) {
match effect {
AccountStateEffect::SetStatus(status) => {
account_pool
.set_status_immediately(account_id, *status)
.await;
}
AccountStateEffect::MarkQuotaLimitedUntil(until) => {
account_pool
.mark_quota_limited_until_immediately(account_id, *until)
.await;
}
}
codex.evict_websocket_account(account_id).await;
}
pub async fn apply_account_state_effect(
account_pool: &AccountPoolService,
codex: &CodexBackendClient,
account_id: &str,
effect: &AccountStateEffect,
) -> bool {
let updated = match effect {
AccountStateEffect::SetStatus(status) => account_pool.set_status(account_id, *status).await,
AccountStateEffect::MarkQuotaLimitedUntil(until) => {
account_pool
.mark_quota_limited_until(account_id, *until)
.await
}
};
codex.evict_websocket_account(account_id).await;
updated
}
pub async fn apply_client_failure_effect(
account_pool: &AccountPoolService,
codex: &CodexBackendClient,
account_id: &str,
error: &CodexClientError,
) -> Option<ClassifiedAccountFailure> {
let classified = classify_client_failure(error)?;
if let Some(effect) = &classified.effect {
apply_account_state_effect(account_pool, codex, account_id, effect).await;
}
Some(classified)
}
struct FailureFields<'a> {
status_code: Option<u16>,
code: Option<&'a str>,
error_type: Option<&'a str>,
message: &'a str,
body: &'a str,
retry_after_seconds: Option<u64>,
}
fn classify_failure(fields: FailureFields<'_>) -> Option<ClassifiedAccountFailure> {
let code = fields.code.unwrap_or_default();
let error_type = fields.error_type.unwrap_or_default();
if [code, fields.message, fields.body]
.into_iter()
.any(is_model_unsupported)
{
return Some(classified(AccountFailureKind::ModelUnsupported, None));
}
if let Some(status) = explicit_account_status(code, error_type, fields.message)
.or_else(|| explicit_account_status(code, error_type, fields.body))
{
return Some(status_failure(status));
}
if fields.status_code == Some(403) && !is_html(fields.body) {
return Some(status_failure(AccountStatus::Banned));
}
if fields.status_code == Some(401) {
return Some(status_failure(AccountStatus::Expired));
}
let quota_kind = [code, error_type]
.into_iter()
.find_map(classify_quota_signal)
.or_else(|| classify_quota_message(fields.message))
.or_else(|| classify_quota_message(fields.body))
.or(match fields.status_code {
Some(429) => Some(AccountFailureKind::RateLimited),
Some(402) => Some(AccountFailureKind::QuotaExhausted),
_ => None,
})?;
Some(match quota_kind {
AccountFailureKind::RateLimited => {
let seconds = fields
.retry_after_seconds
.unwrap_or(DEFAULT_RATE_LIMIT_RETRY_SECONDS)
.min(i64::MAX as u64) as i64;
classified(
AccountFailureKind::RateLimited,
Some(AccountStateEffect::MarkQuotaLimitedUntil(
Utc::now() + Duration::seconds(seconds),
)),
)
}
AccountFailureKind::QuotaExhausted => classified(
AccountFailureKind::QuotaExhausted,
Some(AccountStateEffect::SetStatus(AccountStatus::QuotaExhausted)),
),
AccountFailureKind::ModelUnsupported
| AccountFailureKind::Expired
| AccountFailureKind::Disabled
| AccountFailureKind::Banned => unreachable!("quota classifier returned account status"),
})
}
fn status_failure(status: AccountStatus) -> ClassifiedAccountFailure {
let kind = match status {
AccountStatus::Expired => AccountFailureKind::Expired,
AccountStatus::Disabled => AccountFailureKind::Disabled,
AccountStatus::Banned => AccountFailureKind::Banned,
AccountStatus::Active | AccountStatus::QuotaExhausted => {
unreachable!("explicit account failure must make the account unavailable")
}
};
classified(kind, Some(AccountStateEffect::SetStatus(status)))
}
fn classified(
kind: AccountFailureKind,
effect: Option<AccountStateEffect>,
) -> ClassifiedAccountFailure {
ClassifiedAccountFailure { kind, effect }
}
fn explicit_account_status(code: &str, error_type: &str, message: &str) -> Option<AccountStatus> {
let code = code.trim().to_ascii_lowercase();
let error_type = error_type.trim().to_ascii_lowercase();
let message = message.to_ascii_lowercase();
if matches!(
code.as_str(),
"identity_verification_required" | "verification_required"
) || message.contains("identity verification is required")
{
return Some(AccountStatus::Disabled);
}
if matches!(
code.as_str(),
"account_banned"
| "account_deactivated"
| "account_disabled"
| "account_suspended"
| "deactivated_workspace"
| "organization_disabled"
| "workspace_deactivated"
) || message.contains("account is banned")
|| message.contains("account has been banned")
|| message.contains("account deactivated")
|| message.contains("account has been deactivated")
|| message.contains("account disabled")
|| message.contains("account has been disabled")
|| message.contains("account suspended")
|| message.contains("organization has been disabled")
|| message.contains("workspace has been deactivated")
|| message.contains("deactivated_workspace")
|| code == "forbidden"
|| error_type == "permission_error"
{
return Some(AccountStatus::Banned);
}
is_auth_failure(&code, &error_type, &message).then_some(AccountStatus::Expired)
}
fn is_auth_failure(code: &str, error_type: &str, message: &str) -> bool {
matches!(
code,
"token_invalid"
| "token_invalidated"
| "token_expired"
| "token_revoked"
| "refresh_token_invalidated"
| "unauthorized"
| "invalid_api_key"
| "authentication_error"
) || error_type == "authentication_error"
|| message.contains("token revoked")
|| message.contains("token invalidated")
|| message.contains("token invalid")
|| message.contains("token expired")
|| message.contains("unauthorized")
|| message.contains("invalid api key")
}
fn classify_quota_signal(signal: &str) -> Option<AccountFailureKind> {
let signal = signal.trim().to_ascii_lowercase();
match signal.as_str() {
"usage_limit_reached"
| "rate_limit_exceeded"
| "rate_limit_reached"
| "rate_limit_error"
| "workspace_owner_usage_limit_reached"
| "workspace_member_usage_limit_reached" => Some(AccountFailureKind::RateLimited),
"quota_exhausted"
| "quota_exceeded"
| "payment_required"
| "insufficient_quota"
| "workspace_owner_credits_depleted"
| "workspace_member_credits_depleted" => Some(AccountFailureKind::QuotaExhausted),
signal if signal.starts_with("billing_limit") => Some(AccountFailureKind::QuotaExhausted),
_ => None,
}
}
fn classify_quota_message(message: &str) -> Option<AccountFailureKind> {
let message = message.to_ascii_lowercase();
if message.contains("rate limit") || message.contains("usage limit") {
return Some(AccountFailureKind::RateLimited);
}
if message.contains("quota")
|| message.contains("payment required")
|| message.contains("billing limit")
{
return Some(AccountFailureKind::QuotaExhausted);
}
None
}
fn is_model_unsupported(value: &str) -> bool {
let value = value.to_ascii_lowercase();
value.contains("model_not_supported")
|| value.contains("model_not_available")
|| (value.contains("model")
&& (value.contains("not supported")
|| value.contains("not available")
|| value.contains("not_supported")
|| value.contains("not_available")))
}
fn is_html(value: &str) -> bool {
let value = value.trim_start().to_ascii_lowercase();
value.starts_with("<!doctype") || value.starts_with("<html") || value.contains("<html")
}
backend/src/fleet/manage/probe.rs
+8 -49
View File
@@ -7,16 +7,18 @@ use serde_json::{Value, json};
use tokio::sync::mpsc;
use crate::{
fleet::{account::AccountStatus, store::StoredAccount},
fleet::{
account::AccountStatus,
account_failure::{classify_client_failure, classify_response_failure},
store::StoredAccount,
},
models::{service::ModelRefreshPlanAccount, types::CodexModelInfo},
upstream::openai::{
protocol::{
responses::{CodexResponsesRequest, ResponsesSseFailure},
sse::{SseEvent, encode_sse_event, parse_sse_events},
},
transport::{
CodexClientError, CodexRequestContext, is_banned_auth_signal, is_banned_upstream_error,
},
transport::{CodexClientError, CodexRequestContext},
},
};
@@ -366,54 +368,11 @@ async fn process_sse_event(
}
fn account_status_from_client_error(error: &CodexClientError) -> Option<AccountStatus> {
if is_banned_upstream_error(error) {
Some(AccountStatus::Banned)
} else {
match error {
CodexClientError::Upstream { status, .. } if status.as_u16() == 402 => {
Some(AccountStatus::QuotaExhausted)
}
CodexClientError::Upstream { status, body, .. } if status.as_u16() == 401 => {
Some(if is_banned_auth_signal(body) {
AccountStatus::Banned
} else {
AccountStatus::Expired
})
}
_ => None,
}
}
classify_client_failure(error)?.account_status()
}
fn sse_failure_account_status(failure: &ResponsesSseFailure) -> Option<AccountStatus> {
let code = failure
.upstream_code
.as_deref()
.unwrap_or_default()
.to_ascii_lowercase();
let message = failure.message.to_ascii_lowercase();
if matches!(code.as_str(), "quota_exceeded" | "insufficient_quota") || message.contains("quota")
{
return Some(AccountStatus::QuotaExhausted);
}
let authentication_failed = matches!(
code.as_str(),
"token_invalid"
| "token_expired"
| "token_revoked"
| "account_deactivated"
| "unauthorized"
| "invalid_api_key"
) || message.contains("token revoked")
|| message.contains("token invalid")
|| message.contains("token expired");
authentication_failed.then(|| {
if is_banned_auth_signal(&failure.message) {
AccountStatus::Banned
} else {
AccountStatus::Expired
}
})
classify_response_failure(failure)?.account_status()
}
fn take_sse_frame(buffer: &mut String) -> Option<String> {
backend/src/fleet/manage/quota.rs
+9 -5
View File
@@ -1,6 +1,7 @@
use secrecy::ExposeSecret;
use super::{AccountManageService, types::AccountManageError};
use crate::fleet::account_failure::apply_client_failure_effect;
impl AccountManageService {
async fn usage_cookie_header(&self, account_id: &str) -> Option<String> {
@@ -53,11 +54,14 @@ impl AccountManageService {
client_request_id: None,
turn_id: None,
};
let raw = self
.codex
.fetch_usage(context)
.await
.map_err(|e| AccountManageError::FetchQuota(e.to_string()))?;
let raw = match self.codex.fetch_usage(context).await {
Ok(raw) => raw,
Err(error) => {
apply_client_failure_effect(&self.account_pool, &self.codex, account_id, &error)
.await;
return Err(AccountManageError::FetchQuota(error.to_string()));
}
};
let normalized = crate::fleet::quota::quota_from_usage(&raw);
self.account_pool
.apply_quota_snapshot(account_id, &normalized)
backend/src/fleet/manage/types.rs
+4 -5
View File
@@ -7,6 +7,7 @@ use thiserror::Error;
use crate::{
fleet::{
account::{AccountStatus, AccountStatus as AcctStatus},
account_failure::classify_client_failure,
store::StoredAccountMetadata,
},
upstream::openai::{token_client::RefreshFailure, transport::CodexClientError},
@@ -323,11 +324,9 @@ fn normalized_plan_type(value: &str) -> Option<String> {
}
pub(super) fn import_status_from_usage_error(error: &CodexClientError) -> Option<AccountStatus> {
if crate::upstream::openai::transport::is_banned_upstream_error(error) {
Some(AccountStatus::Banned)
} else {
None
}
classify_client_failure(error)?
.account_status()
.filter(|status| *status == AccountStatus::Banned)
}
pub(super) fn parse_account_status(status: &str) -> Result<AcctStatus, AccountManageError> {
backend/src/fleet/mod.rs
+1
View File
@@ -1,6 +1,7 @@
//! 账号、调度、配额与刷新领域。
pub mod account;
pub mod account_failure;
pub mod cookies;
pub mod import;
pub mod manage;
backend/src/fleet/quota/runtime.rs
+14 -13
View File
@@ -13,6 +13,7 @@ use tracing::warn;
use crate::fleet::{
account::{Account, AccountStatus},
account_failure::apply_client_failure_effect,
pool::AccountPoolService,
quota::{
quota_from_usage, quota_snapshot_limit_reached, quota_snapshot_limit_window_seconds,
@@ -41,7 +42,7 @@ pub struct QuotaRefreshService {
request_spacing: Duration,
account_pseudonymizer: Arc<crate::infra::identity::AccountPseudonymizer>,
cookie_store: Option<PgCookieStore>,
account_pool: Option<Arc<AccountPoolService>>,
account_pool: Arc<AccountPoolService>,
}
impl QuotaRefreshService {
@@ -50,6 +51,7 @@ impl QuotaRefreshService {
store: PgAccountStore,
usage_store: PgAccountUsageStore,
codex: Arc<CodexBackendClient>,
account_pool: Arc<AccountPoolService>,
account_pseudonymizer: Arc<crate::infra::identity::AccountPseudonymizer>,
) -> Self {
Self {
@@ -60,7 +62,7 @@ impl QuotaRefreshService {
request_spacing: Self::default_request_spacing(),
account_pseudonymizer,
cookie_store: None,
account_pool: None,
account_pool,
}
}
@@ -73,6 +75,7 @@ impl QuotaRefreshService {
store: PgAccountStore,
usage_store: PgAccountUsageStore,
codex: Arc<CodexBackendClient>,
account_pool: Arc<AccountPoolService>,
account_pseudonymizer: Arc<crate::infra::identity::AccountPseudonymizer>,
min_refresh_interval_secs: u64,
) -> Self {
@@ -84,7 +87,7 @@ impl QuotaRefreshService {
request_spacing: Self::default_request_spacing(),
account_pseudonymizer,
cookie_store: None,
account_pool: None,
account_pool,
}
}
@@ -100,12 +103,6 @@ impl QuotaRefreshService {
self
}
/// 设置运行时账号池，用于刷新后同步内存调度状态。
pub fn with_account_pool(mut self, account_pool: Arc<AccountPoolService>) -> Self {
self.account_pool = Some(account_pool);
self
}
/// 执行一次配额锁定或待验证账号刷新，复用调用方传入的最近刷新记录。
pub async fn refresh_locked_accounts(
&self,
@@ -211,6 +208,13 @@ impl QuotaRefreshService {
self.sync_runtime_pool_account(&account.id).await;
}
Err(error) => {
apply_client_failure_effect(
&self.account_pool,
&self.codex,
&account.id,
&error,
)
.await;
warn!(
account_id = %account.id,
error = %error,
@@ -238,10 +242,7 @@ impl QuotaRefreshService {
}
async fn sync_runtime_pool_account(&self, account_id: &str) {
let Some(account_pool) = &self.account_pool else {
return;
};
if let Err(error) = account_pool.sync_account_from_store(account_id).await {
if let Err(error) = self.account_pool.sync_account_from_store(account_id).await {
warn!(
account_id = %account_id,
error = %error,
backend/src/telemetry/usage/insights.rs
+12 -5
View File
@@ -49,7 +49,13 @@ error_terminals as (
case when request_id is null then 'event:' || id else 'request:' || request_id end as request_key,
id, request_id, created_at as terminal_at, client_api_key_id, provider,
account_id, transport, coalesce(nullif(model, ''), '未知模型') as model_name,
coalesce(nullif(failure_class, ''), '未分类错误') as failure_class,
coalesce(
nullif(failure_class, ''),
nullif(metadata_json->>'failureClass', ''),
nullif(metadata_json->>'upstreamCode', ''),
nullif(metadata_json->>'terminal', ''),
'未分类错误'
) as failure_class,
latency_ms
from ops_error_logs
where created_at >= $1 and created_at < $2
@@ -166,9 +172,7 @@ impl UsageDiagnosticsDimension {
Self::Account => {
"coalesce(nullif(concat_ws(' → ', nullif(diagnostic_accounts.email, ''), nullif(terminal_requests.account_id, '')), ''), '未知账号')"
}
Self::ApiKey => {
"coalesce(nullif(terminal_requests.client_api_key_id, ''), '未知 API Key')"
}
Self::ApiKey => "coalesce(nullif(diagnostic_api_keys.name, ''), '未知密钥')",
Self::Provider => "coalesce(nullif(terminal_requests.provider, ''), '未知 Provider')",
Self::Transport => "coalesce(nullif(terminal_requests.transport, ''), '未知传输')",
Self::FailureClass => "terminal_requests.failure_class",
@@ -180,7 +184,10 @@ impl UsageDiagnosticsDimension {
Self::Account => {
"terminal_requests left join accounts diagnostic_accounts on diagnostic_accounts.id = terminal_requests.account_id"
}
Self::Model | Self::ApiKey | Self::Provider | Self::Transport | Self::FailureClass => {
Self::ApiKey => {
"terminal_requests left join client_api_keys diagnostic_api_keys on diagnostic_api_keys.id = terminal_requests.client_api_key_id"
}
Self::Model | Self::Provider | Self::Transport | Self::FailureClass => {
"terminal_requests"
}
}
backend/src/dispatch/transport/observation.rs → backend/src/upstream/openai/failure.rs
+5 -6
View File
@@ -1,4 +1,4 @@
//! 上游原始错误到 attempt 通用事实的单次规范化。
//! Codex 上游原始错误到稳定失败事实的规范化。
use serde_json::Value;
@@ -8,7 +8,7 @@ use crate::upstream::openai::transport::{
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(in crate::dispatch) enum UpstreamFailureKind {
pub enum UpstreamFailureKind {
HttpStatus,
HttpConnect,
HttpTimeout,
@@ -25,7 +25,7 @@ pub(in crate::dispatch) enum UpstreamFailureKind {
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(in crate::dispatch) struct UpstreamFailureFacts {
pub struct UpstreamFailureFacts {
pub kind: UpstreamFailureKind,
pub status_code: Option<u16>,
pub code: Option<String>,
@@ -35,9 +35,7 @@ pub(in crate::dispatch) struct UpstreamFailureFacts {
pub retry_after_seconds: Option<u64>,
}
pub(in crate::dispatch) fn upstream_failure_facts(
error: &CodexClientError,
) -> UpstreamFailureFacts {
pub fn upstream_failure_facts(error: &CodexClientError) -> UpstreamFailureFacts {
match error {
CodexClientError::Http(error) => UpstreamFailureFacts {
kind: if error.is_connect() {
@@ -156,6 +154,7 @@ fn error_fields(body: &str) -> (Option<String>, Option<String>, Option<String>)
let error = value
.pointer("/response/error")
.or_else(|| value.get("error"))
.or_else(|| value.get("detail"))
.unwrap_or(&value);
(
error
backend/src/upstream/openai/mod.rs
+1
View File
@@ -1,5 +1,6 @@
//! OpenAI/Codex 上游客户端。
pub mod failure;
pub mod fingerprint;
pub mod protocol;
pub mod token_client;
backend/src/upstream/openai/transport/client.rs
-33
View File
@@ -166,39 +166,6 @@ impl CodexClientError {
}
}
/// 判断上游错误正文是否表示账号已封禁或停用。
pub fn is_banned_auth_signal(value: &str) -> bool {
let value = value.to_ascii_lowercase();
value.contains("account_deactivated")
|| value.contains("account deactivated")
|| value.contains("account has been deactivated")
|| value.contains("deactivated")
|| value.contains("banned")
}
/// 判断 402 错误正文是否带有 OpenAI 工作区停用标记。
pub fn is_deactivated_workspace_error_body(value: &str) -> bool {
serde_json::from_str::<Value>(value).is_ok_and(|value| {
value.pointer("/detail/code").and_then(Value::as_str) == Some("deactivated_workspace")
})
}
/// 判断 Codex 上游错误是否表示账号已封禁或停用。
pub fn is_banned_upstream_error(error: &CodexClientError) -> bool {
matches!(
error,
CodexClientError::Upstream { status, body, .. }
if (status.as_u16() == 403
&& !is_html_error_body(body))
|| (status.as_u16() == 402 && is_deactivated_workspace_error_body(body))
)
}
fn is_html_error_body(value: &str) -> bool {
let value = value.trim_start().to_ascii_lowercase();
value.starts_with("<!doctype") || value.starts_with("<html") || value.contains("<html")
}
/// Codex 客户端结果类型。
pub type CodexClientResult<T> = Result<T, CodexClientError>;
backend/src/upstream/openai/transport/mod.rs
-1
View File
@@ -20,7 +20,6 @@ pub use self::{
CodexModelCatalogClient, CodexModelCatalogClientError, CodexModelCatalogRequest,
CodexRateLimitHeaderUpdates, CodexRequestContext, CodexResponseMetadata,
CodexTransportDecision, CodexTransportMetrics, CodexTurnStateUpdate, build_reqwest_client,
is_banned_auth_signal, is_banned_upstream_error, is_deactivated_workspace_error_body,
},
diagnostics::CodexUpstreamDiagnostics,
endpoints::{
backend/tests/api/admin/accounts_routes/quota_routes.rs
+57
View File
@@ -195,3 +195,60 @@ async fn admin_account_quota_should_send_usage_cookie() {
"2K"
);
}
#[tokio::test]
async fn admin_account_quota_should_ban_deactivated_workspace() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/backend-api/wham/usage"))
.respond_with(
ResponseTemplate::new(402)
.set_body_json(json!({ "detail": { "code": "deactivated_workspace" } })),
)
.expect(1)
.mount(&server)
.await;
let (app, _state, pool, _dir) = admin_accounts_test_app_with_api_base_url(
"admin-account-quota-deactivated",
89,
format!("{}/backend-api", server.uri()),
)
.await;
seed_account(
&pool,
NewAccount {
id: "acct_quota_deactivated".to_string(),
email: Some("quota-deactivated@example.com".to_string()),
account_id: Some("chatgpt-quota-deactivated".to_string()),
user_id: None,
label: None,
plan_type: Some("plus".to_string()),
access_token: SecretString::new("access-quota-deactivated".to_string().into()),
refresh_token: None,
access_token_expires_at: None,
status: AccountStatus::Active,
added_at: None,
},
)
.await;
let response = app
.oneshot(
Request::builder()
.method("GET")
.uri("/api/admin/accounts/quota?id=acct_quota_deactivated")
.header("cookie", "cpr_admin_session=session_1")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let status: String = sqlx::query_scalar("select status from accounts where id = $1")
.bind("acct_quota_deactivated")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
assert_eq!(status, "banned");
}
backend/tests/api/admin/usage_routes.rs
+35 -4
View File
@@ -138,10 +138,24 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() {
.execute(usage.pool())
.await
.unwrap();
sqlx::query(
"insert into client_api_keys
(id, name, prefix, key, enabled, created_at, last_used_at)
values ($1, $2, $3, $4, true, $5, null)",
)
.bind("key_diagnostic")
.bind("生产密钥")
.bind("sk-diagnostic")
.bind("sk-diagnostic-secret")
.bind(now)
.execute(usage.pool())
.await
.unwrap();
let mut success = success_record("eventually succeeded");
success.id = "usage_insights_success".to_string();
success.request_id = Some("req_retried".to_string());
success.client_api_key_id = Some("key_diagnostic".to_string());
success.input_tokens = Some(100);
success.output_tokens = Some(20);
success.cached_tokens = Some(20);
@@ -157,14 +171,18 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() {
retried_error.created_at = now - chrono::Duration::seconds(2);
ops.append(&retried_error).await.unwrap();
for (id, failure_class, seconds) in [
("ops_insights_failed_older", "upstream", 2),
("ops_insights_failed_latest", "response_failed", 1),
for (id, failure_class, seconds, persist_class) in [
("ops_insights_failed_older", "upstream", 2, true),
("ops_insights_failed_latest", "response_failed", 1, false),
] {
let mut error = OpsErrorLog::new("upstream", failure_class);
error.id = id.to_string();
error.request_id = Some("req_failed".to_string());
error.failure_class = Some(failure_class.to_string());
if persist_class {
error.failure_class = Some(failure_class.to_string());
} else {
error.metadata = json!({ "upstreamCode": failure_class });
}
error.created_at = now - chrono::Duration::seconds(seconds);
ops.append(&error).await.unwrap();
}
@@ -211,6 +229,19 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() {
.iter()
.any(|item| item["name"] == "usage-account@example.com → acct_usage")
);
let api_key_diagnostics = admin_get(
&app,
"/api/admin/usage/records/insights/diagnostics?dimension=apiKey",
)
.await;
assert!(
api_key_diagnostics["data"]["items"]
.as_array()
.unwrap()
.iter()
.any(|item| item["name"] == "生产密钥")
);
}
#[tokio::test]
backend/tests/dispatch/service/usage_logging.rs
+1
View File
@@ -796,6 +796,7 @@ async fn responses_stream_should_record_usage_record_after_late_disconnect() {
assert_eq!(metadata["stream"], true);
assert_eq!(metadata["failed"], true);
assert_eq!(metadata["upstreamCode"], "stream_disconnected");
assert_eq!(event.failure_class.as_deref(), Some("stream_disconnected"));
assert_eq!(metadata["failureSource"], "proxy");
assert_eq!(metadata["synthetic"], true);
assert_rate_limit_header(&metadata, "retry-after", "11");
backend/tests/fleet/account_failure.rs
+79
View File
@@ -0,0 +1,79 @@
use codex_proxy_rs::{
fleet::{
account::AccountStatus,
account_failure::{AccountFailureKind, AccountStateEffect, classify_client_failure},
},
upstream::openai::transport::{
CodexBackendTransport, CodexClientError, CodexUpstreamDiagnostics,
},
};
use reqwest::StatusCode;
#[test]
fn deactivated_workspace_should_be_banned_before_generic_payment_classification() {
let failure = classify(402, r#"{"detail":{"code":"deactivated_workspace"}}"#);
assert_eq!(failure.kind, AccountFailureKind::Banned);
assert_eq!(
failure.effect,
Some(AccountStateEffect::SetStatus(AccountStatus::Banned))
);
}
#[test]
fn generic_payment_required_should_exhaust_quota() {
let failure = classify(402, r#"{"detail":{"code":"payment_required"}}"#);
assert_eq!(failure.kind, AccountFailureKind::QuotaExhausted);
assert_eq!(
failure.effect,
Some(AccountStateEffect::SetStatus(AccountStatus::QuotaExhausted))
);
}
#[test]
fn unauthorized_should_expire_account() {
let failure = classify(401, r#"{"error":{"message":"unauthorized"}}"#);
assert_eq!(failure.kind, AccountFailureKind::Expired);
assert_eq!(
failure.effect,
Some(AccountStateEffect::SetStatus(AccountStatus::Expired))
);
}
#[test]
fn rate_limit_should_create_temporary_quota_effect() {
let failure = classify(429, r#"{"error":{"code":"rate_limit_exceeded"}}"#);
assert_eq!(failure.kind, AccountFailureKind::RateLimited);
assert!(matches!(
failure.effect,
Some(AccountStateEffect::MarkQuotaLimitedUntil(_))
));
}
#[test]
fn html_forbidden_page_should_not_ban_account() {
let error = upstream_error(403, "<!doctype html><html>temporary edge failure</html>");
assert!(classify_client_failure(&error).is_none());
}
fn classify(
status: u16,
body: &str,
) -> codex_proxy_rs::fleet::account_failure::ClassifiedAccountFailure {
classify_client_failure(&upstream_error(status, body)).expect("failure should be classified")
}
fn upstream_error(status: u16, body: &str) -> CodexClientError {
CodexClientError::Upstream {
status: StatusCode::from_u16(status).unwrap(),
body: body.to_string(),
retry_after_seconds: None,
diagnostics: CodexUpstreamDiagnostics::with_status(status),
set_cookie_headers: Vec::new(),
transport: CodexBackendTransport::HttpSse,
}
}
backend/tests/fleet/mod.rs
+1
View File
@@ -1,3 +1,4 @@
pub(crate) mod account_failure;
pub(crate) mod cookies;
pub(crate) mod pool;
pub(crate) mod quota;
backend/tests/fleet/quota.rs
+68 -6
View File
@@ -1,13 +1,16 @@
use std::{collections::HashMap, sync::Arc, time::Duration as StdDuration};
use chrono::{Duration, Utc};
use codex_proxy_rs::fleet::account::AccountStatus;
use codex_proxy_rs::fleet::cookies::PgCookieStore;
use codex_proxy_rs::fleet::quota::{
QuotaRefreshService, quota_from_usage, quota_snapshot_limit_reached,
quota_snapshot_limit_window_seconds, quota_snapshot_reset_at,
use codex_proxy_rs::fleet::{
account::AccountStatus,
cookies::PgCookieStore,
pool::{AccountPoolOptions, AccountPoolService},
quota::{
QuotaRefreshService, quota_from_usage, quota_snapshot_limit_reached,
quota_snapshot_limit_window_seconds, quota_snapshot_reset_at,
},
store::{AccountStore, NewAccount, PgAccountStore},
};
use codex_proxy_rs::fleet::store::{NewAccount, PgAccountStore};
use codex_proxy_rs::infra::identity::AccountPseudonymizer;
use codex_proxy_rs::telemetry::account_usage::store::PgAccountUsageStore;
use codex_proxy_rs::upstream::openai::transport::CodexBackendClient;
@@ -24,6 +27,15 @@ fn test_account_pseudonymizer() -> Arc<AccountPseudonymizer> {
Arc::new(AccountPseudonymizer::new([7; 32]))
}
fn test_account_pool(pool: &sqlx::PgPool) -> Arc<AccountPoolService> {
Arc::new(AccountPoolService::new(
Arc::new(PgAccountStore::new(pool.clone())) as Arc<dyn AccountStore>,
Arc::new(PgAccountUsageStore::new(pool.clone())),
AccountPoolOptions::default(),
0,
))
}
#[test]
fn quota_from_usage_should_preserve_spend_control_individual_limit() {
let usage = json!({
@@ -321,6 +333,7 @@ async fn quota_refresh_service_should_send_usage_cookie_when_cookie_store_is_con
store,
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
)
.with_cookie_store(cookies);
@@ -403,6 +416,7 @@ async fn quota_refresh_service_should_fetch_usage_for_quota_locked_accounts_and_
store.clone(),
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
);
@@ -435,6 +449,49 @@ async fn quota_refresh_service_should_fetch_usage_for_quota_locked_accounts_and_
);
}
#[tokio::test]
async fn quota_refresh_service_should_ban_deactivated_workspace() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(
ResponseTemplate::new(402)
.set_body_json(json!({ "detail": { "code": "deactivated_workspace" } })),
)
.expect(1)
.mount(&server)
.await;
let (pool, _guard) = init_test_db("quota-refresh-deactivated-workspace").await;
let store = PgAccountStore::new(pool.clone());
insert_quota_locked_account(&store, &pool, "acct-deactivated", "access-token").await;
let codex = Arc::new(CodexBackendClient::new(
reqwest::Client::new(),
server.uri(),
crate::support::fingerprint::runtime_test_fingerprint(),
));
let service = QuotaRefreshService::new(
store,
PgAccountUsageStore::new(pool.clone()),
codex,
test_account_pool(&pool),
test_account_pseudonymizer(),
);
let summary = service
.refresh_locked_accounts(&mut HashMap::new())
.await
.expect("quota refresh should complete");
let status: String = sqlx::query_scalar("select status from accounts where id = $1")
.bind("acct-deactivated")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(summary.failed, 1);
assert_eq!(status, "banned");
}
#[tokio::test]
async fn quota_refresh_service_should_refresh_quota_exhausted_accounts() {
let server = MockServer::start().await;
@@ -479,6 +536,7 @@ async fn quota_refresh_service_should_refresh_quota_exhausted_accounts() {
store,
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
);
@@ -515,6 +573,7 @@ async fn quota_refresh_service_should_skip_recent_locked_account_before_cooldown
store,
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
1_800,
);
@@ -555,6 +614,7 @@ async fn quota_refresh_service_should_skip_recent_locked_account_inside_cooldown
store,
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
1_800,
);
@@ -612,6 +672,7 @@ async fn quota_refresh_service_should_bypass_recent_skip_after_cooldown_grace()
store,
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
1_800,
);
@@ -667,6 +728,7 @@ async fn quota_refresh_service_should_stagger_multiple_locked_account_requests()
store,
PgAccountUsageStore::new(pool.clone()),
Arc::new(codex),
test_account_pool(&pool),
test_account_pseudonymizer(),
)
.with_request_spacing(StdDuration::from_millis(200));
backend/tests/support/storage.rs
+27 -3
View File
@@ -3,7 +3,7 @@ use std::{env, sync::OnceLock, thread};
use chrono::{DateTime, Utc};
use codex_proxy_rs::{
auth::store::{PgAdminUserStore, RedisAdminSessionStore},
bootstrap::services::BackgroundTaskStores,
bootstrap::{config::BootstrapConfig, services::BackgroundTaskStores},
dispatch::affinity::RedisSessionAffinityStore,
fleet::{cookies::PgCookieStore, refresh::RedisRefreshLeaseStore, store::PgAccountStore},
infra::{database::migrate, redis::RedisConnection},
@@ -24,6 +24,7 @@ const DEFAULT_TEST_REDIS_URL: &str = "redis://127.0.0.1:6379";
static DATABASE_INIT_LIMIT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(2);
static REDIS_TEST_CLEANUP: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();
static REDIS_TEST_RUN_ID: OnceLock<String> = OnceLock::new();
static LOCAL_BOOTSTRAP_CONFIG: OnceLock<Option<BootstrapConfig>> = OnceLock::new();
pub(crate) struct TestDatabaseGuard {
database_name: Option<String>,
@@ -186,11 +187,34 @@ pub(crate) fn background_task_stores(pool: PgPool, redis: RedisConnection) -> Ba
}
pub(crate) fn test_database_url() -> String {
env::var("CPR_TEST_DATABASE_URL").unwrap_or_else(|_| DEFAULT_TEST_DATABASE_URL.to_string())
// 显式测试拓扑优先；本地开发默认跟随 Compose 配置；无本地配置的 CI 使用固定测试值。
env::var("CPR_TEST_DATABASE_URL")
.ok()
.or_else(|| local_bootstrap_config().map(|config| config.database_url().to_string()))
.unwrap_or_else(|| DEFAULT_TEST_DATABASE_URL.to_string())
}
pub(crate) fn test_redis_url() -> String {
env::var("CPR_TEST_REDIS_URL").unwrap_or_else(|_| DEFAULT_TEST_REDIS_URL.to_string())
env::var("CPR_TEST_REDIS_URL")
.ok()
.or_else(|| local_bootstrap_config().map(|config| config.redis_url().to_string()))
.unwrap_or_else(|| DEFAULT_TEST_REDIS_URL.to_string())
}
fn local_bootstrap_config() -> Option<&'static BootstrapConfig> {
LOCAL_BOOTSTRAP_CONFIG
.get_or_init(|| {
let path =
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../deploy/config.yaml");
if !path.is_file() {
return None;
}
Some(
BootstrapConfig::load_from_path(path)
.unwrap_or_else(|error| panic!("load local test service config: {error}")),
)
})
.as_ref()
}
pub(crate) fn timestamp(value: &str) -> DateTime<Utc> {
backend/tests/upstream/openai/transport/client.rs
+1 -17
View File
@@ -5,27 +5,11 @@ use std::sync::{
use super::{accept_codex_test_websocket, completed_websocket_response, request_context};
use codex_proxy_rs::upstream::openai::protocol::responses::CodexResponsesRequest;
use codex_proxy_rs::upstream::openai::transport::{
CodexBackendClient, CodexWebSocketPool, is_deactivated_workspace_error_body,
};
use codex_proxy_rs::upstream::openai::transport::{CodexBackendClient, CodexWebSocketPool};
use futures::{SinkExt, StreamExt};
use tokio::net::TcpListener;
use tokio_tungstenite::tungstenite::Message;
#[test]
fn deactivated_workspace_error_body_should_match_detail_code() {
let body = r#"{"detail":{"code":"deactivated_workspace"}}"#;
assert!(is_deactivated_workspace_error_body(body));
}
#[test]
fn deactivated_workspace_error_body_should_not_match_other_json_paths() {
let body = r#"{"error":{"code":"deactivated_workspace"}}"#;
assert!(!is_deactivated_workspace_error_body(body));
}
#[tokio::test]
async fn codex_backend_client_should_apply_configured_websocket_pool() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
docs/architecture.md
+15 -10
View File
@@ -28,11 +28,11 @@ PostgreSQL 和 Redis 都是启动必需依赖。运行中任一健康检查失
## 架构不变量
1. `v1/*` 使用 Request → Attempt → Stream 三层静态洋葱生命周期；管理端、健康检查和静态资源不进入业务洋葱。
2. 一个功能只有一个规则 owner。阈值、分类、状态机和 decision 只能存在于对应 controller。
2. 一个功能只有一个规则 owner。跨入口的账号失败语义属于 `fleet/account_failure.rs`；`v1/*` 的 retry、exhaustion 和 commit decision 属于对应 controller。
3. API 只解析和编码；lifecycle 只编排顺序、retry、commit 和 finalize；transport 只建立连接并产出 typed facts。
4. 上游响应一旦可能收到 payload，transport 或账号都不得自动重放同一请求。
5. 账号认证身份按 attempt 重建；客户端会话拓扑按原值透传；两者不能共用一个“身份”抽象。
6. 账号明确失效先在线性化的内存路径生效并驱逐 WebSocket，再异步持久化 PostgreSQL。
6. `v1/*` 热路径中的账号明确失效先在线性化的内存路径生效并驱逐 WebSocket，再异步持久化 PostgreSQL；管理端和后台任务可等待同一状态 effect 持久化完成。
7. 每项生产能力只有一条执行路径，不设置兼容 wrapper、双状态机或运行时切换的新旧实现。
8. `backend/src/` 只放生产代码；所有测试和 test-only helper 位于 `backend/tests/`。
@@ -104,6 +104,7 @@ backend/src/
```text
upstream/openai/
├── token_client.rs
├── failure.rs
├── protocol/
│ ├── responses.rs
│ ├── events.rs
@@ -127,6 +128,7 @@ upstream/openai/
```
- `protocol` 定义 Responses body、canonical SSE/WS 事实和序列化，不包含账号策略。
- `failure.rs` 将各 transport 错误规范化为稳定的 `UpstreamFailureFacts`，不解释账号状态。
- `transport` 拥有 HTTP/SSE、WebSocket、TLS、header、连接池、熔断和 transport metrics。
- `fingerprint` 拥有 Codex Desktop 指纹的存储、运行时快照和更新。
- `token_client.rs` 只实现 OAuth token 刷新协议。
@@ -137,6 +139,7 @@ upstream/openai/
- `store`：账号与 quota 的 PostgreSQL 映射和事务写入。
- `pool`：进程内账号快照、并发 slot、请求间隔和候选 lease。
- `account_failure.rs`：HTTP、SSE、手动额度查询和后台额度刷新共享的账号失败分类与状态 effect。
- `scheduler`：候选过滤与排序的唯一 owner，支持 `smart`、`quota_reset_priority`、`round_robin`、`sticky`。
- `scheduler/feedback`：进程内 EWMA 错误率和首字延迟。
- `quota`：配额查询、窗口与运行时限制。
@@ -225,15 +228,16 @@ Stream terminal 包括 completed、incomplete、failed、上游断开、下游
| --- | --- |
| previous response 所有权、scope、full replay | `controllers/history.rs` |
| cyber policy 会话排除和 CAS 清理 | `controllers/cyber_policy/` |
| quota 与 rate limit | `controllers/quota.rs` |
| 封禁、过期、模型不支持等账号事实 | `controllers/account_failure.rs` |
| 内存失效、异步持久化和 WS 驱逐 effect | `controllers/account_state.rs` |
| v1 quota 与 rate limit attempt decision | `controllers/quota.rs` |
| v1 封禁、过期、模型不支持的 exhaustion decision | `controllers/account_failure.rs` |
| transport 错误到稳定上游事实 | `upstream/openai/failure.rs` |
| 跨入口账号失败分类、状态 effect 与 WS 驱逐 | `fleet/account_failure.rs` |
| Cloudflare Cookie 与冷却 | `controllers/cloudflare.rs` |
| 会话亲和写入 | `controllers/affinity.rs` |
| usage 结算 | `controllers/usage.rs` |
| 代理事件与 transport metrics | `controllers/telemetry/` |
Controller 之间不直接调用。跨功能优先级只在 `ControllerSet` 声明；API、transport、store 和 lifecycle 不重复识别功能语义。
Controller 之间不直接调用。跨功能优先级只在 `ControllerSet` 声明；API、transport、store 和 lifecycle 不重复识别功能语义。Controller 只把共享账号分类翻译为 v1 生命周期 decision；手动额度查询和后台额度刷新执行同一状态 effect，但各自保留自己的返回与调度职责。
## 上游双热传输
@@ -327,10 +331,11 @@ Scheduler 先过滤不可用账号，再按运行时 `rotation_strategy` 排序
明确上游事实的处理顺序：
1. transport 提取 status、code、type、message、retry-after 等事实。
2. controller 分类为 quota、rate limit、expired、disabled、banned、model unsupported、Cloudflare 或 cyber policy。
3. 对账号全局事实，账号池 mutex 内立即失效 runtime state 并移除 slot。
4. 驱逐该账号的 idle/busy/connecting WebSocket entry。
5. lifecycle 从冻结候选中选择下一账号。
2. `fleet/account_failure.rs` 将跨入口账号事实统一分类为 quota、rate limit、expired、disabled、banned 或 model unsupported；Cloudflare 与 cyber policy 保持各自 owner。
3. v1 controller 把共享分类翻译为 retry/exhaustion decision。
4. 对账号全局事实，账号池 mutex 内立即失效 runtime state 并移除 slot。
5. 驱逐该账号的 idle/busy/connecting WebSocket entry。
6. lifecycle 从冻结候选中选择下一账号。
6. PostgreSQL 状态写入后台有序队列。
数据库延迟不会延长当前换号路径，也不会使并发请求重新租到已失效账号。
frontend/src/views/usage/components/UsageDiagnosticCard.vue
+8 -2
View File
@@ -79,7 +79,9 @@ const displayItems = computed(() =>
const tableRows = computed(() => (isStale.value ? [] : displayItems.value))
function diagnosticNameDisplay(name: string) {
const full = name.trim() || '未知'
const raw = name.trim() || '未知'
const full =
dimension.value === 'transport' ? ({ websocket: 'WS', http_sse: 'SSE' }[raw] ?? raw) : raw
if (dimension.value !== 'model' && dimension.value !== 'account') {
return { primary: full, secondary: '', full }
}
@@ -141,7 +143,11 @@ function diagnosticNameDisplay(name: string) {
v-if="row.nameDisplay.secondary"
class="flex min-w-0 items-center gap-1.25 text-(--cp-text-secondary)"
>
<CornerDownRight class="size-3.25 shrink-0 text-(--cp-info)" stroke-width="2.4" />
<CornerDownRight
v-if="dimension !== 'account'"
class="size-3.25 shrink-0 text-(--cp-info)"
stroke-width="2.4"
/>
<code class="block truncate font-mono text-[11px] leading-none font-bold">
{{ row.nameDisplay.secondary }}
</code>