Compare commits

...
12 Commits
Author SHA1 Message Date
zyycnandCodex ab4bf75d0e chore: bump version to 3.0.30
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 21:43:11 +08:00
zyycnandCodex 7a9715ed41 chore(db): fold schema updates into initial migration
Collapse the completed response ID, service tier, and quota cooldown changes into the fresh-database baseline. Refresh the migration contract and state consistently that cross-major online upgrades are unsupported.

Co-authored-by: Codex <noreply@openai.com>
2026-07-30 21:42:57 +08:00
zyycnandCodex c353a6159e fix(openai): preserve failover before stream commit
Hold structural response events for a bounded replay grace so a split-frame 429 can rotate accounts without disconnecting the client. Release immediately on semantic output and keep post-commit failures non-replayable.

Co-authored-by: Codex <noreply@openai.com>
2026-07-30 21:42:30 +08:00
zyycnandCodex f5dca4848a fix(openai): harden exhausted quota recovery
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 21:07:53 +08:00
zyycnandCodex 36321a9af4 chore: bump version to 3.0.29
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 04:03:58 +08:00
zyycnandCodex b485b8bd4a fix(openai): reconcile exhausted account quotas
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 04:03:04 +08:00
zyycnandCodex cae2c8656a fix(xai): preserve quota exhaustion state
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 03:10:35 +08:00
zyycnandCodex acf574129f chore: bump version to 3.0.28
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 02:26:35 +08:00
zyycnandCodex 6426e0d3de fix(accounts): align availability summaries
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 02:22:23 +08:00
zyycnandCodex eedc39b33a chore: bump version to 3.0.27
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 01:55:51 +08:00
zyycnandCodex 9672a2e000 fix(xai): prioritize structured quota signals
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 01:53:43 +08:00
zyycnandCodex 0fffd2c550 fix(accounts): converge account state and imports
Co-authored-by: Codex <noreply@openai.com>
2026-07-30 01:23:42 +08:00
54 changed files with 1855 additions and 811 deletions
+1 -1
View File
@@ -626,7 +626,7 @@ jobs:
make_latest: ${{ needs.release-info.outputs.prerelease == 'true' && 'false' || 'true' }}
body: |
> [!WARNING]
> **Upgrading across major versions?** Database schema and configuration are only compatible within a major version, and the one-click self-update refuses cross-major targets (releases before 3.0 lack this gate — do not use their built-in update to reach a newer major). Deploy the new major fresh (new `.runtime/` data directory) and re-import or re-authorize provider accounts and client keys. See the [upgrade notes](https://github.com/${{ github.repository }}/blob/main/deploy/README.md#镜像升级与源码构建).
> **Upgrading across major versions?** Online upgrades across major versions are not supported. Deploy the new major fresh with a new `.runtime/` data directory, then re-import or re-authorize provider accounts and client keys. See the [upgrade notes](https://github.com/${{ github.repository }}/blob/main/deploy/README.md#镜像升级与源码构建).
## 📥 Installation
+2 -4
View File
@@ -168,10 +168,8 @@ docker compose -f deploy/compose.yaml up -d
> 运行状态。
> [!WARNING]
> 跨大版本不是原地升级:数据库 schema 与配置只在同一大版本内兼容,内置一键自更新
> 会拒绝跨大版本目标(3.0 之前的旧版没有这道门禁,不要用旧版界面的一键更新跨大版本)。
> 升级到新的大版本需要全新部署(全新 `.runtime/` 数据目录),并重新导入或重新授权
> Provider 账号与客户端 Key。
> 不支持跨大版本在线升级。升级到新的大版本时,请使用全新的 `.runtime/` 数据目录重新部署,
> 并重新导入或重新授权 Provider 账号与客户端 Key。
Compose 默认只绑定 `127.0.0.1`。公网接入应使用 HTTPS 反向代理,转发 WebSocket
upgrade 与真实客户端 IP;不要暴露 PostgreSQL 或 Redis。
@@ -169,6 +169,8 @@ pub struct AccountPage {
}
/// 统一账号目录的全局状态计数,不受当前筛选和分页影响。
///
/// 配额耗尽是可恢复状态,始终单列;`unavailable` 只包含过期、禁用和封禁账号。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AccountSummary {
pub total: u64,
@@ -460,6 +460,10 @@ pub struct RequestMetricPoint {
}
/// 账号池容量统计。
///
/// 首页概览的 `unavailable` 表示不可参与调度,包含配额耗尽及过期、禁用和封禁账号。
///
/// `quota_exhausted` 仍单列,以便状态详情展示和诊断。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AccountPoolMetrics {
pub total: u64,
@@ -374,6 +374,7 @@ impl ProviderAdmin for FakeProviderAdmin {
pub(super) struct FakeAccountStore {
events: EventLog,
account: AccountRecord,
account_after_probe: Mutex<Option<AccountRecord>>,
fail_commit: Mutex<bool>,
audit_requests: Mutex<Vec<String>>,
quota_window_usage: Mutex<Vec<AccountUsageWindowResult>>,
@@ -388,6 +389,7 @@ impl FakeAccountStore {
Arc::new(Self {
events,
account,
account_after_probe: Mutex::new(None),
fail_commit: Mutex::new(false),
audit_requests: Mutex::new(Vec::new()),
quota_window_usage: Mutex::new(Vec::new()),
@@ -406,6 +408,13 @@ impl FakeAccountStore {
*self.quota_window_usage.lock().expect("quota window usage") = usage;
}
fn set_account_after_probe(&self, account: AccountRecord) {
*self
.account_after_probe
.lock()
.expect("account after probe") = Some(account);
}
fn record(&self, event: &'static str) {
self.events.lock().expect("store events").push(event);
}
@@ -445,7 +454,13 @@ impl AccountStore for FakeAccountStore {
async fn load_account(&self, account_id: &str) -> AdminStoreResult<Option<AccountRecord>> {
self.record("store.load_account");
Ok((account_id == self.account.id).then(|| self.account.clone()))
let account = self
.account_after_probe
.lock()
.expect("account after probe")
.clone()
.unwrap_or_else(|| self.account.clone());
Ok((account_id == account.id.as_str()).then_some(account))
}
async fn load_account_usage(
@@ -802,6 +817,36 @@ async fn connection_test_should_probe_unavailable_account() {
));
}
#[tokio::test]
async fn connection_test_reloads_resettable_usage_limit_status_after_probe() {
let events = events();
let provider = FakeProviderAdmin::new("xai", events.clone());
let store = FakeAccountStore::new("xai", events);
let probe = Arc::new(FreeModelQuotaProbe {
store: Arc::clone(&store),
});
let services = accounts_service_with_probe(provider, store, probe).await;
let events = services
.accounts()
.test_connection(
ProviderAccountId::new("acct_test").expect("account ID"),
gateway_core::routing::UpstreamModelId::new("grok-4.5").expect("model"),
)
.await
.expect("connection test stream")
.collect::<Vec<_>>()
.await;
assert!(matches!(
events.last(),
Some(AccountConnectionTestEvent::Failed {
account_status: gateway_admin::model::accounts::AccountStatus::QuotaExhausted,
..
})
));
}
#[tokio::test]
async fn connection_test_should_preserve_disabled_account_status() {
let provider = FakeProviderAdmin::new("xai", events());
@@ -1234,6 +1279,30 @@ impl AccountProbe for FailingAccountProbe {
}
}
struct FreeModelQuotaProbe {
store: Arc<FakeAccountStore>,
}
impl AccountProbe for FreeModelQuotaProbe {
fn probe(
&self,
_: AccountProbeRequest,
) -> BoxFuture<'_, Result<AccountProbeResult, AccountProbeError>> {
Box::pin(async move {
let mut account = account_record("xai");
account.availability = AccountAvailability::Cooldown;
account.availability_reason = Some("usage_limit_exhausted".to_owned());
account.cooldown_until = Some(Utc::now() + TimeDelta::hours(24));
self.store.set_account_after_probe(account);
Err(GatewayError::new(
GatewayErrorKind::RateLimited,
"xAI free model quota is exhausted",
)
.into())
})
}
}
fn store_unavailable() -> AdminStoreError {
AdminStoreError::new(
AdminStoreErrorKind::Unavailable,
@@ -1653,12 +1653,7 @@ fn format_token_price(cost: &domain::CurrencyCost) -> String {
if cost.currency != "USD" {
return format!("{} {} / 1M Token", cost.currency, cost.amount.as_str());
}
let value = cost
.amount
.as_str()
.parse::<f64>()
.map_or(0.0, |value| value);
format!("${value:.4} / 1M Token")
format!("${} / 1M Token", cost.amount.as_str())
}
fn format_service_tier(service_tier: Option<&str>) -> String {
@@ -340,7 +340,7 @@ mod actions {
upstream_status: Some(429),
upstream_content_type: Some("application/json".to_owned()),
upstream_body: Some(r#"{"error":{"type":"usage_limit_reached"}}"#.to_owned()),
account_status: AccountStatus::Active,
account_status: AccountStatus::QuotaExhausted,
},
]
.map(|event| AccountConnectionTestEvent::from(event).data);
@@ -374,7 +374,7 @@ mod actions {
"upstreamStatus": 429,
"upstreamContentType": "application/json",
"upstreamBody": r#"{"error":{"type":"usage_limit_reached"}}"#,
"accountStatus": "active"
"accountStatus": "quota_exhausted"
}),
]
);
+10 -1
View File
@@ -79,6 +79,7 @@ pub(super) struct AdminTestFixture {
pub auth: Arc<MemoryAuthStore>,
pub settings: Arc<MemorySettingsStore>,
pub usage_records: Arc<Mutex<Vec<UsageRecord>>>,
pub dashboard_observation: Arc<Mutex<Option<DashboardObservation>>>,
}
impl AdminTestFixture {
@@ -92,8 +93,10 @@ impl AdminTestFixture {
let settings = Arc::new(MemorySettingsStore::new(api_key));
let client_keys = Arc::new(MemoryClientKeyStore);
let usage_records = Arc::new(Mutex::new(Vec::new()));
let dashboard_observation = Arc::new(Mutex::new(None));
let unused = Arc::new(UnusedStore {
usage_records: Arc::clone(&usage_records),
dashboard_observation: Arc::clone(&dashboard_observation),
});
let stores = AdminStorePorts::new(
unused.clone(),
@@ -125,6 +128,7 @@ impl AdminTestFixture {
auth,
settings,
usage_records,
dashboard_observation,
}
}
@@ -392,6 +396,7 @@ impl ClientKeyStore for MemoryClientKeyStore {
struct UnusedStore {
usage_records: Arc<Mutex<Vec<UsageRecord>>>,
dashboard_observation: Arc<Mutex<Option<DashboardObservation>>>,
}
#[async_trait]
@@ -503,7 +508,11 @@ impl AccountStore for UnusedStore {
#[async_trait]
impl ObservabilityStore for UnusedStore {
async fn dashboard_summary(&self, _: TimeRange) -> AdminStoreResult<DashboardObservation> {
Err(unavailable("dashboard"))
self.dashboard_observation
.lock()
.expect("dashboard observation")
.clone()
.ok_or_else(|| unavailable("dashboard"))
}
async fn dashboard_trend(&self, _: TimeRange) -> AdminStoreResult<Vec<RequestMetricPoint>> {
@@ -260,6 +260,80 @@ mod response {
}
}
#[tokio::test]
async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headline_count() {
use axum::{
body::{Body, to_bytes},
http::{Request, StatusCode, header},
};
use chrono::{Duration, Utc};
use gateway_admin::model::observability::{
AccountPoolMetrics, AttemptMetrics, DashboardObservation, RequestMetrics, TimeRange,
};
use gateway_api::admin::observability;
use tower::ServiceExt as _;
use super::{AdminTestFixture, AdminTestState};
let fixture = AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
let now = Utc::now();
*fixture
.dashboard_observation
.lock()
.expect("dashboard observation") = Some(DashboardObservation {
range: TimeRange::new(now - Duration::hours(1), now).expect("dashboard range"),
requests: RequestMetrics::default(),
attempts: AttemptMetrics::default(),
provider_accounts: AccountPoolMetrics {
total: 807,
enabled: 804,
unavailable: 792,
active: 15,
expired: 2,
quota_exhausted: 785,
refreshing: None,
disabled: 3,
banned: 2,
},
trend: Vec::new(),
account_usage: Vec::new(),
recent_requests: Vec::new(),
});
let response = observability::router::<AdminTestState>()
.with_state(fixture.state())
.oneshot(
Request::builder()
.uri("/api/admin/dashboard/summary")
.header(header::COOKIE, "cpr_admin_session=valid-session")
.header("x-request-id", "req_dashboard_account_summary")
.body(Body::empty())
.expect("dashboard request"),
)
.await
.expect("dashboard response");
let status = response.status();
let body = to_bytes(response.into_body(), 1024 * 1024)
.await
.expect("dashboard response body");
let value: serde_json::Value = serde_json::from_slice(&body).expect("dashboard response JSON");
assert_eq!(status, StatusCode::OK, "{value}");
assert_eq!(
(
&value["data"]["cards"]["credentials"]["totalValue"],
&value["data"]["cards"]["credentials"]["availableValue"],
&value["data"]["cards"]["credentials"]["unavailableValue"],
),
(
&serde_json::json!(807),
&serde_json::json!(15),
&serde_json::json!(792)
),
);
}
#[tokio::test]
async fn usage_route_should_forward_a_bounded_unknown_outcome_filter() {
use axum::{
@@ -291,12 +365,17 @@ async fn usage_route_should_forward_a_bounded_unknown_outcome_filter() {
#[tokio::test]
async fn usage_route_should_expose_image_and_websocket_facts() {
use std::str::FromStr as _;
use axum::{
body::{Body, to_bytes},
http::{Request, StatusCode, header},
};
use chrono::Utc;
use gateway_admin::model::observability::{RequestOutcome, UsageRecord};
use gateway_admin::model::observability::{
CalculatedBillingBreakdown, CurrencyCost, DecimalAmount, RequestOutcome, UsageBilling,
UsageRecord,
};
use gateway_api::admin::observability;
use tower::ServiceExt as _;
@@ -304,6 +383,10 @@ async fn usage_route_should_expose_image_and_websocket_facts() {
let fixture = AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
let usd = |amount: &str| CurrencyCost {
currency: "USD".to_owned(),
amount: DecimalAmount::from_str(amount).expect("USD amount"),
};
fixture
.usage_records
.lock()
@@ -359,7 +442,22 @@ async fn usage_route_should_expose_image_and_websocket_facts() {
cost_source: "unavailable".to_owned(),
cost_amount: None,
cost_currency: None,
billing: None,
billing: Some(UsageBilling::Calculated(Box::new(
CalculatedBillingBreakdown {
input_amount: usd("0.03"),
output_amount: usd("0.07"),
cache_read_amount: usd("0.00"),
cache_write_amount: usd("0.00"),
standard_amount: usd("0.10"),
total_amount: usd("0.10"),
input_price_per_million: usd("10.0000"),
output_price_per_million: usd("60.0000"),
cache_read_price_per_million: usd("1.0000"),
cache_write_price_per_million: usd("12.5000"),
service_tier: Some("priority".to_owned()),
multiplier_percent: 100,
},
))),
transport_decision_wait_ms: Some(7),
connect_ms: Some(11),
headers_ms: Some(13),
@@ -400,6 +498,19 @@ async fn usage_route_should_expose_image_and_websocket_facts() {
.expect("usage response body");
let value: serde_json::Value = serde_json::from_slice(&body).expect("usage response JSON");
assert_eq!(
value["data"]["items"][0]["billing"]["inputPriceDisplay"],
"$10 / 1M Token"
);
assert_eq!(
value["data"]["items"][0]["billing"]["outputPriceDisplay"],
"$60 / 1M Token"
);
assert_eq!(
value["data"]["items"][0]["billing"]["cacheWritePriceDisplay"],
"$12.5 / 1M Token"
);
assert_eq!(
serde_json::json!({
"route": value["data"]["items"][0]["route"],
@@ -2013,8 +2013,7 @@ async fn provider_account_metrics(
)
select count(*)::bigint as total,
count(*) filter (where enabled)::bigint as enabled,
count(*) filter (where status in ('expired', 'disabled', 'banned'))::bigint
as unavailable,
count(*) filter (where status <> 'active')::bigint as unavailable,
count(*) filter (where status = 'active')::bigint as active,
count(*) filter (where status = 'expired')::bigint as expired,
count(*) filter (where status = 'quota_exhausted')::bigint as quota_exhausted,
@@ -630,7 +630,8 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
else availability_observed_at
end,
updated_at = case when enabled then greatest(now(), $6) else updated_at end
where id = $1 and credential_revision = $2",
where id = $1 and credential_revision = $2
and (availability_observed_at is null or availability_observed_at <= $6)",
)
.bind(update.account_id)
.bind(to_i64(update.expected_revision.get())?)
@@ -675,7 +676,8 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
"update provider_accounts
set provider_quota_json = $3, quota_observed_at = $4,
updated_at = greatest(now(), coalesce($4, now()))
where id = $1 and credential_revision = $2",
where id = $1 and credential_revision = $2
and ($4 is null or quota_observed_at is null or quota_observed_at <= $4)",
)
.bind(account_id)
.bind(to_i64(expected_revision.get())?)
@@ -1610,43 +1612,26 @@ fn admin_account_summary(
accounts: &[ProviderAccountSummary],
now: DateTime<Utc>,
) -> AccountSummary {
let total = u64::try_from(accounts.len()).unwrap_or(u64::MAX);
let active = u64::try_from(
accounts
.iter()
.filter(|account| admin_account_status(account, now) == AdminAccountStatus::Active)
.count(),
)
.unwrap_or(u64::MAX);
let quota_exhausted = u64::try_from(
accounts
.iter()
.filter(|account| {
admin_account_status(account, now) == AdminAccountStatus::QuotaExhausted
})
.count(),
)
.unwrap_or(u64::MAX);
let unavailable = u64::try_from(
accounts
.iter()
.filter(|account| {
matches!(
admin_account_status(account, now),
AdminAccountStatus::Expired
| AdminAccountStatus::Disabled
| AdminAccountStatus::Banned
)
})
.count(),
)
.unwrap_or(u64::MAX);
AccountSummary {
total,
active,
quota_exhausted,
unavailable,
let mut summary = AccountSummary {
total: u64::try_from(accounts.len()).unwrap_or(u64::MAX),
active: 0,
quota_exhausted: 0,
unavailable: 0,
};
for account in accounts {
match admin_account_status(account, now) {
AdminAccountStatus::Active => summary.active = summary.active.saturating_add(1),
AdminAccountStatus::QuotaExhausted => {
summary.quota_exhausted = summary.quota_exhausted.saturating_add(1);
}
AdminAccountStatus::Expired
| AdminAccountStatus::Disabled
| AdminAccountStatus::Banned => {
summary.unavailable = summary.unavailable.saturating_add(1);
}
}
}
summary
}
fn sort_admin_account_items(items: &mut [AdminAccountListItem], sort: Option<AdminAccountSort>) {
@@ -58,30 +58,6 @@ impl TestDatabase {
.execute(&mut *migration)
.await
.expect("apply terminal migration");
sqlx::raw_sql(include_str!(
"../../../../migrations/0002_drop_opaque_response_id_index.sql"
))
.execute(&mut *migration)
.await
.expect("apply opaque response ID migration");
sqlx::raw_sql(include_str!(
"../../../../migrations/0003_store_opaque_response_ids_as_bytes.sql"
))
.execute(&mut *migration)
.await
.expect("apply opaque response storage migration");
sqlx::raw_sql(include_str!(
"../../../../migrations/0004_add_model_request_service_tier.sql"
))
.execute(&mut *migration)
.await
.expect("apply model request service tier migration");
sqlx::raw_sql(include_str!(
"../../../../migrations/0005_allow_quota_exhausted_cooldown.sql"
))
.execute(&mut *migration)
.await
.expect("apply quota exhausted cooldown migration");
migration.commit().await.expect("commit terminal migration");
Some(Self {
admin,
@@ -177,25 +153,19 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly
.expect("drop migration test database");
admin.close().await;
assert_eq!((first_table_count, migration_count), (8, 5));
assert_eq!((first_table_count, migration_count), (8, 1));
assert_eq!(response_id_types, ["bytea", "bytea"]);
assert!(!raw_response_id_index_exists);
}
#[test]
fn migrations_should_leave_transaction_ownership_to_sqlx() {
let transaction_statements = [
include_str!("../../../../migrations/0001_initial.sql"),
include_str!("../../../../migrations/0002_drop_opaque_response_id_index.sql"),
include_str!("../../../../migrations/0003_store_opaque_response_ids_as_bytes.sql"),
include_str!("../../../../migrations/0004_add_model_request_service_tier.sql"),
include_str!("../../../../migrations/0005_allow_quota_exhausted_cooldown.sql"),
]
.into_iter()
.flat_map(str::lines)
.map(str::trim)
.filter(|line| matches!(*line, "begin;" | "commit;"))
.count();
let transaction_statements = [include_str!("../../../../migrations/0001_initial.sql")]
.into_iter()
.flat_map(str::lines)
.map(str::trim)
.filter(|line| matches!(*line, "begin;" | "commit;"))
.count();
assert_eq!(transaction_statements, 0);
}
@@ -112,7 +112,7 @@ async fn observability_preserves_and_filters_opaque_response_ids() {
}
#[tokio::test]
async fn dashboard_account_metrics_should_partition_available_and_unavailable_accounts() {
async fn dashboard_account_metrics_should_partition_account_statuses() {
let Some(database) = TestDatabase::create("dashboard_account_metrics").await else {
return;
};
@@ -134,7 +134,7 @@ async fn dashboard_account_metrics_should_partition_available_and_unavailable_ac
('acct_rate_cooldown', 'openai', 'rate-cooldown', 'user-rate-cooldown', 'oauth',
'{}'::jsonb, 1, false, $1 + interval '1 day', true, 'cooldown',
'rate_limited', $1 + interval '2 hours', $1, $1, $1),
('acct_usage_limit', 'openai', 'usage-limit', 'user-usage-limit', 'oauth',
('acct_usage_limit', 'xai', 'usage-limit', 'user-usage-limit', 'oauth',
'{}'::jsonb, 1, false, $1 + interval '1 day', true, 'cooldown',
'usage_limit_exhausted', $1 + interval '2 hours', $1, $1, $1),
('acct_banned', 'xai', 'banned', 'user-banned', 'oauth',
@@ -169,8 +169,14 @@ async fn dashboard_account_metrics_should_partition_available_and_unavailable_ac
metrics.disabled,
metrics.banned,
),
(6, 5, 2, 3, 1, 1, 1, 1),
(6, 5, 2, 4, 1, 1, 1, 1),
);
// 首页按调度可用性聚合:配额耗尽也属于不可用。
assert_eq!(
metrics.unavailable,
metrics.quota_exhausted + metrics.expired + metrics.disabled + metrics.banned
);
assert_eq!(metrics.total, metrics.active + metrics.unavailable);
database.close().await;
}
@@ -111,6 +111,97 @@ async fn core_quota_batch_reads_only_observed_accounts_in_one_contract_call() {
database.close().await;
}
#[tokio::test]
async fn delayed_provider_observations_cannot_overwrite_newer_state_or_quota() {
let Some(database) = TestDatabase::create("provider_account_observation_fence").await else {
return;
};
let repository = PgProviderAccountRepository::new(database.pool.clone());
let account_id = ProviderAccountId::new("acct_observation_fence").expect("account id");
repository
.insert_provider_account(account(account_id.as_str(), "user-observation-fence"))
.await
.expect("insert observation fence fixture");
let revision = CredentialRevision::new(1).expect("revision");
let newer = SystemTime::now()
.checked_add(Duration::from_secs(60))
.expect("newer observation time");
let older = newer
.checked_sub(Duration::from_secs(30))
.expect("older observation time");
let quota = |marker: &str, observed_at| QuotaObservation {
account_id: account_id.clone(),
expected_revision: revision,
quota: Some(OpaqueProviderData::new(
json!({"marker": marker})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(observed_at),
};
assert_eq!(
repository
.compare_and_swap_quota(quota("newer", newer))
.await
.expect("persist newer quota"),
QuotaWriteOutcome::Updated
);
assert_eq!(
repository
.compare_and_swap_quota(quota("older", older))
.await
.expect("reject older quota"),
QuotaWriteOutcome::Conflict
);
repository
.apply_state_change(AccountStateChange {
account_id: account_id.clone(),
expected_revision: revision,
availability: AccountAvailability::QuotaExhausted,
reason: Some("quota_exhausted".to_owned()),
cooldown_until: Some(newer),
observed_at: newer,
})
.await
.expect("persist newer account state");
assert!(
repository
.apply_state_change(AccountStateChange {
account_id: account_id.clone(),
expected_revision: revision,
availability: AccountAvailability::Ready,
reason: None,
cooldown_until: None,
observed_at: older,
})
.await
.is_err(),
"a delayed state observation must conflict"
);
let current = repository
.get_account(&account_id)
.await
.expect("load fenced account")
.expect("fenced account");
assert_eq!(current.availability(), AccountAvailability::QuotaExhausted);
let observed = repository
.get_quotas(std::slice::from_ref(&account_id))
.await
.expect("load fenced quota")
.pop()
.expect("fenced quota");
assert_eq!(
observed.quota.expect("quota document").expose_to_provider()["marker"],
"newer"
);
database.close().await;
}
#[tokio::test]
async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_usage() {
let Some(database) = TestDatabase::create("provider_account_terminal_list").await else {
@@ -130,10 +221,13 @@ async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_u
let mut invalid = account("acct_invalid", "user-invalid");
invalid.email = Some("invalid@example.invalid".to_owned());
invalid.availability = AccountAvailability::Invalid;
let mut disabled = account("acct_disabled", "user-disabled");
disabled.email = Some("disabled@example.invalid".to_owned());
disabled.enabled = false;
let mut quota_exhausted = account("acct_quota_exhausted", "user-quota-exhausted");
quota_exhausted.email = Some("quota-exhausted@example.invalid".to_owned());
quota_exhausted.availability = AccountAvailability::QuotaExhausted;
for account in [alpha, beta, charlie, invalid, quota_exhausted] {
for account in [alpha, beta, charlie, invalid, disabled, quota_exhausted] {
repository
.insert_provider_account(account)
.await
@@ -213,10 +307,17 @@ async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_u
.await
.expect("sort accounts by retained usage");
assert_eq!(usage_page.config_revision.get(), 1);
assert_eq!(usage_page.total, 5);
assert_eq!(usage_page.summary.total, 5);
assert_eq!(usage_page.total, 6);
assert_eq!(usage_page.summary.total, 6);
assert_eq!(usage_page.summary.active, 2);
assert_eq!(usage_page.summary.quota_exhausted, 1);
assert_eq!(usage_page.summary.unavailable, 3);
assert_eq!(
usage_page.summary.total,
usage_page.summary.active
+ usage_page.summary.quota_exhausted
+ usage_page.summary.unavailable
);
assert_eq!(
usage_page
.items
@@ -1104,6 +1205,103 @@ async fn verified_credential_rotation_preserves_quota_exhaustion() {
database.close().await;
}
#[tokio::test]
async fn xai_free_quota_exhaustion_state_is_persisted() {
let Some(database) = TestDatabase::create("provider_account_xai_free_quota").await else {
return;
};
let repository = PgProviderAccountRepository::new(database.pool.clone());
let account_id = ProviderAccountId::new("acct_xai_free_quota").expect("account ID");
let mut imported = account(account_id.as_str(), "user-xai-free-quota");
imported.provider_kind = "xai".to_owned();
repository
.insert_provider_account(imported)
.await
.expect("insert xAI account");
repository
.apply_state_change(AccountStateChange {
account_id: account_id.clone(),
expected_revision: CredentialRevision::new(1).expect("credential revision"),
availability: AccountAvailability::QuotaExhausted,
reason: Some("upstream_free_quota_exhausted".to_owned()),
cooldown_until: None,
observed_at: SystemTime::now(),
})
.await
.expect("persist xAI free quota exhaustion");
let current: (String, String, Option<String>) = sqlx::query_as(
"select provider_kind, availability, availability_reason
from provider_accounts where id = $1",
)
.bind(account_id.as_str())
.fetch_one(&database.pool)
.await
.expect("load persisted xAI quota state");
assert_eq!(
current,
(
"xai".to_owned(),
"quota_exhausted".to_owned(),
Some("upstream_free_quota_exhausted".to_owned()),
),
);
database.close().await;
}
#[tokio::test]
async fn xai_resettable_usage_limit_state_is_persisted() {
let Some(database) = TestDatabase::create("provider_account_xai_usage_limit").await else {
return;
};
let repository = PgProviderAccountRepository::new(database.pool.clone());
let account_id = ProviderAccountId::new("acct_xai_usage_limit").expect("account ID");
let mut imported = account(account_id.as_str(), "user-xai-usage-limit");
imported.provider_kind = "xai".to_owned();
repository
.insert_provider_account(imported)
.await
.expect("insert xAI account");
let observed_at = SystemTime::now();
let cooldown_until = observed_at
.checked_add(Duration::from_secs(24 * 60 * 60))
.expect("cooldown deadline");
repository
.apply_state_change(AccountStateChange {
account_id: account_id.clone(),
expected_revision: CredentialRevision::new(1).expect("credential revision"),
availability: AccountAvailability::Cooldown,
reason: Some("usage_limit_exhausted".to_owned()),
cooldown_until: Some(cooldown_until),
observed_at,
})
.await
.expect("persist xAI resettable usage limit");
let current: (String, String, Option<String>, bool) = sqlx::query_as(
"select provider_kind, availability, availability_reason, cooldown_until is not null
from provider_accounts where id = $1",
)
.bind(account_id.as_str())
.fetch_one(&database.pool)
.await
.expect("load persisted xAI usage limit state");
assert_eq!(
current,
(
"xai".to_owned(),
"cooldown".to_owned(),
Some("usage_limit_exhausted".to_owned()),
true,
),
);
database.close().await;
}
#[tokio::test]
async fn disabled_account_preserves_user_state_during_refresh_writes() {
let Some(database) = TestDatabase::create("provider_account_disabled_refresh").await else {
+83 -7
View File
@@ -25,8 +25,9 @@ use gateway_admin::model::{MutationActor, MutationContext, Revision};
use gateway_admin::ports::provider::{ProviderAdmin, ProviderAdminError, ProviderAdminErrorKind};
use gateway_core::accounting::Money;
use gateway_core::engine::credential::{
CredentialRevision, LoadedCredential, NewProviderAccount, OpaqueProviderData,
PlaintextCredential, ProviderAccount, ProviderAccountId, ProviderAccountStore,
AccountAvailability, CredentialRevision, LoadedCredential, NewProviderAccount,
OpaqueProviderData, PlaintextCredential, ProviderAccount, ProviderAccountId,
ProviderAccountStore,
};
use gateway_core::error::StoreErrorKind;
use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
@@ -458,21 +459,23 @@ impl ProviderAdmin for OpenAiAdminProvider {
refresh,
rolling_usage: _,
} = request;
self.account(&account_id).await?;
let mut account = self.account(&account_id).await?;
let snapshot = if refresh {
Some(
let snapshot = Some(
self.quota
.refresh_account(&account_id)
.await
.map_err(map_quota_error)?,
)
);
account = self.account(&account_id).await?;
snapshot
} else {
self.quota
.read_account(&account_id)
.await
.map_err(map_quota_error)?
};
Ok(snapshot.map_or_else(empty_quota, project_quota))
Ok(project_quota(snapshot, &account))
}
async fn models(
@@ -746,7 +749,18 @@ fn empty_quota() -> ProviderQuota {
}
}
fn project_quota(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
fn project_quota(
snapshot: Option<CodexAccountQuotaSnapshot>,
account: &ProviderAccount,
) -> ProviderQuota {
let mut quota = snapshot.map_or_else(empty_quota, project_quota_snapshot);
if account.availability() == AccountAvailability::QuotaExhausted {
force_confirmed_exhaustion_projection(&mut quota, account.cooldown_until());
}
quota
}
fn project_quota_snapshot(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
let mut provider_data = Map::new();
provider_data.insert(
"remaining_percent".to_owned(),
@@ -799,6 +813,68 @@ fn project_quota(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
}
}
fn force_confirmed_exhaustion_projection(
quota: &mut ProviderQuota,
cooldown_until: Option<SystemTime>,
) {
// 真实 429 对“当前已耗尽”的证明强于可能延迟结算的 usage 文档。这里只覆盖
// Admin 展示投影,原始 Provider JSON 仍作为后续判断窗口是否真正恢复的基线。
quota.provider_data = Some(ProviderDocument::new(OpaqueProviderData::new(
Map::from_iter([
(
"remaining_percent".to_owned(),
Value::Number(Number::from(0)),
),
("exhausted".to_owned(), Value::Bool(true)),
]),
)));
let reset_at = cooldown_until.map(DateTime::<Utc>::from);
if quota.windows.is_empty() {
quota.windows.push(ProviderQuotaWindow {
key: "confirmed-quota-exhaustion".to_owned(),
group: "other".to_owned(),
label: "额度".to_owned(),
source: None,
window_seconds: None,
used_percent: Some(100.0),
reset_at,
local_usage: None,
provider_data: None,
});
return;
}
let matched_reset = reset_at.and_then(|reset_at| {
quota
.windows
.iter()
.enumerate()
.filter_map(|(index, window)| {
let distance = window.reset_at.map(|window_reset| {
(window_reset.timestamp() - reset_at.timestamp()).unsigned_abs()
})?;
Some((index, distance))
})
.min_by_key(|(_, distance)| *distance)
.map(|(index, _)| index)
});
let index = matched_reset
.or_else(|| {
quota
.windows
.iter()
.enumerate()
.max_by(|(_, left), (_, right)| {
left.used_percent
.unwrap_or(-1.0)
.total_cmp(&right.used_percent.unwrap_or(-1.0))
})
.map(|(index, _)| index)
})
.unwrap_or(0);
quota.windows[index].used_percent = Some(100.0);
}
const fn quota_group(kind: CodexQuotaWindowKind) -> &'static str {
match kind {
CodexQuotaWindowKind::Monthly => "monthly",
@@ -217,7 +217,7 @@ pub struct CodexQuotaSettings {
impl Default for CodexQuotaSettings {
fn default() -> Self {
Self {
refresh_interval_minutes: 5,
refresh_interval_minutes: 15,
skip_exhausted: true,
}
}
@@ -16,9 +16,16 @@ pub(crate) fn derive_codex_session_affinity_key(
.map(str::trim)
.filter(|value| !value.is_empty())
{
return ProviderSessionAffinityKey::try_new(conversation_id.to_owned()).ok();
if let Ok(key) = ProviderSessionAffinityKey::try_new(conversation_id.to_owned()) {
return Some(key);
}
return opaque_affinity_key("local-conversation", conversation_id);
}
let (domain, value) = derive_conversation_anchor(request)?;
opaque_affinity_key(domain, &value)
}
fn opaque_affinity_key(domain: &str, value: &str) -> Option<ProviderSessionAffinityKey> {
let mut hasher = Sha256::new();
hasher.update(b"codex-session-affinity-v1\0");
hasher.update(domain.as_bytes());
@@ -30,15 +30,12 @@ const MAX_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(60 * 60);
const QUOTA_SCHEDULING_TTL: Duration = Duration::from_secs(10 * 60);
const QUOTA_HYDRATION_FAILURE_TTL: Duration = Duration::from_secs(5);
const QUOTA_PERIODIC_REFRESH_MIN_INTERVAL: Duration = Duration::from_secs(30 * 60);
const QUOTA_RESET_RECHECK_GRACE: Duration = Duration::from_secs(30);
// ready 账号主要依赖真实请求的被动观察;这里只提供低频、定额的状态兜底。
const QUOTA_READY_REFRESH_MAX_AGE: Duration = Duration::from_secs(2 * 60 * 60);
const QUOTA_READY_REFRESH_BATCH_SIZE: usize = 8;
/// OpenAI Provider 主动额度刷新的调度策略。
///
/// 账号筛选仍由 core 的可调度状态统一处理;该策略只控制 Provider 自己的
/// quota worker 频率,避免把 OpenAI 额度语义泄漏到公共调度层。
/// 正常账号依赖请求响应的被动额度同步;周期 worker 仅复核已耗尽账号。
/// 该策略只控制 Provider 自己的 quota worker 频率,避免把 OpenAI
/// 额度语义泄漏到公共调度层。
#[derive(Debug, Clone, Copy)]
pub struct CodexQuotaRefreshPolicy {
interval: Duration,
@@ -197,6 +194,19 @@ pub struct CodexQuotaSyncSummary {
pub stale: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum QuotaRecoveryEvidence {
SuccessfulResponse,
RefreshedSnapshot { confirms_recovery: bool },
PostFailureSnapshot,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum QuotaRefreshMode {
Reconcile,
AfterConfirmedFailure,
}
impl CodexQuotaSyncSummary {
#[must_use]
pub const fn has_operational_failures(self) -> bool {
@@ -249,8 +259,6 @@ struct CodexQuotaProjectionState {
next_version: u64,
entries: BTreeMap<ProviderAccountId, CodexQuotaSchedulingEntry>,
last_periodic_refresh_at: BTreeMap<ProviderAccountId, Instant>,
last_exhausted_reset_recheck: BTreeMap<ProviderAccountId, CodexQuotaResetMarker>,
ready_refresh_cursor: Option<ProviderAccountId>,
}
#[derive(Debug, Clone, Copy)]
@@ -261,24 +269,6 @@ struct CodexQuotaSchedulingEntry {
signals: Option<AccountQuotaSignals>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct CodexQuotaResetMarker {
revision: CredentialRevision,
reset_at: SystemTime,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CodexQuotaRefreshKind {
Exhausted { reset_at: Option<SystemTime> },
Ready,
}
#[derive(Clone)]
struct CodexQuotaRefreshCandidate {
account: ProviderAccount,
kind: CodexQuotaRefreshKind,
}
#[derive(Clone)]
struct CodexQuotaHydrationTarget {
account: ProviderAccount,
@@ -461,16 +451,15 @@ impl CodexQuotaSchedulingProjection {
fn reserve_periodic_refreshes(
&self,
accounts: Vec<ProviderAccount>,
observations: Option<&BTreeMap<ProviderAccountId, QuotaObservation>>,
now: SystemTime,
) -> Vec<ProviderAccount> {
let candidates = accounts
.into_iter()
.filter_map(|account| quota_refresh_candidate(account, observations, now))
.filter_map(|account| quota_refresh_candidate(account, now))
.collect::<Vec<_>>();
let candidate_ids = candidates
.iter()
.map(|candidate| candidate.account.id().clone())
.map(|account| account.id().clone())
.collect::<BTreeSet<_>>();
let refreshed_at = Instant::now();
let mut state = self
@@ -480,95 +469,27 @@ impl CodexQuotaSchedulingProjection {
state
.last_periodic_refresh_at
.retain(|account_id, _| candidate_ids.contains(account_id));
state
.last_exhausted_reset_recheck
.retain(|account_id, _| candidate_ids.contains(account_id));
// 正常账号只由真实请求的响应头和 `codex.rate_limits` 被动同步。定时器只兜底
// 已耗尽账号,且不依据 reset 时间跳过复核;每个账号仍至少间隔 30 分钟。
let mut reserved = Vec::new();
let mut ready = Vec::new();
for candidate in candidates {
match candidate.kind {
CodexQuotaRefreshKind::Exhausted { reset_at } => {
if reserve_exhausted_quota_refresh(
&mut state,
&candidate.account,
reset_at,
now,
refreshed_at,
) {
reserved.push(candidate.account);
}
}
CodexQuotaRefreshKind::Ready => ready.push(candidate.account),
for account in candidates {
if !periodic_quota_refresh_due(&state, account.id(), refreshed_at) {
continue;
}
}
ready.sort_by(|left, right| left.id().cmp(right.id()));
let ready = ready
.into_iter()
.filter(|account| periodic_quota_refresh_due(&state, account.id(), refreshed_at))
.collect::<Vec<_>>();
let ready_len = ready.len();
if ready_len == 0 {
return reserved;
}
let start = state
.ready_refresh_cursor
.as_ref()
.and_then(|cursor| ready.iter().position(|account| account.id() > cursor))
.unwrap_or(0);
let ready = ready
.into_iter()
.cycle()
.skip(start)
.take(ready_len.min(QUOTA_READY_REFRESH_BATCH_SIZE))
.collect::<Vec<_>>();
if let Some(last) = ready.last() {
state.ready_refresh_cursor = Some(last.id().clone());
}
for account in &ready {
state
.last_periodic_refresh_at
.insert(account.id().clone(), refreshed_at);
reserved.push(account);
}
reserved.extend(ready);
reserved
}
}
fn quota_refresh_candidate(
account: ProviderAccount,
observations: Option<&BTreeMap<ProviderAccountId, QuotaObservation>>,
now: SystemTime,
) -> Option<CodexQuotaRefreshCandidate> {
if !eligible_periodic_quota_refresh(&account, now) {
return None;
}
let observation = observations.and_then(|observations| observations.get(account.id()));
let snapshot = quota_snapshot_for_account(&account, observation);
let kind = match account.availability() {
AccountAvailability::QuotaExhausted => CodexQuotaRefreshKind::Exhausted {
reset_at: account
.cooldown_until()
.or_else(|| quota_reset_at_from_snapshot(snapshot.as_ref())),
},
AccountAvailability::Cooldown
if snapshot
.as_ref()
.is_some_and(|snapshot| snapshot.fact().exhausted()) =>
{
CodexQuotaRefreshKind::Exhausted {
reset_at: quota_reset_at_from_snapshot(snapshot.as_ref()),
}
}
AccountAvailability::Ready
if observations.is_some() && quota_observation_is_stale(&account, observation, now) =>
{
CodexQuotaRefreshKind::Ready
}
_ => return None,
};
Some(CodexQuotaRefreshCandidate { account, kind })
fn quota_refresh_candidate(account: ProviderAccount, now: SystemTime) -> Option<ProviderAccount> {
(eligible_periodic_quota_refresh(&account, now)
&& account.availability() == AccountAvailability::QuotaExhausted)
.then_some(account)
}
fn quota_snapshot_for_account(
@@ -597,72 +518,6 @@ fn quota_observation_confirms_active_exhaustion(
.is_some_and(|snapshot| quota_fact_is_active_exhaustion(snapshot.fact(), now))
}
fn quota_observation_is_stale(
account: &ProviderAccount,
observation: Option<&QuotaObservation>,
now: SystemTime,
) -> bool {
let Some(observation) = observation.filter(|observation| {
observation.expected_revision == account.revision()
&& quota_snapshot_from_observation(observation).is_some()
}) else {
return true;
};
observation
.observed_at
.and_then(|observed_at| now.duration_since(observed_at).ok())
.is_none_or(|age| age >= QUOTA_READY_REFRESH_MAX_AGE)
}
fn reserve_exhausted_quota_refresh(
state: &mut CodexQuotaProjectionState,
account: &ProviderAccount,
reset_at: Option<SystemTime>,
now: SystemTime,
refreshed_at: Instant,
) -> bool {
if let Some(reset_at) = reset_at {
if !quota_reset_recheck_ready(reset_at, now) {
return false;
}
let same_reset_already_checked = state
.last_exhausted_reset_recheck
.get(account.id())
.is_some_and(|marker| {
marker.revision == account.revision() && marker.reset_at == reset_at
});
if same_reset_already_checked
&& !periodic_quota_refresh_due(state, account.id(), refreshed_at)
{
return false;
}
state
.last_periodic_refresh_at
.insert(account.id().clone(), refreshed_at);
state.last_exhausted_reset_recheck.insert(
account.id().clone(),
CodexQuotaResetMarker {
revision: account.revision(),
reset_at,
},
);
return true;
}
if !periodic_quota_refresh_due(state, account.id(), refreshed_at) {
return false;
}
state
.last_periodic_refresh_at
.insert(account.id().clone(), refreshed_at);
true
}
fn quota_reset_recheck_ready(reset_at: SystemTime, now: SystemTime) -> bool {
reset_at
.checked_add(QUOTA_RESET_RECHECK_GRACE)
.is_some_and(|ready_at| ready_at <= now)
}
fn periodic_quota_refresh_due(
state: &CodexQuotaProjectionState,
account_id: &ProviderAccountId,
@@ -774,16 +629,14 @@ impl CodexCredentialQuotaService {
.collect::<BTreeMap<_, _>>(),
),
Err(error) => {
// 持久观察不可用时不盲扫 ready;保留已耗尽账号的既有恢复轮询。
tracing::warn!(error = %error, "OpenAI quota refresh skipped ready-account scan");
// 持久观察不可用时跳过旧状态修正;仍复核当前已标记耗尽的账号。
tracing::warn!(error = %error, "OpenAI quota refresh skipped persisted exhaustion reconciliation");
None
}
};
self.reconcile_persisted_exhaustion(&mut accounts, observations.as_ref(), now)
.await;
let accounts =
self.scheduling
.reserve_periodic_refreshes(accounts, observations.as_ref(), now);
let accounts = self.scheduling.reserve_periodic_refreshes(accounts, now);
if accounts.is_empty() {
return Ok(summary);
}
@@ -796,9 +649,20 @@ impl CodexCredentialQuotaService {
let observed_at = SystemTime::now();
match self.fetch_usage_with_recovery(&client, &account).await {
Ok(FetchedCodexQuota { account, value }) => {
let previous_snapshot = observations
.as_ref()
.and_then(|observations| observations.get(account.id()))
.filter(|observation| observation.expected_revision == account.revision())
.and_then(quota_snapshot_from_observation);
// 单账号解析或落库失败只影响该账号;其余账号继续同步。
if let Err(error) = self
.apply_fetched_quota(&account, &value, observed_at, &mut summary)
.apply_fetched_quota(
&account,
&value,
observed_at,
previous_snapshot.as_ref(),
&mut summary,
)
.await
{
summary.transient += 1;
@@ -913,9 +777,16 @@ impl CodexCredentialQuotaService {
account: &ProviderAccount,
value: &Value,
observed_at: SystemTime,
previous_snapshot: Option<&CodexAccountQuotaSnapshot>,
summary: &mut CodexQuotaSyncSummary,
) -> Result<(), CodexCredentialQuotaError> {
let fact = parse_codex_quota_usage(value)?;
let snapshot = parse_account_quota_snapshot(
account.id().clone(),
account.revision(),
observed_at,
value,
)?;
let fact = snapshot.fact();
let object = value
.as_object()
.cloned()
@@ -953,6 +824,9 @@ impl CodexCredentialQuotaService {
current.cooldown_until(),
fact,
SystemTime::now(),
QuotaRecoveryEvidence::RefreshedSnapshot {
confirms_recovery: quota_snapshot_confirms_recovery(previous_snapshot, &snapshot),
},
) {
let _ = self
.repository
@@ -1019,6 +893,7 @@ impl CodexCredentialQuotaService {
current.cooldown_until(),
fact,
observed_at,
QuotaRecoveryEvidence::SuccessfulResponse,
) {
self.repository
.apply_state(
@@ -1094,6 +969,24 @@ impl CodexCredentialQuotaService {
pub async fn refresh_account(
&self,
account_id: &ProviderAccountId,
) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
self.refresh_account_with_mode(account_id, QuotaRefreshMode::Reconcile)
.await
}
/// 真实限额失败后的异步刷新只补齐展示快照,不允许同一轮陈旧 usage 撤销失败状态。
pub(crate) async fn refresh_account_after_failure(
&self,
account_id: &ProviderAccountId,
) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
self.refresh_account_with_mode(account_id, QuotaRefreshMode::AfterConfirmedFailure)
.await
}
async fn refresh_account_with_mode(
&self,
account_id: &ProviderAccountId,
mode: QuotaRefreshMode,
) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
let account = self
.store
@@ -1101,6 +994,7 @@ impl CodexCredentialQuotaService {
.await?
.filter(|account| account.provider().as_str() == "openai")
.ok_or(CodexCredentialQuotaError::NotFound)?;
let previous_snapshot = self.read_snapshot_for(&account).await?;
let observed_at = SystemTime::now();
if !access_token_is_current(&account, observed_at) {
return Err(CodexCredentialQuotaError::CredentialRefreshRequired);
@@ -1177,6 +1071,17 @@ impl CodexCredentialQuotaService {
current.cooldown_until(),
snapshot.fact(),
SystemTime::now(),
match mode {
QuotaRefreshMode::AfterConfirmedFailure => {
QuotaRecoveryEvidence::PostFailureSnapshot
}
QuotaRefreshMode::Reconcile => QuotaRecoveryEvidence::RefreshedSnapshot {
confirms_recovery: quota_snapshot_confirms_recovery(
previous_snapshot.as_ref(),
&snapshot,
),
},
},
) {
self.repository
.apply_state(
@@ -1399,9 +1304,13 @@ fn quota_success_state(
cooldown_until: Option<SystemTime>,
fact: CodexQuotaFact,
now: SystemTime,
recovery_evidence: QuotaRecoveryEvidence,
) -> Option<(AccountAvailability, Option<SystemTime>)> {
if fact.exhausted() {
let observed_cooldown = fact.resets_at().map(SystemTime::from).or(cooldown_until);
let observed_cooldown = match (fact.resets_at().map(SystemTime::from), cooldown_until) {
(Some(observed), Some(confirmed)) => Some(observed.max(confirmed)),
(observed, confirmed) => observed.or(confirmed),
};
return (current != AccountAvailability::QuotaExhausted
|| cooldown_until != observed_cooldown)
.then_some((AccountAvailability::QuotaExhausted, observed_cooldown));
@@ -1411,8 +1320,13 @@ fn quota_success_state(
AccountAvailability::Invalid
| AccountAvailability::Expired
| AccountAvailability::Banned
| AccountAvailability::Unknown
| AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
| AccountAvailability::Unknown => Some(AccountAvailability::Ready),
AccountAvailability::QuotaExhausted
if !quota_recovery_confirmed(recovery_evidence, cooldown_until, now) =>
{
None
}
AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
AccountAvailability::Ready => None,
AccountAvailability::Cooldown
if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) =>
@@ -1424,6 +1338,59 @@ fn quota_success_state(
availability.map(|availability| (availability, None))
}
fn quota_recovery_confirmed(
evidence: QuotaRecoveryEvidence,
cooldown_until: Option<SystemTime>,
now: SystemTime,
) -> bool {
if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) {
return true;
}
match evidence {
QuotaRecoveryEvidence::SuccessfulResponse => true,
QuotaRecoveryEvidence::RefreshedSnapshot { confirms_recovery } => confirms_recovery,
QuotaRecoveryEvidence::PostFailureSnapshot => false,
}
}
fn quota_snapshot_confirms_recovery(
previous: Option<&CodexAccountQuotaSnapshot>,
current: &CodexAccountQuotaSnapshot,
) -> bool {
if current.fact().exhausted() {
return false;
}
let Some(previous) = previous else {
return false;
};
if previous
.fact()
.remaining_percent()
.zip(current.fact().remaining_percent())
.is_some_and(|(previous, current)| current > previous)
{
return true;
}
// 单窗口账号可以用同一窗口的用量回落或 reset 前移证明上游提前翻新;多窗口
// 无法从账号级 429 确认具体受限窗口,保守等待整体剩余额度改善或真实请求成功。
let ([previous], [current]) = (previous.windows(), current.windows()) else {
return false;
};
if previous.key() != current.key() {
return false;
}
let usage_recovered = previous
.used_percent()
.zip(current.used_percent())
.is_some_and(|(previous, current)| current < previous);
let window_advanced = previous
.reset_at()
.zip(current.reset_at())
.is_some_and(|(previous, current)| current > previous);
usage_recovered || window_advanced
}
fn quota_state_transition(
error: &CodexClientError,
observed_at: SystemTime,
+90 -42
View File
@@ -84,6 +84,12 @@ const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024;
/// 提交边界前最多保留 64 KiB 原始上游 chunk;达到阈值后结束无感换号窗口,
/// 但不会把上游数据改写成协议失败。
const MAX_STREAM_PREFETCH_BYTES: usize = 64 * 1024;
/// 短暂保留 response.created 等结构事件,让随后到达的明确拒绝可以无感换号;
/// 到期即放行,避免模型长时间思考时让客户端一直收不到首事件。
const STREAM_REPLAY_GRACE: Duration = Duration::from_millis(1_200);
// 额度拒绝后先给上游额度结算留出时间,再以受限时长同步 usage 快照。
const QUOTA_FAILURE_REFRESH_DELAY: Duration = Duration::from_secs(2);
const QUOTA_FAILURE_REFRESH_TIMEOUT: Duration = Duration::from_secs(5);
pub const OFFICIAL_CODEX_BASE_PATH: &str = "/backend-api";
pub const OFFICIAL_CODEX_BASE_URL: &str = "https://chatgpt.com/backend-api";
@@ -679,8 +685,6 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
lease.escape_reason(),
lease.account_switch(),
);
let stream_commit_policy =
StreamCommitPolicy::for_continuation(context.continuation_attempt());
let mut response = create_response_attempt(
&client,
&request,
@@ -826,12 +830,13 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
let mut decoder = CodexCanonicalDecoder::new(upstream_model.as_str())
.with_requested_service_tier(request.service_tier())
.with_raw_sse_passthrough();
let mut pre_commit_events = PreCommitClientEvents::new(stream_commit_policy);
let mut pre_commit_events = PreCommitClientEvents::new();
loop {
let Some(stream_deadline) = remaining(context.deadline()) else {
Err(provider_error(ProviderErrorKind::Timeout, UpstreamSendState::Sent))?;
return;
};
let replay_grace_deadline = pre_commit_events.replay_grace_deadline();
let next = tokio::select! {
biased;
_ = cancellation.cancelled() => Err(MappedProviderFailure::plain(provider_error(
@@ -842,14 +847,21 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
ProviderErrorKind::Timeout,
UpstreamSendState::Sent,
))),
_ = wait_for_replay_grace(replay_grace_deadline) => Ok(PreCommitPoll::GraceElapsed),
chunk = body.next() => match chunk {
Some(Ok(chunk)) => Ok(Some(chunk)),
Some(Ok(chunk)) => Ok(PreCommitPoll::Upstream(Some(chunk))),
Some(Err(error)) => Err(map_stream_error(error)),
None => Ok(None),
None => Ok(PreCommitPoll::Upstream(None)),
},
};
let next = match next {
Ok(next) => next,
Ok(PreCommitPoll::Upstream(next)) => next,
Ok(PreCommitPoll::GraceElapsed) => {
for event in pre_commit_events.commit_pending() {
yield event;
}
continue;
}
Err(failure) => {
let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
if !updates.is_empty() {
@@ -1798,54 +1810,37 @@ impl MappedProviderFailure {
}
}
/// 与 2.x 一致的流提交策略:普通请求优先首事件,续写/重放优先可恢复性。
#[derive(Clone, Copy)]
enum StreamCommitPolicy {
FirstForwardableEvent,
UntilOutputOrTerminal,
enum PreCommitPoll<T> {
Upstream(T),
GraceElapsed,
}
impl StreamCommitPolicy {
const fn for_continuation(attempt: ContinuationAttempt) -> Self {
match attempt {
ContinuationAttempt::None => Self::FirstForwardableEvent,
ContinuationAttempt::Native
| ContinuationAttempt::ReplayOwner
| ContinuationAttempt::ReplayAny => Self::UntilOutputOrTerminal,
}
}
fn boundary_reached(
self,
events: &[ProviderEvent],
timing_signals: ResponseEventSignals,
completed: bool,
) -> bool {
match self {
Self::FirstForwardableEvent => events.iter().any(ProviderEvent::has_client_event),
Self::UntilOutputOrTerminal => timing_signals.semantic_output || completed,
}
async fn wait_for_replay_grace(deadline: Option<Instant>) {
if let Some(deadline) = deadline {
tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
} else {
std::future::pending::<()>().await;
}
}
/// 提交边界前的上游事件预取。
///
/// 原始 chunk 计数而不是重编码后的 event 大小。64 KiB 只限定无感换号窗口;达到
/// 阈值会提交已缓存 wire,绝不能因为网关私有资源规则伪造上游协议失败。一旦提交,
/// 后续事件不再具备无痕重放资格。
/// 原始 chunk 计数而不是重编码后的 event 大小。时间与 64 KiB 共同限定无感换号
/// 窗口;任一边界到达都会提交已缓存 wire,不能因网关私有资源规则伪造上游协议
/// 失败。一旦提交,后续事件不再具备无痕重放资格。
struct PreCommitClientEvents {
policy: StreamCommitPolicy,
pending: Vec<ProviderEvent>,
prefetched_bytes: usize,
replay_grace_deadline: Option<Instant>,
committed: bool,
}
impl PreCommitClientEvents {
const fn new(policy: StreamCommitPolicy) -> Self {
const fn new() -> Self {
Self {
policy,
pending: Vec::new(),
prefetched_bytes: 0,
replay_grace_deadline: None,
committed: false,
}
}
@@ -1865,13 +1860,17 @@ impl PreCommitClientEvents {
if self.committed {
return incoming;
}
let boundary_reached = self
.policy
.boundary_reached(&incoming, timing_signals, completed);
let starts_replay_grace = incoming.iter().any(ProviderEvent::has_client_event);
self.pending.extend(incoming);
if boundary_reached || self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES {
if timing_signals.semantic_output
|| completed
|| self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES
{
return self.commit_pending();
}
if starts_replay_grace && self.replay_grace_deadline.is_none() {
self.replay_grace_deadline = Instant::now().checked_add(STREAM_REPLAY_GRACE);
}
Vec::new()
}
@@ -1900,9 +1899,14 @@ impl PreCommitClientEvents {
fn take_for_failure(&mut self, incoming: Vec<ProviderEvent>) -> Vec<ProviderEvent> {
self.pending.extend(incoming);
self.prefetched_bytes = 0;
self.replay_grace_deadline = None;
std::mem::take(&mut self.pending)
}
const fn replay_grace_deadline(&self) -> Option<Instant> {
self.replay_grace_deadline
}
const fn is_committed(&self) -> bool {
self.committed
}
@@ -1910,6 +1914,7 @@ impl PreCommitClientEvents {
fn commit_pending(&mut self) -> Vec<ProviderEvent> {
self.committed = true;
self.prefetched_bytes = 0;
self.replay_grace_deadline = None;
std::mem::take(&mut self.pending)
}
}
@@ -1937,7 +1942,7 @@ impl ReplayBoundary {
struct OpenAiFailureContext<'a> {
client: &'a CodexBackendClient,
selector: &'a CodexCredentialSelector,
quota: &'a CodexCredentialQuotaService,
quota: &'a Arc<CodexCredentialQuotaService>,
response_origin: &'a Url,
cyber_policy_scope: Option<&'a CodexCyberPolicyScope>,
allows_account_state_mutation: bool,
@@ -1952,6 +1957,10 @@ async fn apply_failure(
return;
}
synchronize_passive_quota(context.quota, account, &failure.rate_limit_headers).await;
let needs_authoritative_quota_refresh = matches!(
failure.account_failure,
Some(CodexAccountFailure::QuotaExhausted | CodexAccountFailure::UsageLimitExhausted { .. })
);
if failure.cyber_policy_failure {
context
.selector
@@ -1975,6 +1984,9 @@ async fn apply_failure(
);
}
}
if needs_authoritative_quota_refresh {
schedule_authoritative_quota_refresh_after_failure(context.quota, account);
}
if failure.capture_response_cookies
&& !failure.set_cookie_headers.is_empty()
&& let Err(error) = context
@@ -1994,6 +2006,42 @@ async fn apply_failure(
}
}
fn schedule_authoritative_quota_refresh_after_failure(
quota: &Arc<CodexCredentialQuotaService>,
account: &ProviderAccount,
) {
// 限额错误可以确认账号状态,但 Responses 事件中的 used_percent 可能仍停在上一结算点。
// usage 快照在后台补齐展示基线,不得撤销同一轮真实失败,也不能阻塞原始响应。
// 先等待 2 秒让上游结算,再查询;5 秒仅限制实际查询自身。
let quota = Arc::clone(quota);
let account_id = account.id().clone();
drop(tokio::spawn(async move {
tokio::time::sleep(QUOTA_FAILURE_REFRESH_DELAY).await;
match tokio::time::timeout(
QUOTA_FAILURE_REFRESH_TIMEOUT,
quota.refresh_account_after_failure(&account_id),
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(error)) => {
tracing::warn!(
account_id = %account_id,
error = %error,
"OpenAI authoritative quota synchronization after quota rejection failed"
);
}
Err(_) => {
tracing::warn!(
account_id = %account_id,
timeout_ms = QUOTA_FAILURE_REFRESH_TIMEOUT.as_millis(),
"OpenAI authoritative quota synchronization after quota rejection timed out"
);
}
}
}));
}
fn map_handshake_error(error: CodexClientError) -> MappedProviderFailure {
map_client_error(error, UpstreamSendState::Ambiguous, true)
}
@@ -161,13 +161,14 @@ pub(crate) fn derive_conversation_anchor(
request: &CodexResponsesRequest,
) -> Option<(&'static str, String)> {
request
.prompt_cache_key()
.map(|value| ("prompt-cache", value.to_owned()))
.client_session_id
.as_deref()
.map(|value| ("session", value.to_owned()))
.or_else(|| {
request
.client_session_id
.client_conversation_id
.as_deref()
.map(|value| ("session", value.to_owned()))
.map(|value| ("conversation", value.to_owned()))
})
.or_else(|| {
request
@@ -177,9 +178,8 @@ pub(crate) fn derive_conversation_anchor(
})
.or_else(|| {
request
.client_conversation_id
.as_deref()
.map(|value| ("conversation", value.to_owned()))
.prompt_cache_key()
.map(|value| ("prompt-cache", value.to_owned()))
})
.or_else(|| derive_stable_conversation_key(request).map(|value| ("request", value)))
}
+104 -2
View File
@@ -20,8 +20,8 @@ use gateway_admin::model::provider_credentials::{
use gateway_admin::model::{MutationActor, MutationContext, Revision};
use gateway_admin::ports::provider::ProviderAdminErrorKind;
use gateway_core::engine::credential::{
CredentialRevision, OpaqueProviderData, ProviderAccount, ProviderAccountId,
ProviderAccountStore, QuotaObservation,
AccountAvailability as CoreAccountAvailability, AccountStateChange, CredentialRevision,
OpaqueProviderData, ProviderAccount, ProviderAccountId, ProviderAccountStore, QuotaObservation,
};
use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
use gateway_core::provider_ports::{
@@ -503,6 +503,108 @@ async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_qu
assert_eq!(monthly[0].used_percent, Some(2.0));
}
#[tokio::test]
async fn openai_admin_projects_confirmed_quota_exhaustion_as_full_without_mutating_raw_usage() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_admin_confirmed_exhaustion";
store
.seed_oauth_credential(ImportCodexOAuthCredential {
account_id: account_id.to_owned(),
name: "admin confirmed exhaustion".to_owned(),
secret: secret("admin-confirmed-exhaustion-access"),
verified_account: profile("chatgpt-admin-confirmed-exhaustion"),
next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)),
enabled: true,
})
.await;
let account = store.account(account_id).expect("stored account");
let reset_at = 1_900_000_000_u64;
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 86, "reset_at": reset_at}
}
})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
})
.await
.expect("persist raw quota");
store
.apply_state_change(AccountStateChange {
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: CoreAccountAvailability::QuotaExhausted,
reason: Some("usage_limit_exhausted".to_owned()),
cooldown_until: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset_at)),
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
let config = valid_config();
let bundle = provider_openai::initialize(
config.config.clone(),
provider_ports_with(Arc::clone(&store), Arc::new(TestOAuthPending::default())),
)
.await
.expect("OpenAI bundle");
let exhausted = bundle
.admin_provider()
.quota(ProviderQuotaRequest {
account_id: account.id().clone(),
refresh: false,
rolling_usage: None,
})
.await
.expect("project exhausted quota");
assert_eq!(exhausted.windows.len(), 1);
assert_eq!(exhausted.windows[0].used_percent, Some(100.0));
let raw = store
.get_quotas(std::slice::from_ref(account.id()))
.await
.expect("read raw quota")
.pop()
.expect("raw quota");
assert_eq!(
raw.quota.expect("raw document").expose_to_provider()["rate_limit"]["primary_window"]["used_percent"],
86
);
store
.apply_state_change(AccountStateChange {
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: CoreAccountAvailability::Ready,
reason: None,
cooldown_until: None,
observed_at: SystemTime::now(),
})
.await
.expect("recover account");
let recovered = bundle
.admin_provider()
.quota(ProviderQuotaRequest {
account_id: account.id().clone(),
refresh: false,
rolling_usage: None,
})
.await
.expect("project recovered quota");
assert_eq!(recovered.windows[0].used_percent, Some(86.0));
}
#[tokio::test]
async fn openai_admin_provider_rejects_unprepared_mutations_before_store_commit() {
let store = Arc::new(MemoryAccountStore::default());
@@ -117,7 +117,7 @@ fn openai_config_defaults_to_the_provider_owned_operating_values() {
64,
8,
20_000,
5,
15,
true,
true,
"app_EMoamEEZ73f0CkXaXp7hrann",
@@ -498,6 +498,38 @@ async fn manual_quota_refresh_updates_account_state_from_provider_response() {
AccountAvailability::QuotaExhausted
);
server.reset().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.and(header(
"authorization",
format!("Bearer token-{account_id}"),
))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
}
})))
.mount(&server)
.await;
let stale = service
.refresh_account(account.id())
.await
.expect("refresh contradictory full quota");
assert!(!stale.fact().exhausted());
assert_eq!(
store
.account(account_id)
.expect("account after contradictory quota")
.availability(),
AccountAvailability::QuotaExhausted,
"an unchanged usage window cannot override a confirmed quota failure"
);
server.reset().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
@@ -755,7 +787,7 @@ async fn synchronize_without_accounts_is_a_noop_before_network_io() {
}
#[tokio::test]
async fn periodic_quota_synchronization_skips_fresh_ready_accounts() {
async fn periodic_quota_synchronization_does_not_scan_stale_ready_accounts() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_periodic_ready").await;
let account = store
@@ -776,15 +808,16 @@ async fn periodic_quota_synchronization_skips_fresh_ready_accounts() {
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
// 正常账号即使缓存已旧,也只等待真实请求被动同步,不能被定时任务主动拉取。
observed_at: Some(SystemTime::UNIX_EPOCH),
})
.await
.expect("persist fresh quota");
.expect("persist stale quota");
let summary = blocked_network_quota_service(&store)
.synchronize()
.await
.expect("fresh ready account must not query upstream");
.expect("ready account must not query upstream");
assert_eq!(summary, CodexQuotaSyncSummary::default());
}
@@ -830,7 +863,13 @@ async fn periodic_quota_synchronization_repairs_cooldown_with_active_exhausted_q
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(429))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": false,
"limit_reached": true,
"primary_window": {"used_percent": 100, "reset_at": reset_at}
}
})))
.mount(&server)
.await;
let service = quota_service_with_base_url(
@@ -939,72 +978,6 @@ async fn periodic_quota_synchronization_skips_ordinary_cooldown_without_exhauste
);
}
#[tokio::test]
async fn periodic_quota_synchronization_limits_stale_ready_account_scans() {
let store = Arc::new(MemoryAccountStore::default());
for index in 0..9 {
let account_id = format!("acct_periodic_stale_ready_{index}");
create_account(&store, &account_id).await;
let account = store.account(&account_id).expect("created account");
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": true,
"primary_window": {"used_percent": 10, "reset_at": 1_900_000_000}
}
})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::UNIX_EPOCH),
})
.await
.expect("persist stale quota");
}
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": false,
"primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
}
})))
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let summary = service.synchronize().await.expect("quota synchronization");
assert_eq!(summary.exhausted, 8);
assert_eq!(
store
.account("acct_periodic_stale_ready_0")
.expect("scanned account")
.availability(),
AccountAvailability::QuotaExhausted
);
assert_eq!(
store
.account("acct_periodic_stale_ready_8")
.expect("deferred account")
.availability(),
AccountAvailability::Ready
);
}
#[tokio::test]
async fn periodic_quota_synchronization_attempts_quota_exhausted_accounts() {
let store = Arc::new(MemoryAccountStore::default());
@@ -1033,7 +1006,7 @@ async fn periodic_quota_synchronization_attempts_quota_exhausted_accounts() {
}
#[tokio::test]
async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_grace() {
async fn periodic_quota_synchronization_rechecks_exhausted_accounts_without_waiting_for_reset() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_periodic_exhausted_cooldown").await;
let account = store
@@ -1045,38 +1018,29 @@ async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_g
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
reason: Some("usage_limit_exhausted".to_owned()),
cooldown_until: SystemTime::now().checked_sub(Duration::from_secs(29)),
cooldown_until: SystemTime::now().checked_add(Duration::from_secs(24 * 60 * 60)),
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted with recent cooldown");
let service = blocked_network_quota_service(&store);
.expect("mark account exhausted with a future reset");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(500))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let waiting = service
.synchronize()
.await
.expect("quota synchronization before grace");
assert_eq!(waiting, CodexQuotaSyncSummary::default());
let current = store
.account("acct_periodic_exhausted_cooldown")
.expect("current account");
store
.apply_state_change(AccountStateChange {
account_id: current.id().clone(),
expected_revision: current.revision(),
availability: AccountAvailability::QuotaExhausted,
reason: Some("usage_limit_exhausted".to_owned()),
cooldown_until: SystemTime::now().checked_sub(Duration::from_secs(31)),
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted past grace");
let due = service
.synchronize()
.await
.expect("quota synchronization after grace");
// reset 仅用于展示/调度状态,不能推迟额度耗尽账号的定时复核。
let due = service.synchronize().await.expect("quota synchronization");
let repeated = service
.synchronize()
.await
@@ -1086,6 +1050,118 @@ async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_g
assert_eq!(repeated, CodexQuotaSyncSummary::default());
}
#[tokio::test]
async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_recover() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_periodic_recovery_evidence";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
let reset_at = 1_900_000_000;
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 86, "reset_at": reset_at}
}
})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
})
.await
.expect("seed post-failure quota snapshot");
store
.apply_state_change(AccountStateChange {
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
reason: Some("usage_limit_exhausted".to_owned()),
cooldown_until: SystemTime::now().checked_add(Duration::from_secs(24 * 60 * 60)),
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 86, "reset_at": reset_at}
}
})))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let unchanged = service
.synchronize()
.await
.expect("unchanged quota recheck");
assert_eq!(unchanged.updated, 1);
assert_eq!(
store
.account(account_id)
.expect("account after unchanged snapshot")
.availability(),
AccountAvailability::QuotaExhausted
);
server.reset().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 20, "reset_at": reset_at}
}
})))
.expect(1)
.mount(&server)
.await;
// 新 worker 实例模拟下一次 30 分钟复核,避免测试等待真实节流周期。
let next_cycle = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let improved = next_cycle
.synchronize()
.await
.expect("improved quota recheck");
assert_eq!(improved.updated, 1);
assert_eq!(
store
.account(account_id)
.expect("account after improved snapshot")
.availability(),
AccountAvailability::Ready
);
}
#[tokio::test]
async fn periodic_quota_synchronization_throttles_the_same_exhausted_account() {
let store = Arc::new(MemoryAccountStore::default());
@@ -1113,91 +1189,7 @@ async fn periodic_quota_synchronization_throttles_the_same_exhausted_account() {
}
#[tokio::test]
async fn periodic_quota_synchronization_rechecks_an_exhausted_account_once_when_its_reset_is_due() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_periodic_reset_due").await;
let account = store
.account("acct_periodic_reset_due")
.expect("created account");
store
.apply_state_change(AccountStateChange {
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
reason: Some("quota_exhausted".to_owned()),
cooldown_until: None,
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": false,
"primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
}
})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
})
.await
.expect("persist future reset");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(500))
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let waiting = service.synchronize().await.expect("initial cooldown wait");
assert_eq!(waiting, CodexQuotaSyncSummary::default());
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": false,
"primary_window": {"used_percent": 100, "reset_at": 1}
}
})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
})
.await
.expect("persist due reset");
let reset_recheck = service.synchronize().await.expect("reset recheck");
let repeated = service
.synchronize()
.await
.expect("throttled reset recheck");
assert_eq!(reset_recheck.transient, 1);
assert_eq!(repeated, CodexQuotaSyncSummary::default());
}
#[tokio::test]
async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
async fn periodic_quota_synchronization_does_not_bypass_throttle_for_a_new_reset() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_periodic_new_reset";
create_account(&store, account_id).await;
@@ -1241,7 +1233,10 @@ async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
.account(account_id)
.expect("account after first recheck")
.cooldown_until();
let second = service.synchronize().await.expect("new reset recheck");
let second = service
.synchronize()
.await
.expect("throttled reset recheck");
let repeated = service.synchronize().await.expect("same reset throttled");
let request_count = server
.received_requests()
@@ -1260,9 +1255,9 @@ async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
(
1,
Some(confirmed_reset),
1,
0,
CodexQuotaSyncSummary::default(),
2,
1,
)
);
}
@@ -7,7 +7,8 @@ use base64::{Engine as _, engine::general_purpose::STANDARD};
use chrono::Utc;
use futures::StreamExt;
use gateway_core::engine::credential::{
AccountAvailability, AccountFeedbackStats, ProviderAccountId, ProviderAccountStore as _,
AccountAvailability, AccountFeedbackStats, OpaqueProviderData, ProviderAccountId,
ProviderAccountStore as _, QuotaObservation,
};
use gateway_core::engine::provider::{Provider as _, ProviderRequest};
use gateway_core::engine::{
@@ -784,6 +785,94 @@ async fn prompt_cache_key_should_become_an_opaque_session_affinity_lookup_key()
assert!(keys[0].bytes().all(|byte| byte.is_ascii_hexdigit()));
}
#[tokio::test]
async fn explicit_session_id_should_override_turn_specific_prompt_cache_keys_for_affinity() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_session_affinity").await;
let affinity = Arc::new(MemorySessionAffinity::default());
let provider = provider_with_affinity(&store, Arc::clone(&affinity));
for (request_id, prompt_cache_key) in [
("req_session_affinity_first", "turn-cache-first"),
("req_session_affinity_second", "turn-cache-second"),
] {
let generation = GenerateRequest::from_protocol_payload(
ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("gpt-5.4")),
("input".to_owned(), json!("hello")),
("prompt_cache_key".to_owned(), json!(prompt_cache_key)),
]),
)
.expect("OpenAI payload")
.with_context(Map::from_iter([(
"session_id".to_owned(),
json!("stable-client-session"),
)])),
);
let stream = provider
.execute(
planned_request("openai", Operation::Generate(generation)),
context(request_id, CancellationToken::new()),
)
.await
.expect("prepare provider stream");
drop(stream);
}
let keys = affinity.lookup_keys();
assert_eq!(keys.len(), 2);
assert_eq!(keys[0], keys[1]);
}
#[tokio::test]
async fn invalid_local_conversation_id_should_fall_back_to_an_opaque_affinity_key() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_local_affinity").await;
let affinity = Arc::new(MemorySessionAffinity::default());
let generation = GenerateRequest::from_protocol_payload(
ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("gpt-5.4")),
("input".to_owned(), json!("hello")),
]),
)
.expect("OpenAI payload"),
)
.with_provider_session_state(
ProviderSessionState::new(
"openai",
Map::from_iter([
("account_id".to_owned(), json!("acct_local_affinity")),
(
"conversation_id".to_owned(),
json!("lc_UppercaseBase64ConversationId"),
),
("continuation_scope".to_owned(), json!("replay_required")),
("transcript".to_owned(), json!([])),
]),
)
.expect("provider session state"),
);
let stream = provider_with_affinity(&store, Arc::clone(&affinity))
.execute(
planned_request("openai", Operation::Generate(generation)),
context("req_local_affinity", CancellationToken::new()),
)
.await
.expect("prepare provider stream");
drop(stream);
let keys = affinity.lookup_keys();
assert_eq!(keys.len(), 1);
assert_eq!(keys[0].len(), 64);
assert!(keys[0].bytes().all(|byte| byte.is_ascii_hexdigit()));
}
#[tokio::test]
async fn completed_response_persists_session_affinity_before_stream_consumer_stops_polling() {
let store = Arc::new(MemoryAccountStore::default());
@@ -918,16 +1007,50 @@ async fn response_failed_before_semantic_output_is_atomic_and_persists_quota_loc
}
#[tokio::test]
async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_quota_snapshot() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_usage_limit_request_path";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 99, "reset_at": 1_900_000_000}
}
})
.as_object()
.expect("stale quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
})
.await
.expect("seed stale passive quota");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.and(header("authorization", format!("Bearer at-{account_id}")))
.respond_with(ResponseTemplate::new(500))
.expect(0)
// 生产环境会短暂返回互相矛盾的快照:数字已到 100%,但布尔位仍声称可用。
// 后台同步应覆盖展示数据,不能据此撤销真实请求已经确认的额度耗尽。
.respond_with(
ResponseTemplate::new(200)
// 验证后台 usage 同步不能把原始的额度错误响应拖到查询完成之后。
.set_delay(Duration::from_millis(750))
.set_body_json(json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
}
})),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("POST"))
@@ -952,13 +1075,17 @@ async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
)
.await
.expect("prepare provider stream");
let failure = loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("usage-limit failure must surface a typed failure"),
let failure = timeout(Duration::from_millis(300), async {
loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("usage-limit failure must surface a typed failure"),
}
}
};
})
.await
.expect("usage refresh must not block the original quota failure");
assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted);
let account = store.account(account_id).expect("usage-limit account");
@@ -968,22 +1095,69 @@ async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
.cooldown_until()
.is_some_and(|until| until > SystemTime::now())
);
assert!(!store.has_quota(account_id));
tokio::time::sleep(Duration::from_millis(200)).await;
let requests = server.received_requests().await.expect("received requests");
assert_eq!(
requests
.iter()
.filter(|request| request.url.path() == "/api/codex/usage")
.count(),
0
0,
"usage refresh must wait for the upstream quota settlement delay"
);
let quota = timeout(Duration::from_secs(5), async {
loop {
let quota = store
.get_quotas(&[account.id().clone()])
.await
.expect("read refreshed quota")
.into_iter()
.next()
.and_then(|observation| observation.quota)
.expect("authoritative quota snapshot");
let quota = Value::Object(quota.into_inner());
if quota
.pointer("/rate_limit/primary_window/used_percent")
.and_then(Value::as_u64)
== Some(100)
{
break quota;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("background usage refresh must replace the stale quota snapshot");
assert_eq!(
quota
.pointer("/rate_limit/primary_window/used_percent")
.and_then(Value::as_u64),
Some(100)
);
let requests = server.received_requests().await.expect("received requests");
assert_eq!(
requests
.iter()
.filter(|request| request.url.path() == "/api/codex/usage")
.count(),
1
);
assert_eq!(
store
.account(account_id)
.expect("account after stale full usage refresh")
.availability(),
AccountAvailability::QuotaExhausted,
"a contradictory full usage snapshot must not unlock a confirmed exhausted account"
);
}
#[tokio::test]
async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_first_event_latency").await;
let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
let (base_url, release, first_chunk_sent, server) = paused_chunked_sse_server(
concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_first_event\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n"
@@ -1004,41 +1178,111 @@ async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
.await
.expect("prepare provider stream");
let first_event = loop {
let next = timeout(Duration::from_secs(1), stream.next())
.await
.expect("first chunk must not wait for the withheld failure")
.expect("provider stream must stay open")
.expect("provider event");
if next.has_client_event() {
break next;
let mut first_chunk_sent = Box::pin(first_chunk_sent);
loop {
tokio::select! {
sent = &mut first_chunk_sent => {
sent.expect("first upstream chunk");
break;
}
next = stream.next() => {
let event = next
.expect("provider stream must stay open")
.expect("provider event");
assert!(!event.has_client_event(), "response.created must remain replayable");
}
}
};
assert_eq!(
first_event.wire_event().and_then(|wire| wire.event_type()),
Some("response.created")
}
let exposed = timeout(Duration::from_millis(100), async {
loop {
let next = stream
.next()
.await
.expect("provider stream must stay open")
.expect("provider event");
if next.has_client_event() {
return next;
}
}
})
.await;
assert!(
exposed.is_err(),
"a structural event must not commit the downstream before a later 429"
);
release.send(()).expect("release second upstream chunk");
let mut later_wire_types = Vec::new();
let failure = loop {
let mut failure = loop {
let next = timeout(Duration::from_secs(1), stream.next())
.await
.expect("later failure must arrive after release")
.expect("provider stream must return typed failure");
match next {
Ok(event) => {
if let Some(event_type) = event.wire_event().and_then(|wire| wire.event_type()) {
later_wire_types.push(event_type.to_owned());
}
}
Ok(event) => assert!(
!event.has_client_event(),
"failure attempt leaked downstream"
),
Err(error) => break error,
}
};
assert_eq!(later_wire_types, vec!["response.failed"]);
assert!(!failure.replay_is_safe());
assert!(!failure.has_atomic_client_events());
assert!(failure.replay_is_safe());
assert_eq!(
failure
.take_atomic_client_events()
.iter()
.filter_map(|event| event.wire_event()?.event_type())
.collect::<Vec<_>>(),
vec!["response.created", "response.failed"]
);
server.await.expect("chunked SSE server");
}
#[tokio::test]
async fn ordinary_request_should_bound_structural_event_replay_grace() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_bounded_replay_grace").await;
let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_bounded_grace\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n"
)
.to_owned(),
String::new(),
)
.await;
let mut stream = provider_with_base_url(&store, base_url)
.execute(
planned_request("openai", http_generate_operation()),
context("req_bounded_replay_grace", CancellationToken::new()),
)
.await
.expect("prepare provider stream");
let first_event = timeout(Duration::from_secs(2), async {
loop {
let event = stream
.next()
.await
.expect("provider stream must stay open")
.expect("provider event");
if event.has_client_event() {
return event;
}
}
})
.await
.expect("response.created must be released after the bounded grace period");
assert_eq!(
first_event.wire_event().and_then(|wire| wire.event_type()),
Some("response.created")
);
release.send(()).expect("finish upstream response");
while let Some(event) = stream.next().await {
event.expect("clean upstream EOF");
}
server.await.expect("chunked SSE server");
}
@@ -36,6 +36,7 @@ struct StoredAccount {
account: ProviderAccount,
credential: gateway_core::engine::credential::PlaintextCredential,
quota: Option<QuotaObservation>,
state_observed_at: Option<SystemTime>,
}
#[derive(Default)]
@@ -129,6 +130,7 @@ impl ProviderAccountStore for MemoryAccountStore {
account: input.account,
credential: input.credential,
quota: None,
state_observed_at: None,
},
);
Ok(())
@@ -260,6 +262,15 @@ impl ProviderAccountStore for MemoryAccountStore {
if stored.account.revision() != observation.expected_revision {
return Ok(QuotaWriteOutcome::Conflict);
}
if stored
.quota
.as_ref()
.and_then(|quota| quota.observed_at)
.zip(observation.observed_at)
.is_some_and(|(current, incoming)| current > incoming)
{
return Ok(QuotaWriteOutcome::Conflict);
}
stored.quota = Some(observation);
Ok(QuotaWriteOutcome::Updated)
}
@@ -272,6 +283,12 @@ impl ProviderAccountStore for MemoryAccountStore {
if stored.account.revision() != change.expected_revision {
return Err(store_error(StoreErrorKind::Conflict));
}
if stored
.state_observed_at
.is_some_and(|current| current > change.observed_at)
{
return Err(store_error(StoreErrorKind::Conflict));
}
stored.account = rebuild_account(
&stored.account,
AccountRebuild {
@@ -285,6 +302,7 @@ impl ProviderAccountStore for MemoryAccountStore {
profile: None,
},
);
stored.state_observed_at = Some(change.observed_at);
Ok(())
}
@@ -469,7 +469,7 @@ impl GrokCredentialQuotaService {
if quota_refresh_may_update_state(&loaded.account, &current.account)
&& let Some(availability) = quota_refresh_availability(
current.account.availability(),
quota_is_exhausted(snapshot.billing()),
authoritative_quota_exhaustion(snapshot.billing()),
)
{
self.repository
@@ -1329,6 +1329,13 @@ fn quota_is_exhausted(billing: &GrokBillingPresentation) -> bool {
)
}
/// `None` means the billing document cannot prove either quota state.
fn authoritative_quota_exhaustion(billing: &GrokBillingPresentation) -> Option<bool> {
billing
.has_authoritative_quota()
.then(|| quota_is_exhausted(billing))
}
fn quota_refresh_may_update_state(before: &ProviderAccount, current: &ProviderAccount) -> bool {
current.enabled()
&& current
@@ -1341,22 +1348,26 @@ fn quota_refresh_may_update_state(before: &ProviderAccount, current: &ProviderAc
fn quota_refresh_availability(
current: AccountAvailability,
exhausted: bool,
exhausted: Option<bool>,
) -> Option<GrokCredentialAvailability> {
match current {
AccountAvailability::Invalid
| AccountAvailability::Expired
| AccountAvailability::Banned => None,
AccountAvailability::QuotaExhausted => {
(!exhausted).then_some(GrokCredentialAvailability::Ready)
matches!(exhausted, Some(false)).then_some(GrokCredentialAvailability::Ready)
}
AccountAvailability::Cooldown | AccountAvailability::Unknown => {
exhausted.map(|exhausted| {
if exhausted {
GrokCredentialAvailability::QuotaExhausted
} else {
GrokCredentialAvailability::Ready
}
})
}
AccountAvailability::Cooldown | AccountAvailability::Unknown => Some(if exhausted {
GrokCredentialAvailability::QuotaExhausted
} else {
GrokCredentialAvailability::Ready
}),
AccountAvailability::Ready => {
exhausted.then_some(GrokCredentialAvailability::QuotaExhausted)
matches!(exhausted, Some(true)).then_some(GrokCredentialAvailability::QuotaExhausted)
}
}
}
@@ -479,15 +479,16 @@ impl GrokSessionSelector for GrokAccountSessionSelector {
"upstream_payment_quota_exhausted",
None,
),
GrokCredentialFailure::ModelQuotaExhausted {
upstream_model,
retry_after,
} => {
GrokCredentialFailure::ModelQuotaExhausted { retry_after, .. } => {
let retry_after =
bounded_cooldown(retry_after, MODEL_QUOTA_COOLDOWN, MAX_MODEL_COOLDOWN);
self.record_model_runtime_cooldown(session, upstream_model, retry_after)
.await;
return;
// xAI 免费模型额度按账号滚动窗口恢复;持久化使页面和各调度节点收敛。
self.record_runtime_cooldown(session, retry_after).await;
(
GrokCredentialAvailability::Cooldown,
"usage_limit_exhausted",
persistent_cooldown_until(observed_at, retry_after),
)
}
GrokCredentialFailure::ModelAccessDenied {
upstream_model,
+87 -40
View File
@@ -60,8 +60,9 @@ use crate::transport::{
GrokCompactionSummaryDecoder, GrokCredentialFailure, GrokInferenceChunkStream,
GrokInferenceRequest, GrokInferenceResponse, GrokInferenceTransport,
GrokInferenceTransportError, GrokInferenceTransportErrorKind, GrokInferenceTransportMetrics,
GrokProviderConfigError, GrokRequestEncodeError, GrokResponsesRequest, GrokSessionAffinityKey,
GrokSessionSelection, GrokSessionSelector, GrokSessionSelectorError, SelectedGrokSession,
GrokProviderConfigError, GrokQuotaFailureKind, GrokRequestEncodeError, GrokResponsesRequest,
GrokSessionAffinityKey, GrokSessionSelection, GrokSessionSelector, GrokSessionSelectorError,
SelectedGrokSession, classify_grok_quota_failure,
};
use crate::{GrokCatalogCapabilityEvidence, GrokCatalogModel};
@@ -1118,9 +1119,21 @@ fn cold_compaction_http_sse_stream(
.await
.map_err(mark_transient_compaction_failure)?
{
let events = canonical.push(&chunk).map_err(|error| {
mark_transient_compaction_failure(map_continuation_failure(&context, error))
})?;
let events = match canonical.push(&chunk) {
Ok(events) => events,
Err(error) => {
let error = map_continuation_failure(&context, error);
let error = record_stream_failure(
selector.as_ref(),
&session,
error,
&upstream_model,
)
.await;
Err(mark_transient_compaction_failure(error))?;
return;
}
};
for event in events {
summary.observe(&event).map_err(map_compaction_decode_error)?;
facts.observe(&event);
@@ -1131,9 +1144,22 @@ fn cold_compaction_http_sse_stream(
}
if facts.completed.is_none() {
for event in canonical.finish_without_terminal().map_err(|error| {
mark_transient_compaction_failure(map_continuation_failure(&context, error))
})? {
let events = match canonical.finish_without_terminal() {
Ok(events) => events,
Err(error) => {
let error = map_continuation_failure(&context, error);
let error = record_stream_failure(
selector.as_ref(),
&session,
error,
&upstream_model,
)
.await;
Err(mark_transient_compaction_failure(error))?;
return;
}
};
for event in events {
summary.observe(&event).map_err(map_compaction_decode_error)?;
facts.observe(&event);
if facts.completed.is_some() {
@@ -1371,7 +1397,13 @@ fn cold_http_sse_stream(
Ok(events) => events,
Err(error) => {
let error = map_continuation_failure(&context, error);
let error = record_stream_failure(selector.as_ref(), &session, error).await;
let error = record_stream_failure(
selector.as_ref(),
&session,
error,
&upstream_model,
)
.await;
Err(error)?;
return;
}
@@ -1399,7 +1431,13 @@ fn cold_http_sse_stream(
Ok(events) => events,
Err(error) => {
let error = map_continuation_failure(&context, error);
let error = record_stream_failure(selector.as_ref(), &session, error).await;
let error = record_stream_failure(
selector.as_ref(),
&session,
error,
&upstream_model,
)
.await;
Err(error)?;
return;
}
@@ -1508,28 +1546,6 @@ fn xai_transport_metadata(
ProviderResponseMetadata::new(serde_json::to_string(&Value::Object(metadata)).ok()?)
}
async fn record_failure(
selector: &dyn GrokSessionSelector,
session: &SelectedGrokSession,
error: ProviderError,
) -> ProviderError {
if !session.allows_account_state_mutation() {
return error;
}
let failure = match error.kind() {
ProviderErrorKind::Unauthorized => Some(GrokCredentialFailure::Unauthorized),
ProviderErrorKind::RateLimited => Some(GrokCredentialFailure::RateLimited {
retry_after: error.retry_after(),
}),
ProviderErrorKind::QuotaExhausted => Some(GrokCredentialFailure::QuotaExhausted),
_ => None,
};
if let Some(failure) = failure {
selector.record_failure(session, failure).await;
}
error
}
async fn record_credential_failure(
selector: &dyn GrokSessionSelector,
session: &SelectedGrokSession,
@@ -1546,17 +1562,13 @@ async fn record_stream_failure(
selector: &dyn GrokSessionSelector,
session: &SelectedGrokSession,
error: ProviderError,
upstream_model: &UpstreamModelId,
) -> ProviderError {
if !session.allows_account_state_mutation() {
return error;
}
if matches!(
error.kind(),
ProviderErrorKind::Unauthorized
| ProviderErrorKind::RateLimited
| ProviderErrorKind::QuotaExhausted
) {
return record_failure(selector, session, error).await;
if let Some(failure) = stream_credential_failure(&error, upstream_model) {
return record_credential_failure(selector, session, error, failure).await;
}
if error.kind() != ProviderErrorKind::Cancelled {
selector
@@ -1566,6 +1578,41 @@ async fn record_stream_failure(
error
}
fn stream_credential_failure(
error: &ProviderError,
upstream_model: &UpstreamModelId,
) -> Option<GrokCredentialFailure> {
match error.kind() {
ProviderErrorKind::Unauthorized => Some(GrokCredentialFailure::Unauthorized),
ProviderErrorKind::RateLimited => Some(GrokCredentialFailure::RateLimited {
retry_after: error.retry_after(),
}),
ProviderErrorKind::QuotaExhausted => {
let visible = error.client_visible_upstream_error();
let quota_kind = classify_grok_quota_failure(
visible
.and_then(ClientVisibleUpstreamError::code)
.or_else(|| error.upstream_code().map(|code| code.as_str())),
visible.and_then(ClientVisibleUpstreamError::error_type),
visible.map(ClientVisibleUpstreamError::message),
);
Some(match quota_kind {
Some(GrokQuotaFailureKind::FreeAccount) => {
GrokCredentialFailure::FreeQuotaExhausted
}
Some(GrokQuotaFailureKind::FreeModelUsage) => {
GrokCredentialFailure::ModelQuotaExhausted {
upstream_model: upstream_model.clone(),
retry_after: error.retry_after(),
}
}
Some(GrokQuotaFailureKind::Account) | None => GrokCredentialFailure::QuotaExhausted,
})
}
_ => None,
}
}
async fn recover_or_record_failure(
selector: &dyn GrokSessionSelector,
recovery: &dyn GrokCredentialRecovery,
@@ -1644,7 +1691,7 @@ async fn map_and_record_stream_transport_failure(
match credential_failure {
Some(failure) => record_credential_failure(selector, session, error, failure).await,
None if request_scoped => error,
None => record_stream_failure(selector, session, error).await,
None => record_stream_failure(selector, session, error, upstream_model).await,
}
}
@@ -19,7 +19,7 @@ use gateway_protocol::openai::sse::{SseEvent, SseEventDecoder};
use serde_json::Value;
use super::request::{GrokResponseTransform, GrokResponsesRequest};
use super::scrub_account_fingerprints;
use super::{classify_grok_quota_failure, scrub_account_fingerprints};
const CONTENTS_PER_OUTPUT: u32 = 1_024;
const LONG_CONTEXT_THRESHOLD: u64 = 200_000;
@@ -785,14 +785,19 @@ fn incomplete_finish_reason(response: &Value) -> FinishReason {
fn upstream_event_error(value: &Value) -> ProviderError {
let code = upstream_error_field(value, "code");
let kind = match code {
Some("invalid_request" | "invalid_prompt") => ProviderErrorKind::InvalidRequest,
Some("unsupported" | "unsupported_feature") => ProviderErrorKind::Unsupported,
Some("unauthorized" | "invalid_token") => ProviderErrorKind::Unauthorized,
Some("permission_denied") => ProviderErrorKind::PermissionDenied,
Some("rate_limit_exceeded") => ProviderErrorKind::RateLimited,
Some("quota_exceeded" | "insufficient_quota") => ProviderErrorKind::QuotaExhausted,
_ => ProviderErrorKind::Unavailable,
let error_type = upstream_error_field(value, "type");
let message = upstream_error_field(value, "message");
let kind = if classify_grok_quota_failure(code, error_type, message).is_some() {
ProviderErrorKind::QuotaExhausted
} else {
match code {
Some("invalid_request" | "invalid_prompt") => ProviderErrorKind::InvalidRequest,
Some("unsupported" | "unsupported_feature") => ProviderErrorKind::Unsupported,
Some("unauthorized" | "invalid_token") => ProviderErrorKind::Unauthorized,
Some("permission_denied") => ProviderErrorKind::PermissionDenied,
Some("rate_limit_exceeded") => ProviderErrorKind::RateLimited,
_ => ProviderErrorKind::Unavailable,
}
};
let mut error = ProviderError::new(kind, UpstreamSendState::Sent)
.redact_sensitive_context("upstream event");
@@ -801,11 +806,11 @@ fn upstream_event_error(value: &Value) -> ProviderError {
}
// 结构化 message/code/type 供原客户端展示与重试分类;message 先脱去
// 账号指纹(上游限流文案内嵌 team UUID),非结构化正文不透出。
if let Some(message) = upstream_error_field(value, "message") {
if let Some(message) = message {
error = error.with_client_visible_upstream_error(ClientVisibleUpstreamError::new(
scrub_account_fingerprints(message),
code.map(str::to_owned),
upstream_error_field(value, "type").map(str::to_owned),
error_type.map(str::to_owned),
));
}
error
@@ -56,6 +56,85 @@ pub use transport::{
};
const UUID_TEXT_LEN: usize = 36;
const FREE_USAGE_SIGNALS: &[&str] = &[
"subscription:free-usage-exhausted",
"subscription_free_usage_exhausted",
"free-usage-exhausted",
"free_usage_exhausted",
"used all the included free usage",
"used all your free usage",
];
const ACCOUNT_QUOTA_SIGNALS: &[&str] = &[
"personal-team-blocked:spending-limit",
"personal_team_blocked_spending_limit",
"quota_exceeded",
"insufficient_quota",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum GrokQuotaFailureKind {
Account,
FreeAccount,
FreeModelUsage,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GrokQuotaSignal {
Account,
FreeUsage,
}
pub(crate) fn classify_grok_quota_failure(
code: Option<&str>,
error_type: Option<&str>,
message: Option<&str>,
) -> Option<GrokQuotaFailureKind> {
let code = code.map(str::to_ascii_lowercase);
let error_type = error_type.map(str::to_ascii_lowercase);
let message = message.map(str::to_ascii_lowercase);
// `code` and `type` are stable upstream fields. Do not let a conflicting
// human-readable message override them; use the message only as a
// fallback or to refine a confirmed free-usage signal's rolling window.
let structured_signal = code
.as_deref()
.and_then(classify_quota_signal)
.or_else(|| error_type.as_deref().and_then(classify_quota_signal));
match structured_signal {
Some(GrokQuotaSignal::Account) => Some(GrokQuotaFailureKind::Account),
Some(GrokQuotaSignal::FreeUsage) => Some(free_usage_failure_kind(message.as_deref())),
None => classify_message_quota_failure(message.as_deref()),
}
}
fn classify_quota_signal(value: &str) -> Option<GrokQuotaSignal> {
if contains_any(value, ACCOUNT_QUOTA_SIGNALS) {
Some(GrokQuotaSignal::Account)
} else if contains_any(value, FREE_USAGE_SIGNALS) {
Some(GrokQuotaSignal::FreeUsage)
} else {
None
}
}
fn classify_message_quota_failure(message: Option<&str>) -> Option<GrokQuotaFailureKind> {
match message.and_then(classify_quota_signal) {
Some(GrokQuotaSignal::Account) => Some(GrokQuotaFailureKind::Account),
Some(GrokQuotaSignal::FreeUsage) => Some(free_usage_failure_kind(message)),
None => None,
}
}
fn free_usage_failure_kind(message: Option<&str>) -> GrokQuotaFailureKind {
if message.is_some_and(|value| value.contains("for model")) {
GrokQuotaFailureKind::FreeModelUsage
} else {
GrokQuotaFailureKind::FreeAccount
}
}
fn contains_any(value: &str, needles: &[&str]) -> bool {
needles.iter().any(|needle| value.contains(needle))
}
/// 把 message 中的 UUID 替换为占位符,控制字符归一为空格,
/// 使文本满足客户端可见错误的约束且不携带可定位账号的标识。
@@ -33,7 +33,7 @@ use gateway_core::engine::UpstreamSendState;
use gateway_core::error::{ClientVisibleUpstreamError, OpaqueUpstreamValue};
use gateway_core::event::UpstreamHttpVersion;
use super::scrub_account_fingerprints;
use super::{GrokQuotaFailureKind, classify_grok_quota_failure, scrub_account_fingerprints};
pub(crate) const OFFICIAL_OAUTH_HOST: &str = "auth.x.ai";
const OFFICIAL_INFERENCE_HOST: &str = "cli-chat-proxy.grok.com";
@@ -1082,15 +1082,10 @@ fn first_string(object: &serde_json::Map<String, Value>, fields: &[&str]) -> Opt
}
fn classify_payment_required(metadata: &InferenceErrorMetadata) -> GrokInferenceTransportErrorKind {
if model_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::ModelQuotaExhausted
} else if paid_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::QuotaExhausted
} else if free_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::FreeQuotaExhausted
} else {
GrokInferenceTransportErrorKind::PaymentRequired
}
quota_failure(metadata).map_or(
GrokInferenceTransportErrorKind::PaymentRequired,
quota_transport_error_kind,
)
}
fn classify_forbidden(
@@ -1103,12 +1098,8 @@ fn classify_forbidden(
GrokInferenceTransportErrorKind::SafetyRejected
} else if credential_rejected(&text) {
GrokInferenceTransportErrorKind::Unauthorized
} else if model_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::ModelQuotaExhausted
} else if paid_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::QuotaExhausted
} else if free_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::FreeQuotaExhausted
} else if let Some(failure) = quota_failure(metadata) {
quota_transport_error_kind(failure)
} else if model_access_denied(metadata) {
GrokInferenceTransportErrorKind::ModelAccessDenied
} else {
@@ -1119,15 +1110,10 @@ fn classify_forbidden(
fn classify_too_many_requests(
metadata: &InferenceErrorMetadata,
) -> GrokInferenceTransportErrorKind {
if model_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::ModelQuotaExhausted
} else if paid_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::QuotaExhausted
} else if free_quota_exhausted(metadata) {
GrokInferenceTransportErrorKind::FreeQuotaExhausted
} else {
GrokInferenceTransportErrorKind::RateLimited
}
quota_failure(metadata).map_or(
GrokInferenceTransportErrorKind::RateLimited,
quota_transport_error_kind,
)
}
fn reasoning_decode_failed(metadata: &InferenceErrorMetadata) -> bool {
@@ -1137,32 +1123,22 @@ fn reasoning_decode_failed(metadata: &InferenceErrorMetadata) -> bool {
})
}
fn paid_quota_exhausted(metadata: &InferenceErrorMetadata) -> bool {
contains_any(
&inference_metadata_text(metadata),
&[
"personal-team-blocked:spending-limit",
"personal_team_blocked_spending_limit",
],
fn quota_failure(metadata: &InferenceErrorMetadata) -> Option<GrokQuotaFailureKind> {
classify_grok_quota_failure(
metadata.code.as_deref(),
metadata.error_type.as_deref(),
metadata.message.as_deref(),
)
}
fn free_quota_exhausted(metadata: &InferenceErrorMetadata) -> bool {
contains_any(
&inference_metadata_text(metadata),
&[
"subscription:free-usage-exhausted",
"subscription_free_usage_exhausted",
"free-usage-exhausted",
"free_usage_exhausted",
"used all the included free usage",
"used all your free usage",
],
)
}
fn model_quota_exhausted(metadata: &InferenceErrorMetadata) -> bool {
inference_metadata_text(metadata).contains("used all the included free usage for model")
fn quota_transport_error_kind(failure: GrokQuotaFailureKind) -> GrokInferenceTransportErrorKind {
match failure {
GrokQuotaFailureKind::Account => GrokInferenceTransportErrorKind::QuotaExhausted,
GrokQuotaFailureKind::FreeAccount => GrokInferenceTransportErrorKind::FreeQuotaExhausted,
GrokQuotaFailureKind::FreeModelUsage => {
GrokInferenceTransportErrorKind::ModelQuotaExhausted
}
}
}
fn model_access_denied(metadata: &InferenceErrorMetadata) -> bool {
@@ -319,7 +319,7 @@ pub enum GrokCredentialFailure {
/// 已解析并脱敏的延迟;缺失时由 selector 使用短暂默认值。
retry_after: Option<Duration>,
},
/// 选中账号只耗尽了指定模型的额度。
/// 上游声明指定模型的免费用量耗尽;selector 会收敛为账号级可恢复额度状态。
ModelQuotaExhausted {
upstream_model: UpstreamModelId,
retry_after: Option<Duration>,
@@ -282,7 +282,7 @@ pub enum GrokInferenceTransportErrorKind {
QuotaExhausted,
/// 账号订阅级免费额度耗尽。
FreeQuotaExhausted,
/// 当前账号只耗尽了指定模型的免费额度。
/// 上游声明指定模型的免费用量耗尽;selector 会收敛为账号级可恢复额度状态。
ModelQuotaExhausted,
/// 当前账号只缺少指定模型的访问权限。
ModelAccessDenied,
@@ -874,17 +874,24 @@ async fn quota_projection_preserves_unknown_period_for_dynamic_duration_fallback
}
#[tokio::test]
async fn weekly_period_without_reported_allowance_is_not_authoritative_quota() {
let (_, repository) = repository_with_accounts(&[("free-quota", "subject-free-quota")]).await;
async fn non_authoritative_quota_refresh_does_not_clear_existing_quota_exhaustion() {
let (store, repository) =
repository_with_accounts(&[("free-quota", "subject-free-quota")]).await;
let id = account_id("free-quota");
set_account_state(&store, &id, AccountAvailability::QuotaExhausted, None).await;
let transport = QueueBillingTransport::success(
br#"{"config":{"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY","start":"2026-07-15T00:00:00Z","end":"2026-07-22T00:00:00Z"},"onDemandCap":{"val":0},"onDemandUsed":{"val":0},"prepaidBalance":{"val":0}}}"#,
);
let snapshot = crate::support::grok_quota_service(repository, transport)
.refresh_account(&account_id("free-quota"))
.refresh_account(&id)
.await
.expect("refresh Free quota");
assert!(!snapshot.billing().has_authoritative_quota());
assert_eq!(
store.account(&id).expect("account").availability(),
AccountAvailability::QuotaExhausted
);
}
#[tokio::test]
@@ -490,7 +490,7 @@ async fn payment_required_feedback_persists_account_quota_exhaustion() {
}
#[tokio::test]
async fn model_quota_feedback_blocks_only_the_failed_account_and_model() {
async fn free_model_quota_feedback_persists_resettable_account_exhaustion() {
let fixture = SelectorFixture::new(&["model-quota", "available"]).await;
let failed_model = UpstreamModelId::new("grok-4.5").expect("failed model");
let session = fixture
@@ -505,38 +505,47 @@ async fn model_quota_feedback_blocks_only_the_failed_account_and_model() {
&session,
GrokCredentialFailure::ModelQuotaExhausted {
upstream_model: failed_model.clone(),
retry_after: Some(Duration::from_secs(5)),
retry_after: None,
},
)
.await;
assert_eq!(
fixture
.store
.account(&selected)
.expect("selected account")
.availability(),
AccountAvailability::Ready
let account = fixture.store.account(&selected).expect("selected account");
assert_eq!(account.availability(), AccountAvailability::Cooldown);
let minimum_until = SystemTime::now() + Duration::from_secs(23 * 60 * 60);
assert!(
account
.cooldown_until()
.is_some_and(|until| until > minimum_until)
);
let scope = ProviderCooldownScope::upstream_model(failed_model);
assert!(
fixture
.cooldowns
.scoped_cooldown(&selected, &scope)
.is_some()
.is_none()
);
let unrelated_model = fixture
assert!(
fixture
.cooldowns
.cooldown(&selected)
.is_some_and(|cooldown| cooldown.until() > minimum_until)
);
assert!(matches!(
fixture
.selector
.select(fixture.request_for_model("grok-4.6", Some(selected.clone())))
.await,
Err(GrokSessionSelectorError::AccountCoolingDown {
retry_after: Some(_)
})
));
let next = fixture
.selector
.select(fixture.request_for_model("grok-4.6", Some(selected.clone())))
.select(fixture.request_for_model("grok-4.6", None))
.await
.expect("same account remains valid for another model");
assert_eq!(unrelated_model.account_id(), &selected);
let failed_model_retry = fixture
.selector
.select(fixture.request_for_model("grok-4.5", None))
.await
.expect("another account serves the failed model");
assert_ne!(failed_model_retry.account_id(), &selected);
.expect("another account remains schedulable");
assert_ne!(next.account_id(), &selected);
}
#[tokio::test]
@@ -1518,6 +1518,73 @@ async fn model_quota_transport_failure_preserves_model_scoped_feedback() {
);
}
#[tokio::test]
async fn free_quota_sse_failure_marks_the_account_exhausted() {
let selector = StubSelector::success();
let body = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_free_quota\",\"model\":\"grok-4.5\"}}\n\n",
"event: response.failed\n",
"data: {\"type\":\"response.failed\",\"error\":{\"code\":\"subscription:free-usage-exhausted\",\"message\":\"You have used all your free usage\"}}\n\n",
)
.as_bytes()
.to_vec();
let transport = StubInferenceTransport::sequence([InferenceMode::SuccessBody(body)]);
let provider = provider(selector.clone(), transport).await;
let mut stream = provider
.execute(
provider_request("xai"),
context(CancellationToken::new(), None),
)
.await
.expect("stream");
let error = next_provider_error(&mut stream).await;
assert_eq!(error.kind(), ProviderErrorKind::QuotaExhausted);
assert_eq!(
selector.feedback.lock().expect("feedback").as_slice(),
&[GrokCredentialFailure::FreeQuotaExhausted]
);
}
#[tokio::test]
async fn free_model_quota_sse_failure_is_classified_for_resettable_account_cooldown() {
let selector = StubSelector::success();
let body = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_model_quota\",\"model\":\"grok-4.5\"}}\n\n",
"event: response.failed\n",
"data: {\"type\":\"response.failed\",\"error\":{\"status\":null,\"contentType\":null,\"body\":null,\"code\":\"subscription_free_usage_exhausted\",\"type\":null,\"message\":\"You've used all the included free usage for model grok-4.5-0722 for now. Usage resets over a rolling 24-hour window \u{2014} tokens (actual/limit): 500505/500000. Upgrade to a Grok subscription for higher limits: https://grok.com/supergrok\"}}\n\n",
)
.as_bytes()
.to_vec();
let transport = StubInferenceTransport::sequence([InferenceMode::SuccessBody(body)]);
let provider = provider(selector.clone(), transport).await;
let mut stream = provider
.execute(
provider_request("xai"),
context(CancellationToken::new(), None),
)
.await
.expect("stream");
let error = next_provider_error(&mut stream).await;
assert_eq!(error.kind(), ProviderErrorKind::QuotaExhausted);
assert_eq!(
error.upstream_code().map(|code| code.as_str()),
Some("subscription_free_usage_exhausted")
);
assert_eq!(
selector.feedback.lock().expect("feedback").as_slice(),
&[GrokCredentialFailure::ModelQuotaExhausted {
upstream_model: UpstreamModelId::new("grok-4.5").expect("model"),
retry_after: None,
}]
);
}
#[tokio::test]
async fn model_access_denial_is_retryable_without_credential_recovery() {
let selector = StubSelector::success();
@@ -802,6 +802,28 @@ fn decoder_should_classify_failed_event_without_retaining_body() {
assert!(!format!("{error:?}").contains("secret"));
}
#[test]
fn decoder_should_classify_free_quota_failed_event() {
let body = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_quota\"}}\n\n",
"event: response.failed\n",
"data: {\"type\":\"response.failed\",\"error\":{\"code\":\"subscription:free-usage-exhausted\",\"type\":\"billing_error\",\"message\":\"You have used all your free usage\"}}\n\n",
);
let error = GrokCanonicalDecoder::new("fallback")
.push(body.as_bytes())
.expect_err("free quota failure must surface");
assert_eq!(
error.kind(),
gateway_core::error::ProviderErrorKind::QuotaExhausted
);
assert_eq!(
error.upstream_code().map(|code| code.as_str()),
Some("subscription:free-usage-exhausted")
);
}
#[test]
fn decoder_should_scrub_account_fingerprints_from_failure_messages() {
let body = concat!(
@@ -471,8 +471,12 @@ async fn inference_transport_should_classify_model_quota_for_403_and_429() {
.and(path("/v1/responses"))
.respond_with(ResponseTemplate::new(status).set_body_json(json!({
"error": {
"code": "subscription:free-usage-exhausted",
"message": "You have used all the included free usage for model grok-4.5"
"status": null,
"contentType": null,
"body": null,
"code": "subscription_free_usage_exhausted",
"type": null,
"message": "You've used all the included free usage for model grok-4.5-0722 for now. Usage resets over a rolling 24-hour window - tokens (actual/limit): 500505/500000. Upgrade to a Grok subscription for higher limits: https://grok.com/supergrok"
}
})))
.expect(1)
@@ -491,6 +495,38 @@ async fn inference_transport_should_classify_model_quota_for_403_and_429() {
}
}
#[tokio::test]
async fn inference_transport_should_prefer_structured_quota_fields_over_conflicting_message() {
for (code, error_type) in [
("quota_exceeded", Some("subscription_free_usage_exhausted")),
("rate_limit_exceeded", Some("quota_exceeded")),
] {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(ResponseTemplate::new(429).set_body_json(json!({
"error": {
"code": code,
"type": error_type,
"message": "You've used all the included free usage for model grok-4.5-0722 for now."
}
})))
.expect(1)
.mount(&server)
.await;
let origin = Url::parse(&server.uri()).expect("wiremock origin");
let error = inference_transport(&origin)
.execute(inference_request(&origin))
.await
.expect_err("quota response must fail");
assert_eq!(
error.kind(),
GrokInferenceTransportErrorKind::QuotaExhausted
);
}
}
#[tokio::test]
async fn inference_transport_should_classify_free_usage_429_as_account_free_quota_exhausted() {
let server = MockServer::start().await;
+1 -5
View File
@@ -1,5 +1 @@
0554fed953bd1440b50bc4033dc8bad96b51e132c57c2ccf1553728a991d5cb2 0001_initial.sql
0ae3945c02461347ab8dcfc64821027cd2366505778e0450d71bd82772c4e99d 0002_drop_opaque_response_id_index.sql
e608ed89991f9d6317491714447a00a61ede90caca4fb030eb30695654afb664 0003_store_opaque_response_ids_as_bytes.sql
df4e02cf2335733326c12429ced4cbd0b35b8a22b3fc24981c3c0549477d7c43 0004_add_model_request_service_tier.sql
fe8f36913c6db7053148db450f05e008dab6cdb9ca8cdcfcb22d97ad9a87bf49 0005_allow_quota_exhausted_cooldown.sql
a4681a5c040e80c5234e879293fc34972080e5251dfbc7c40e1374f8e4170a4c 0001_initial.sql
+20 -7
View File
@@ -166,7 +166,14 @@ create table provider_accounts (
)
),
constraint provider_accounts_cooldown_ck check (
(availability = 'cooldown') = (cooldown_until is not null)
(
availability != 'cooldown'
or cooldown_until is not null
)
and (
cooldown_until is null
or availability in ('cooldown', 'quota_exhausted')
)
),
constraint provider_accounts_time_ck check (
created_at <= updated_at
@@ -208,6 +215,7 @@ create table model_requests (
requested_model_id text not null,
provider_kind text,
upstream_model_id text,
service_tier text,
provider_account_id text,
provider_account_ref text,
upstream_transport text,
@@ -220,9 +228,11 @@ create table model_requests (
outcome text not null default 'running',
client_status_code integer,
upstream_status_code integer,
client_response_id text,
-- Provider response IDs are opaque JSON strings. Bytes preserve embedded NUL
-- and avoid imposing PostgreSQL text-index limits on protocol handles.
client_response_id bytea,
upstream_request_id text,
upstream_response_id text,
upstream_response_id bytea,
error_kind text,
provider_error_code text,
error_message text,
@@ -321,6 +331,13 @@ create table model_requests (
provider_observation_json is null
or jsonb_typeof(provider_observation_json) = 'object'
),
constraint model_requests_service_tier_ck check (
service_tier is null
or (
octet_length(service_tier) between 1 and 64
and service_tier !~ '[[:cntrl:]]'
)
),
constraint model_requests_cost_ck check (
cost_source in ('provider_reported', 'calculated', 'unavailable')
and (
@@ -414,10 +431,6 @@ create index model_requests_running_deadline_idx
create index model_requests_retention_idx
on model_requests (completed_at)
where outcome <> 'running';
create unique index model_requests_client_response_uq
on model_requests (client_response_id)
where client_response_id is not null;
create table ops_events (
id text primary key,
model_request_id text,
@@ -1,3 +0,0 @@
-- Provider response IDs are opaque protocol values, not relational identifiers.
-- Indexing the raw text makes PostgreSQL reject otherwise valid long handles.
drop index model_requests_client_response_uq;
@@ -1,7 +0,0 @@
-- Response IDs are opaque JSON strings. PostgreSQL text rejects NUL even though JSON accepts it.
-- Store their UTF-8 bytes without an index; Rust decodes them only at the audit/API boundary.
alter table model_requests
alter column client_response_id type bytea
using convert_to(client_response_id, 'UTF8'),
alter column upstream_response_id type bytea
using convert_to(upstream_response_id, 'UTF8');
@@ -1,9 +0,0 @@
alter table model_requests
add column service_tier text,
add constraint model_requests_service_tier_ck check (
service_tier is null
or (
octet_length(service_tier) between 1 and 64
and service_tier !~ '[[:cntrl:]]'
)
);
@@ -1,12 +0,0 @@
alter table provider_accounts
drop constraint provider_accounts_cooldown_ck,
add constraint provider_accounts_cooldown_ck check (
(
availability != 'cooldown'
or cooldown_until is not null
)
and (
cooldown_until is null
or availability in ('cooldown', 'quota_exhausted')
)
);
+7 -5
View File
@@ -6,14 +6,16 @@ sqlx 在服务启动时(`serve` 监听之前)按编号顺序执行本目录
## 冻结规则
- 已合入 main 的迁移文件**永久冻结**,一个字节都不能改——包括注释和空白。
任何改动都会让所有已部署实例(含自更新链路上的存量用户)在下次重启时
因 checksum 失配而无法启动,且没有自动恢复路径。
- schema 变更一律新增编号迁移(`0003_...`、`0004_...`),永远不回改旧文件。
- 同一大版本内,已合入 main 的迁移文件**永久冻结**,一个字节都不能改——包括注释和
空白。任何改动都会让该大版本的已部署实例在下次重启时因 checksum 失配而无法启动,
且没有自动恢复路径。
- 同一大版本内的 schema 变更一律新增编号迁移(`0002_...`、`0003_...`),永远不回改
旧文件。切换大版本时允许把已完成升级的 schema 折叠成新的 `0001` 基线,因为跨大版本
不支持在线升级,目标版本必须使用全新数据目录部署。
- `.frozen-sha256` 是冻结清单(`sha256sum` 格式)。CI 会校验:清单内文件
字节未变,且目录里每个 `*.sql` 都已入册。新增迁移时在同一提交里执行
`sha256sum 000N_xxx.sql >> .frozen-sha256` 入册。
- 若确需修正已冻结迁移里的错误,用新迁移做补偿性变更(`alter` / 回填),
- 若同一大版本内确需修正已冻结迁移里的错误,用新迁移做补偿性变更(`alter` / 回填),
不要动原文件。
## 本地测试库
+2 -4
View File
@@ -130,10 +130,8 @@ Provider schema 以明文 JSON 保存于 PostgreSQL。Redis 只保存可重建
## 镜像升级与源码构建
> [!WARNING]
> 以下命令只适用于同一大版本内的升级。数据库 schema 与配置只在同一大版本内兼容,
> 跨大版本直接换镜像会导致服务无法启动;内置一键自更新会拒绝跨大版本目标(3.0 之前
> 的旧版没有这道门禁,不要用旧版界面的一键更新跨大版本)。跨大版本请全新部署(全新
> `.runtime/` 数据目录),并重新导入或重新授权 Provider 账号与客户端 Key。
> 以下命令只适用于同一大版本内的升级,不支持跨大版本在线升级。跨大版本请使用全新的
> `.runtime/` 数据目录重新部署,并重新导入或重新授权 Provider 账号与客户端 Key。
```bash
docker compose -f deploy/compose.yaml build codex-proxy-rs
+5 -5
View File
@@ -170,11 +170,11 @@ Admin API 不要求客户端提交配置 revision,也不向客户端暴露配
客户端可见协议结果。队列不是第二份业务权威,也不做无幂等键的盲目重试。
- retention 只删除已满足保留规则的历史事实,不改变运行中请求。
业务 schema 由 `backend/migrations/` 下的编号迁移按序定义:`0001_initial.sql` 建表,`0002` 删除
opaque response ID 唯一索引,`0003` 将 client/upstream response ID 改为无格式假设的 UTF-8 bytes,
`0004` 增加响应观测得到的 `model_requests.service_tier`。已合入的迁移永久冻结——
`.frozen-sha256` 是冻结清单,CI 校验清单相对 PR base 只增不改,启动时 sqlx 重校验已应用迁移的
checksum,不一致直接拒绝启动。规则见 `backend/migrations/README.md`。
业务 schema 由 `backend/migrations/0001_initial.sql` 直接定义当前大版本的完整新库基线,其中
response ID 使用无格式假设的 UTF-8 bytes,`model_requests.service_tier` 保存响应观测到的服务档位,
且额度耗尽账号可以携带冷却截止时间。同一大版本内已合入的迁移永久冻结——
`.frozen-sha256` 是冻结清单,CI 校验清单相对 PR base 只增不改,启动时 sqlx 重校验已应用
迁移的 checksum,不一致直接拒绝启动。规则见 `backend/migrations/README.md`。
## 9. Redis 终态
@@ -193,19 +193,6 @@ function connectionLogClass(tone: string) {
</div>
</BaseScrollbar>
</div>
<div v-if="error" class="mt-3 rounded-lg bg-(--cp-danger-bg) px-3 py-2.5">
<p class="m-0 text-[11px] font-heavy text-(--cp-danger-text)">
错误信息
</p>
<BaseScrollbar max-height="118px" view-class="pt-1.5 pr-2">
<p
class="m-0 wrap-break-word text-[12px] leading-[1.55] font-emphasis text-(--cp-danger-text)"
>
{{ error }}
</p>
</BaseScrollbar>
</div>
</section>
</div>
@@ -33,7 +33,7 @@ const overviewItems = computed(() => [
icon: Gauge,
},
{
label: '不可用账号',
label: '异常账号',
value: formatCount(props.summary.unavailable),
caption: '过期 / 禁用 / 封禁',
tone: 'danger',
@@ -280,17 +280,7 @@ export function useAccountConnectionTest(options: AccountConnectionTestOptions)
if (event.type === 'error') {
applyAccountStatus(event.accountStatus)
connectionTestError.value = connectionTestErrorText(event)
const detail = event.upstreamStatus || event.upstreamContentType || event.upstreamBody
|| event.providerErrorCode || event.providerErrorType
? {
status: event.upstreamStatus,
contentType: event.upstreamContentType,
body: event.upstreamBody,
code: event.providerErrorCode,
type: event.providerErrorType,
}
: undefined
appendConnectionTestLog(connectionTestError.value, 'danger', detail)
appendConnectionTestLog('上游返回错误', 'danger', connectionTestError.value)
finishConnectionTest('error')
clearConnectionTestRun()
}
@@ -148,15 +148,20 @@ export function useAccountOnboarding(options: {
}
async function importAccountDocument() {
const data = parseImportJson(createForm.value.importText)
if (Array.isArray(data) || typeof data !== 'object' || data === null)
throw new Error('导入文件必须是 JSON object')
const result = await importAccounts({
provider: createForm.value.provider,
data,
})
const provider = createForm.value.provider === 'xai' ? 'xAI' : 'OpenAI'
return `${provider} 账号已导入 ${result.importedCount} 个`
const provider = requireImportProvider(createForm.value.provider)
const documents = providerImportDocuments(
parseImportJson(createForm.value.importText),
provider,
)
let importedCount = 0
for (const entry of documents) {
const result = await importAccounts({
provider,
data: entry.document,
})
importedCount += result.importedCount
}
return `${providerDisplayName(provider) ?? provider} 账号已导入 ${importedCount} 个`
}
async function importMixedAccountDocument() {
@@ -259,6 +264,27 @@ function parseImportJson(value: string) {
}
}
function requireImportProvider(value: string): ImportProvider {
if (value === 'openai' || value === 'xai')
return value
throw new Error('请选择要导入的账号平台')
}
function providerImportDocuments(value: unknown, provider: ImportProvider): MixedImportDocument[] {
if (isRecord(value) && Array.isArray(value.documents)) {
const documents = parseMixedImportDocuments(value)
.filter(entry => entry.provider === provider)
if (documents.length === 0) {
const label = providerDisplayName(provider) ?? provider
throw new Error(`批量导入文件不包含 ${label} 账号文档`)
}
return documents
}
if (!isRecord(value))
throw new Error('导入文件必须是 JSON object')
return [{ provider, document: value }]
}
function parseMixedImportDocuments(value: unknown): MixedImportDocument[] {
if (!isRecord(value) || !Array.isArray(value.documents))
throw new Error('批量导入文件必须是 CPR 多平台导出文件')
@@ -56,17 +56,17 @@ function itemValueClass(tone?: string, accent?: boolean) {
<UsageDetailPopover v-if="billing" title="计费明细" width="248px" trigger-label="查看费用明细">
<div class="grid gap-1.5 text-(--cp-text-secondary)">
<div v-for="item in amountItems" :key="item.label" class="flex justify-between gap-4">
<span>{{ item.label }}</span>
<span class="font-mono font-heavy" :class="itemValueClass(undefined, item.accent)">
<div v-for="item in amountItems" :key="item.label" class="grid grid-cols-[auto_minmax(0,1fr)] items-center gap-4">
<span class="whitespace-nowrap">{{ item.label }}</span>
<span class="justify-self-end whitespace-nowrap font-mono font-heavy" :class="itemValueClass(undefined, item.accent)">
{{ item.value }}
</span>
</div>
</div>
<div class="mt-1 grid gap-1.5 rounded-(--cp-input-radius-base) bg-(--cp-bg-subtle) p-2 text-(--cp-text-secondary)">
<div v-for="item in billingItems" :key="item.label" class="flex justify-between gap-4">
<span>{{ item.label }}</span>
<span class="font-mono font-heavy" :class="itemValueClass(item.tone)">
<div v-for="item in billingItems" :key="item.label" class="grid grid-cols-[auto_minmax(0,1fr)] items-center gap-4">
<span class="whitespace-nowrap">{{ item.label }}</span>
<span class="justify-self-end whitespace-nowrap font-mono font-heavy" :class="itemValueClass(item.tone)">
{{ item.value }}
</span>
</div>
+1 -1
View File
@@ -1 +1 @@
version: "3.0.26"
version: "3.0.30"
@@ -40,9 +40,9 @@
## 存储
- PostgreSQL 业务表只有 `0001_initial.sql` 定义的七张;后续编号迁移只做增量。当前 `0002` 删除
opaque response ID 索引,`0003` 把 response ID 改为 bytes,`0004` 增加响应观测的 service tier。
- 已应用迁移按字节冻结:`backend/migrations/.frozen-sha256` 是冻结清单,CI 相对 PR base 做 append-only 校验;schema 变更一律新增编号迁移,规则见 `backend/migrations/README.md`。
- PostgreSQL 业务表只有 `0001_initial.sql` 定义的七张;该文件直接表达当前大版本的完整新库基线,
包括 bytes response ID、响应观测 service tier 与额度耗尽冷却约束。
- 同一大版本内已应用迁移按字节冻结:`backend/migrations/.frozen-sha256` 是冻结清单,CI 相对 PR base 做 append-only 校验;schema 变更一律新增编号迁移,规则见 `backend/migrations/README.md`。
- `config_revision` 只用于会改变调度快照或安全配置的管理 mutation。
- quota、cooldown、catalog generation、自动 refresh 不推进全局 revision。
- refresh 只推进账号 `credential_revision`;Redis cooldown 是可丢失热缓存。