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

fix: stop recording account state usage events

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-07 01:54:14 +08:00
co-authored by Codex
parent 352b781340
commit 1ac9803c34
10 changed files with 81 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/admin/monitoring/usage_record_routes.rs
+1 -1
View File
@@ -463,7 +463,7 @@ fn log_error(error: &AdminUsageRecordError) -> AdminError {
| AdminUsageRecordError::Get
| AdminUsageRecordError::Clear
| AdminUsageRecordError::Append
| AdminUsageRecordError::Trim => AdminError::internal(error.to_string()),
| AdminUsageRecordError::Retention => AdminError::internal(error.to_string()),
AdminUsageRecordError::Accounts => {
AdminError::usage_record_accounts_failed(error.to_string())
}
backend/src/admin/monitoring/usage_record_service.rs
+6 -13
View File
@@ -117,9 +117,9 @@ pub enum AdminUsageRecordError {
/// 写入失败。
#[error("failed to append usage record")]
Append,
/// 裁剪失败。
#[error("failed to trim usage records")]
Trim,
/// 保留期清理失败。
#[error("failed to trim expired usage records")]
Retention,
/// 账号关联信息读取失败。
#[error("failed to load usage record accounts")]
Accounts,
@@ -135,23 +135,16 @@ pub struct AdminUsageRecordService {
#[derive(Debug, Clone, Copy)]
struct AdminUsageRecordSettings {
enabled: bool,
capacity: u32,
capture_body: bool,
}
impl AdminUsageRecordService {
/// 构造管理端使用记录服务。
pub fn new(
store: SqliteUsageRecordStore,
enabled: bool,
capacity: u32,
capture_body: bool,
) -> Self {
pub fn new(store: SqliteUsageRecordStore, enabled: bool, capture_body: bool) -> Self {
Self {
store,
settings: Arc::new(tokio::sync::RwLock::new(AdminUsageRecordSettings {
enabled,
capacity,
capture_body,
})),
}
@@ -289,9 +282,9 @@ impl AdminUsageRecordService {
.await
.map_err(|_| AdminUsageRecordError::Append)?;
self.store
.trim_to_capacity(settings.capacity)
.trim_to_retention(Utc::now())
.await
.map_err(|_| AdminUsageRecordError::Trim)?;
.map_err(|_| AdminUsageRecordError::Retention)?;
Ok(())
}
}
backend/src/admin/monitoring/usage_record_store.rs
+15 -17
View File
@@ -2,7 +2,7 @@
use std::collections::{BTreeMap, HashMap};
use chrono::{DateTime, Utc};
use chrono::{DateTime, Duration, Utc};
use serde_json::Value;
use sqlx::{QueryBuilder, Row, Sqlite, SqlitePool};
use thiserror::Error;
@@ -19,8 +19,8 @@ use crate::infra::{
time::{china_quarter_hour_start, parse_rfc3339_utc as parse_rfc3339},
};
/// 默认保留最新使用记录数量。
pub const DEFAULT_USAGE_RECORD_CAPACITY: u32 = 2_000;
/// 使用记录保留天数。
pub const USAGE_RECORD_RETENTION_DAYS: i64 = 30;
/// 默认不保存请求/响应体，避免把用户内容写入诊断日志。
pub const DEFAULT_USAGE_RECORD_CAPTURE_BODY: bool = false;
@@ -105,9 +105,9 @@ impl SqliteUsageRecordStore {
append_event(&self.pool, event).await
}
/// 保留最新的指定数量使用记录。
pub async fn trim_to_capacity(&self, capacity: u32) -> SqliteUsageRecordStoreResult<u64> {
trim_to_capacity(&self.pool, capacity).await
/// 按保留期清理使用记录。
pub async fn trim_to_retention(&self, now: DateTime<Utc>) -> SqliteUsageRecordStoreResult<u64> {
trim_to_retention(&self.pool, now).await
}
/// 分页查询使用记录。
@@ -425,17 +425,15 @@ fn metadata_first_token_latency(metadata: &Value) -> Option<i64> {
})
}
async fn trim_to_capacity(pool: &SqlitePool, capacity: u32) -> SqliteUsageRecordStoreResult<u64> {
let result = sqlx::query(
"delete from usage_records where id in (
select id from usage_records
order by created_at desc, id desc
limit -1 offset ?
)",
)
.bind(i64::from(capacity))
.execute(pool)
.await?;
async fn trim_to_retention(
pool: &SqlitePool,
now: DateTime<Utc>,
) -> SqliteUsageRecordStoreResult<u64> {
let cutoff = now - Duration::days(USAGE_RECORD_RETENTION_DAYS);
let result = sqlx::query("delete from usage_records where created_at < ?")
.bind(cutoff.to_rfc3339())
.execute(pool)
.await?;
Ok(result.rows_affected())
}
backend/src/runtime/services.rs
+6 -15
View File
@@ -7,10 +7,7 @@ use crate::{
account_usage_service::AdminUsageService,
account_usage_store::SqliteUsageStore,
usage_record_service::AdminUsageRecordService,
usage_record_store::{
SqliteUsageRecordStore, DEFAULT_USAGE_RECORD_CAPACITY,
DEFAULT_USAGE_RECORD_CAPTURE_BODY,
},
usage_record_store::{SqliteUsageRecordStore, DEFAULT_USAGE_RECORD_CAPTURE_BODY},
},
admin::{
accounts::service::{AdminAccountService, AdminAccountServiceParts},
@@ -110,7 +107,6 @@ pub struct Services {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UsageRecordOptions {
pub enabled: bool,
pub capacity: u32,
pub capture_body: bool,
}
@@ -118,7 +114,6 @@ impl UsageRecordOptions {
pub fn from_config(config: &AppConfig) -> Self {
Self {
enabled: config.logging.enabled,
capacity: DEFAULT_USAGE_RECORD_CAPACITY,
capture_body: DEFAULT_USAGE_RECORD_CAPTURE_BODY,
}
}
@@ -192,17 +187,13 @@ impl Services {
let usage_records = StdArc::new(AdminUsageRecordService::new(
stores.usage_records.clone(),
usage_record_options.enabled,
usage_record_options.capacity,
usage_record_options.capture_body,
));
let account_pool = StdArc::new(
RuntimeAccountPoolService::new(
account_store_trait.clone(),
account_pool_options_from_config(config),
config.auth.request_interval_ms,
)
.with_usage_records(usage_records.clone()),
);
let account_pool = StdArc::new(RuntimeAccountPoolService::new(
account_store_trait.clone(),
account_pool_options_from_config(config),
config.auth.request_interval_ms,
));
let refresh_policy = RuntimeRefreshPolicy::new(RefreshPolicy {
refresh_margin_seconds: config.auth.refresh_margin_seconds,
refresh_concurrency: config.auth.refresh_concurrency,
backend/src/upstream/accounts/pool.rs
+21 -251
View File
@@ -9,13 +9,8 @@ use std::{
use chrono::{DateTime, Duration, Utc};
use indexmap::IndexMap;
use serde_json::{json, Value};
use thiserror::Error;
use crate::admin::monitoring::{
usage_record_model::{UsageRecord, UsageRecordLevel},
usage_record_service::AdminUsageRecordService,
};
use crate::infra::time::parse_rfc3339_utc;
use crate::upstream::accounts::quota::{quota_snapshot_limit_reached, quota_snapshot_reset_at};
use crate::upstream::accounts::store::AccountStore;
@@ -232,102 +227,12 @@ impl RuntimeAccountStateSnapshot {
|| self.window_reset_at != other.window_reset_at
|| self.limit_window_seconds != other.limit_window_seconds
}
fn metadata_value(&self) -> Value {
json!({
"status": self.status.as_str(),
"quotaLimitReached": self.quota_limit_reached,
"quotaVerifyRequired": self.quota_verify_required,
"quotaCooldownUntil": self.quota_cooldown_until,
"edgeCooldownUntil": self.cloudflare_cooldown_until,
"windowRequestCount": self.window_request_count,
"windowInputTokens": self.window_input_tokens,
"windowOutputTokens": self.window_output_tokens,
"windowCachedTokens": self.window_cached_tokens,
"windowImageInputTokens": self.window_image_input_tokens,
"windowImageOutputTokens": self.window_image_output_tokens,
"windowImageRequestCount": self.window_image_request_count,
"windowImageRequestFailedCount": self.window_image_request_failed_count,
"windowStartedAt": self.window_started_at,
"windowResetAt": self.window_reset_at,
"limitWindowSeconds": self.limit_window_seconds,
})
}
}
#[derive(Debug, Clone)]
struct RefreshedAccountState {
account: Account,
sync_usage_window: bool,
transition: AccountStateTransition,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AccountStateTransitionReason {
AccessTokenExpired,
RuntimeQuotaWindowRefreshed,
RuntimeStateRefreshed,
StatusSet,
QuotaCooldownSet,
QuotaSnapshotApplied,
EdgeCooldownSet,
}
impl AccountStateTransitionReason {
fn as_str(self) -> &'static str {
match self {
Self::AccessTokenExpired => "access_token_expired",
Self::RuntimeQuotaWindowRefreshed => "runtime_quota_window_refreshed",
Self::RuntimeStateRefreshed => "runtime_state_refreshed",
Self::StatusSet => "status_set",
Self::QuotaCooldownSet => "quota_cooldown_set",
Self::QuotaSnapshotApplied => "quota_snapshot_applied",
Self::EdgeCooldownSet => "edge_cooldown_set",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct AccountStateTransition {
account_id: String,
reason: AccountStateTransitionReason,
before: RuntimeAccountStateSnapshot,
after: RuntimeAccountStateSnapshot,
sync_usage_window: bool,
}
impl AccountStateTransition {
fn new(
account_id: &str,
reason: AccountStateTransitionReason,
before: Option<RuntimeAccountStateSnapshot>,
after: Option<RuntimeAccountStateSnapshot>,
) -> Option<Self> {
let before = before?;
let after = after?;
if before == after {
return None;
}
let sync_usage_window = before.usage_window_changed(&after);
Some(Self {
account_id: account_id.to_string(),
reason,
before,
after,
sync_usage_window,
})
}
fn metadata(&self) -> Value {
json!({
"reason": self.reason.as_str(),
"oldStatus": self.before.status.as_str(),
"newStatus": self.after.status.as_str(),
"syncUsageWindow": self.sync_usage_window,
"before": self.before.metadata_value(),
"after": self.after.metadata_value(),
})
}
}
/// 运行时窗口用量增量。
@@ -657,18 +562,10 @@ impl AccountPool {
}
let after = RuntimeAccountStateSnapshot::from(&*account);
if before != after {
let reason = runtime_refresh_transition_reason(&before, &after);
let sync_usage_window = before.usage_window_changed(&after);
refreshed_accounts.push(RefreshedAccountState {
account: account.clone(),
sync_usage_window,
transition: AccountStateTransition {
account_id: account_id.clone(),
reason,
before,
after,
sync_usage_window,
},
});
}
}
@@ -990,26 +887,12 @@ fn access_token_expired(account: &Account, now: DateTime<Utc>) -> bool {
}
}
fn runtime_refresh_transition_reason(
before: &RuntimeAccountStateSnapshot,
after: &RuntimeAccountStateSnapshot,
) -> AccountStateTransitionReason {
if before.status == AccountStatus::Active && after.status == AccountStatus::Expired {
return AccountStateTransitionReason::AccessTokenExpired;
}
if before.quota_limit_reached && !after.quota_limit_reached && after.quota_verify_required {
return AccountStateTransitionReason::RuntimeQuotaWindowRefreshed;
}
AccountStateTransitionReason::RuntimeStateRefreshed
}
/// 运行时账号池服务。
#[derive(Clone)]
pub struct RuntimeAccountPoolService {
pool: StdArc<tokio::sync::Mutex<AccountPool>>,
store: StdArc<dyn AccountStore>,
request_interval: StdArc<std::sync::RwLock<StdDuration>>,
usage_records: Option<StdArc<AdminUsageRecordService>>,
}
impl RuntimeAccountPoolService {
@@ -1025,16 +908,9 @@ impl RuntimeAccountPoolService {
request_interval: StdArc::new(std::sync::RwLock::new(StdDuration::from_millis(
request_interval_ms,
))),
usage_records: None,
}
}
/// 注入账号状态审计事件记录器。
pub fn with_usage_records(mut self, usage_records: StdArc<AdminUsageRecordService>) -> Self {
self.usage_records = Some(usage_records);
self
}
/// 更新账号池运行参数。
pub async fn apply_options(&self, mut options: AccountPoolOptions, request_interval_ms: u64) {
let mut pool = self.pool.lock().await;
@@ -1146,28 +1022,7 @@ impl RuntimeAccountPoolService {
false
}
};
let (in_memory, transition) = {
let mut pool = self.pool.lock().await;
let before = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
let in_memory = pool.set_status(account_id, status);
let after = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
(
in_memory,
AccountStateTransition::new(
account_id,
AccountStateTransitionReason::StatusSet,
before,
after,
),
)
};
if persisted {
self.record_account_state_transition(transition).await;
}
let in_memory = self.pool.lock().await.set_status(account_id, status);
persisted || in_memory
}
@@ -1184,28 +1039,11 @@ impl RuntimeAccountPoolService {
false
}
};
let (in_memory, transition) = {
let mut pool = self.pool.lock().await;
let before = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
let in_memory = pool.mark_quota_limited_until(account_id, until);
let after = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
(
in_memory,
AccountStateTransition::new(
account_id,
AccountStateTransitionReason::QuotaCooldownSet,
before,
after,
),
)
};
if persisted {
self.record_account_state_transition(transition).await;
}
let in_memory = self
.pool
.lock()
.await
.mark_quota_limited_until(account_id, until);
persisted || in_memory
}
@@ -1231,7 +1069,6 @@ impl RuntimeAccountPoolService {
async fn persist_runtime_account_states(&self, refreshed_states: Vec<RefreshedAccountState>) {
for refreshed in refreshed_states {
let transition = refreshed.transition.clone();
if let Err(error) = self
.store
.sync_runtime_account_state(&refreshed.account, refreshed.sync_usage_window)
@@ -1242,37 +1079,10 @@ impl RuntimeAccountPoolService {
error = %error,
"failed to persist refreshed runtime account state"
);
} else {
self.record_account_state_transition(Some(transition)).await;
}
}
}
async fn record_account_state_transition(&self, transition: Option<AccountStateTransition>) {
let Some(transition) = transition else {
return;
};
let Some(usage_records) = &self.usage_records else {
return;
};
let mut event = UsageRecord::new(
"account.state",
UsageRecordLevel::Info,
"account runtime state changed",
);
event.account_id = Some(transition.account_id.clone());
event.metadata = transition.metadata();
if let Err(error) = usage_records.record(event).await {
tracing::warn!(
account_id = %transition.account_id,
reason = transition.reason.as_str(),
error = %error,
"failed to record account state transition"
);
}
}
/// 按账号上一个在途槽位控制请求间隔。
pub async fn wait_for_request_interval(&self, acquired: &AcquiredAccount) {
let request_interval = *self
@@ -1487,51 +1297,28 @@ impl RuntimeAccountPoolService {
}
};
let window_persisted = if let Some(reset_at) = reset_at {
match self
if let Some(reset_at) = reset_at {
if let Err(error) = self
.store
.sync_rate_limit_window(account_id, reset_at, limit_window_seconds)
.await
{
Ok(()) => true,
Err(error) => {
tracing::warn!(
account_id,
error = %error,
"failed to persist verified quota window"
);
false
}
tracing::warn!(
account_id,
error = %error,
"failed to persist verified quota window"
);
}
} else {
true
};
}
let (in_memory, transition) = {
let in_memory = {
let mut pool = self.pool.lock().await;
let before = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
let in_memory = pool.apply_quota_state(account_id, limit_reached, cooldown_until);
if let Some(reset_at) = reset_at {
pool.sync_rate_limit_window(account_id, reset_at, limit_window_seconds);
}
let after = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
(
in_memory,
AccountStateTransition::new(
account_id,
AccountStateTransitionReason::QuotaSnapshotApplied,
before,
after,
),
)
in_memory
};
if persisted && window_persisted {
self.record_account_state_transition(transition).await;
}
persisted || in_memory
}
@@ -1580,28 +1367,11 @@ impl RuntimeAccountPoolService {
false
}
};
let (in_memory, transition) = {
let mut pool = self.pool.lock().await;
let before = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
let in_memory = pool.set_cloudflare_cooldown_until(account_id, cooldown_until);
let after = pool
.get(account_id)
.map(|account| RuntimeAccountStateSnapshot::from(&account));
(
in_memory,
AccountStateTransition::new(
account_id,
AccountStateTransitionReason::EdgeCooldownSet,
before,
after,
),
)
};
if persisted {
self.record_account_state_transition(transition).await;
}
let in_memory = self
.pool
.lock()
.await
.set_cloudflare_cooldown_until(account_id, cooldown_until);
persisted || in_memory
}
}
backend/tests/admin/monitoring/usage_record_store.rs
+23
View File
@@ -1,3 +1,4 @@
use chrono::{Duration, Utc};
use codex_proxy_rs::{
admin::monitoring::{
usage_record_model::{UsageRecord, UsageRecordLevel},
@@ -124,6 +125,28 @@ async fn usage_record_store_should_get_and_clear_events() {
assert!(page.items.is_empty());
}
#[tokio::test]
async fn usage_record_store_should_trim_records_older_than_thirty_days() {
let (store, _dir) = usage_record_store("usage-records-retention.sqlite").await;
let now = Utc::now();
let mut expired = UsageRecord::new("v1.response", UsageRecordLevel::Info, "expired");
expired.id = "usage_expired".to_string();
expired.route = Some("/v1/responses".to_string());
expired.created_at = now - Duration::days(31);
store.append(&expired).await.unwrap();
let mut retained = UsageRecord::new("v1.response", UsageRecordLevel::Info, "retained");
retained.id = "usage_retained".to_string();
retained.route = Some("/v1/responses".to_string());
retained.created_at = now - Duration::days(29);
store.append(&retained).await.unwrap();
assert_eq!(store.trim_to_retention(now).await.unwrap(), 1);
assert!(store.get("usage_expired").await.unwrap().is_none());
assert!(store.get("usage_retained").await.unwrap().is_some());
}
async fn usage_record_store(db_name: &str) -> (SqliteUsageRecordStore, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
let db = dir.path().join(db_name);
backend/tests/proxy/dispatch/chat_upstream/mod.rs
+1 -4
View File
@@ -16,9 +16,7 @@ use codex_proxy_rs::{
admin::keys::service::SqliteClientKeyStore,
admin::monitoring::{
usage_record_model::UsageRecordLevel,
usage_record_store::{
SqliteUsageRecordStore, UsageRecordFilter, DEFAULT_USAGE_RECORD_CAPACITY,
},
usage_record_store::{SqliteUsageRecordStore, UsageRecordFilter},
},
config::schema::AppConfig,
http::router,
@@ -388,7 +386,6 @@ async fn test_app_with_account_pool_and_logging_capture_body(
config.logging.enabled = true;
let usage_record_options = UsageRecordOptions {
enabled: true,
capacity: DEFAULT_USAGE_RECORD_CAPACITY,
capture_body: true,
};
let state = test_app_state_with_pool_installation_id_and_usage_record_options(
backend/tests/upstream/accounts/account_pool/selection.rs
+8 -27
View File
@@ -270,22 +270,11 @@ async fn runtime_account_pool_should_persist_quota_window_reset_when_cooldown_ex
.execute(&pool)
.await
.unwrap();
let usage_records = std::sync::Arc::new(
codex_proxy_rs::admin::monitoring::usage_record_service::AdminUsageRecordService::new(
codex_proxy_rs::admin::monitoring::usage_record_store::SqliteUsageRecordStore::new(
pool.clone(),
),
true,
2_000,
false,
),
);
let runtime_pool = codex_proxy_rs::upstream::accounts::pool::RuntimeAccountPoolService::new(
std::sync::Arc::new(store),
AccountPoolOptions::default(),
0,
)
.with_usage_records(usage_records);
);
runtime_pool.restore_from_repository().await.unwrap();
let acquired = runtime_pool
@@ -363,22 +352,14 @@ async fn runtime_account_pool_should_persist_quota_window_reset_when_cooldown_ex
);
assert!(persisted_reset_at.is_some_and(|reset_at| reset_at > now));
assert_eq!(stored_limit_window_seconds, Some(60));
let audit_event: (String, String) = sqlx::query_as(
"select kind, metadata_json from usage_records where account_id = ? order by created_at desc limit 1",
)
.bind("acct_quota_reset")
.fetch_one(&pool)
.await
.unwrap();
let metadata: serde_json::Value = serde_json::from_str(&audit_event.1).unwrap();
let usage_record_count: i64 =
sqlx::query_scalar("select count(*) from usage_records where account_id = ?")
.bind("acct_quota_reset")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(audit_event.0, "account.state");
assert_eq!(metadata["reason"], "runtime_quota_window_refreshed");
assert_eq!(metadata["oldStatus"], "quota_exhausted");
assert_eq!(metadata["newStatus"], "active");
assert_eq!(metadata["syncUsageWindow"], true);
assert_eq!(metadata["after"]["quotaVerifyRequired"], true);
assert_eq!(metadata["after"]["windowRequestCount"], 0);
assert_eq!(usage_record_count, 0);
}
#[test]
frontend/src/views/usage/constants.ts
-1
View File
@@ -111,7 +111,6 @@ export const usageTimeRangeOptions = [
{ label: '今天', value: 'today' },
{ label: '最近 7 天', value: '7d' },
{ label: '最近 30 天', value: '30d' },
{ label: '全部时间', value: 'all' },
]
export const levelLabels: Record<string, string> = {
frontend/src/views/usage/index.vue
-4
View File
@@ -47,10 +47,6 @@ const timeRangeParams = computed<Record<string, string>>(() => {
}
}
if (timeRange.value === 'all') {
return {} as Record<string, string>
}
return {
startTime: now.subtract(6, 'day').startOf('day').toISOString(),
endTime: now.toISOString(),