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

feat(settings): configure request history retention (#435)

Browse Source 
Allow 1–31 day retention and keep published analytics readable during cleanup.

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-15 22:38:53 +08:00
committed by GitHub
co-authored by Zyy Codex
parent 70ab8ebc8d
commit 57c8ef358f
15 changed files with 512 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-admin/src/model/retention.rs
+2 -2
View File
@@ -24,7 +24,7 @@ impl RetentionPolicy {
audit_days: u32,
) -> Result<(), &'static str> {
for (valid, field) in [
(usage_days >= 31, "usage_retention_days"),
((1..=31).contains(&usage_days), "usage_retention_days"),
(ops_days > 0, "ops_event_retention_days"),
(audit_days > 0, "audit_retention_days"),
] {
@@ -38,7 +38,7 @@ impl RetentionPolicy {
pub fn try_new(usage_days: u32, ops_days: u32, audit_days: u32) -> Result<Self, AdminError> {
if Self::validate_values(usage_days, ops_days, audit_days).is_err() {
return Err(AdminError::invalid(
"请求保留期至少 31 天，事件与审计保留期必须为正数",
"请求保留期必须为 1～31 天，事件与审计保留期必须为正数",
));
}
Ok(Self {
backend/crates/gateway-admin/src/use_case/observability/store_cache.rs
+2 -2
View File
@@ -1,4 +1,4 @@
//! 有界统计缓存：身份和完整筛选参与键，删除版本阻止旧数据复活
//! 有界统计缓存：身份和完整筛选参与键，统计发布代次使旧快照失效
use crate::model::observability::*;
use crate::ports::store::{
@@ -121,7 +121,7 @@ impl<K: Eq + Hash, T: AnalyticsValue> Cache<K, T> {
};
// 同键刷新只执行一次；失败不会缓存成成功，也不延长旧结果寿命
let mut cached = slot.lock().await;
// 等锁期间代次可能推进；取得同键锁后再核对，失败时也不能回用已删除事实
// 等锁期间代次可能推进；取得同键锁后再核对，失败时也不能回用已被新统计替代的快照
let generation = store.analytics_cache_generation().await?;
if let Some(entry) = cached.as_mut()
&& entry.value.metadata().generation == generation
backend/crates/gateway-admin/tests/model/retention.rs
+17 -1
View File
@@ -4,7 +4,13 @@ use gateway_admin::model::retention::{RetentionPolicy, RetentionTarget};
#[test]
fn retention_policy_rejects_invalid_windows_and_preserves_independent_windows() {
for (usage, ops, audit) in [(30, 1, 1), (31, 0, 1), (31, 1, 0)] {
for (usage, ops, audit) in [
(0, 1, 1),
(32, 1, 1),
(u32::MAX, 1, 1),
(1, 0, 1),
(1, 1, 0),
] {
assert!(RetentionPolicy::try_new(usage, ops, audit).is_err());
}
let policy = RetentionPolicy::try_new(31, 7, 90).unwrap();
@@ -12,3 +18,13 @@ fn retention_policy_rejects_invalid_windows_and_preserves_independent_windows()
assert_eq!(policy.days(RetentionTarget::OpsEvents), 7);
assert_eq!(policy.days(RetentionTarget::AdminAuditEvents), 90);
}
#[test]
fn request_retention_accepts_short_windows_without_changing_other_targets() {
for days in [1, 3, 7, 15, 30, 31] {
let policy = RetentionPolicy::try_new(days, 30, 90).unwrap();
assert_eq!(policy.days(RetentionTarget::ModelRequests), days);
assert_eq!(policy.days(RetentionTarget::OpsEvents), 30);
assert_eq!(policy.days(RetentionTarget::AdminAuditEvents), 90);
}
}
backend/crates/gateway-api/tests/admin/settings.rs
+53 -3
View File
@@ -62,7 +62,7 @@ fn update_body() -> Value {
"smartScheduling": gateway_core::account::SmartSchedulingConfig::default(),
"minCodexDesktopVersion": "26.825.6671",
"minCodexCliVersion": "0.40.0",
"usageRetentionDays": 32,
"usageRetentionDays": 31,
"opsEventRetentionDays": 31,
"auditRetentionDays": 91,
"accountAutoFreezeEnabled": true,
@@ -216,6 +216,56 @@ fn settings_request_should_reject_unknown_rotation_strategy() {
assert_eq!(request.validate().unwrap_err().field(), "rotationStrategy");
}
#[tokio::test]
async fn short_request_retention_round_trips_and_invalid_days_do_not_replace_it() {
let fixture = AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
let mut body = update_body();
for days in [1, 3, 7, 15, 30, 31] {
body["usageRetentionDays"] = json!(days);
let response = app(fixture.state())
.oneshot(request(
Method::POST,
"/api/admin/settings/update",
Some(body.clone()),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let saved = response_json(response).await;
assert_eq!(saved["data"]["usageRetentionDays"], days);
body["configRevision"] = saved["data"]["configRevision"].clone();
}
for (days, expected_status) in [
(json!(0), StatusCode::BAD_REQUEST),
(json!(32), StatusCode::BAD_REQUEST),
(json!(u32::MAX), StatusCode::BAD_REQUEST),
(json!(-1), StatusCode::UNPROCESSABLE_ENTITY),
(json!(1.5), StatusCode::UNPROCESSABLE_ENTITY),
(json!(null), StatusCode::UNPROCESSABLE_ENTITY),
] {
body["usageRetentionDays"] = days;
let response = app(fixture.state())
.oneshot(request(
Method::POST,
"/api/admin/settings/update",
Some(body.clone()),
))
.await
.unwrap();
assert_eq!(response.status(), expected_status);
}
let response = app(fixture.state())
.oneshot(request(Method::GET, "/api/admin/settings", None))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response_json(response).await["data"]["usageRetentionDays"],
31
);
}
#[test]
fn settings_request_accepts_unlimited_default_account_concurrency() {
let mut body = update_body();
@@ -298,7 +348,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() {
smart_scheduling: gateway_core::account::SmartSchedulingConfig::default(),
min_codex_desktop_version: Some("26.825.6671".to_owned()),
min_codex_cli_version: Some("0.40.0".to_owned()),
usage_retention_days: 32,
usage_retention_days: 31,
ops_event_retention_days: 31,
audit_retention_days: 91,
account_auto_freeze_enabled: true,
@@ -349,7 +399,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() {
"smartSchedulingDefaults": gateway_core::account::SmartSchedulingConfig::default(),
"minCodexDesktopVersion": "26.825.6671",
"minCodexCliVersion": "0.40.0",
"usageRetentionDays": 32,
"usageRetentionDays": 31,
"opsEventRetentionDays": 31,
"auditRetentionDays": 91,
"accountAutoFreezeEnabled": true,
backend/crates/gateway-store/src/postgres/observability/analytics/query.rs
+5 -32
View File
@@ -55,7 +55,7 @@ impl<'a> AnalyticsSnapshot<'a> {
})
}
pub async fn require_visible(&mut self, range: Option<ObservabilityRange>) -> StoreResult<()> {
pub fn require_visible(&self, range: Option<ObservabilityRange>) -> StoreResult<()> {
if self.status.building
&& !range.is_some_and(|range| {
self.status
@@ -68,34 +68,7 @@ impl<'a> AnalyticsSnapshot<'a> {
"request analytics has not backfilled the complete requested range",
));
}
// 未知旧桶与已知范围分开查，避免通用计划为 OR 选择扫描全部删除积压
// 时间移动和未 ACK 的投影会使队列旧桶偏离当前事实，删除可见性必须同时检查事实所在范围
let deletion_pending: bool = if let Some(range) = range {
let lower = range
.start
.checked_sub_signed(TimeDelta::minutes(15))
.ok_or_else(|| invalid("analytics deletion boundary overflow"))?;
let (intersects, any): (bool, bool) = sqlx::query_as("select exists(select 1 from request_analytics_pending_requests where deletion_pending and previous_bucket is null) or exists(select 1 from request_analytics_pending_requests where deletion_pending and previous_bucket > $1 and previous_bucket < $2), exists(select 1 from request_analytics_pending_requests where deletion_pending)")
.bind(lower).bind(range.end).fetch_one(&mut *self.tx).await
.map_err(|error|postgres_unavailable("check analytics deletion visibility",error))?;
if intersects || !any {
intersects
} else {
// 没有待删版本时不引用 facts，使完整干净桶保持只读汇总的路径
sqlx::query_scalar("select exists(select 1 from request_analytics_pending_requests p join request_analytics_facts f on f.id=p.request_id where p.deletion_pending and f.started_at >= $1 and f.started_at < $2)")
.bind(range.start).bind(range.end).fetch_one(&mut *self.tx).await
.map_err(|error|postgres_unavailable("check projected analytics deletion visibility",error))?
}
} else {
sqlx::query_scalar("select exists(select 1 from request_analytics_pending_requests where deletion_pending)")
.fetch_one(&mut *self.tx).await
.map_err(|error|postgres_unavailable("check analytics deletion visibility",error))?
};
if deletion_pending {
return Err(unavailable(
"request analytics is removing deleted source contributions in this range; retry after projection",
));
}
// 源清理尚未投影时继续读取已发布事实，Worker 在同一事务撤销事实与汇总贡献
Ok(())
}
@@ -133,7 +106,7 @@ impl<'a> AnalyticsSnapshot<'a> {
}
pub async fn totals(&mut self) -> StoreResult<DashboardTotals> {
self.require_visible(None).await?;
self.require_visible(None)?;
let bounds=sqlx::query("select (select started_at from request_analytics_facts order by started_at,id limit 1) as first_started_at,(select started_at from request_analytics_facts order by started_at desc,id desc limit 1) as last_started_at")
.fetch_one(&mut *self.tx).await.map_err(|error|postgres_unavailable("read retained analytics bounds",error))?;
let start =
@@ -440,7 +413,7 @@ impl<'a> AnalyticsSnapshot<'a> {
limit: u16,
) -> StoreResult<(u64, Vec<String>)> {
filter.validate()?;
self.require_visible(Some(range)).await?;
self.require_visible(Some(range))?;
let (collection, column, reference) = if dimension == DiagnosticDimension::Account {
(
"account",
@@ -636,7 +609,7 @@ impl<'a> AnalyticsSnapshot<'a> {
distribution_names,
} = selection;
filter.validate()?;
self.require_visible(Some(range)).await?;
self.require_visible(Some(range))?;
let selected = collection_for(filter, dimension);
let boundaries = if granularity == Some(ObservationGranularity::Day) {
Some(calendar_buckets(range, timezone)?)
backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs
+1 -1
View File
@@ -102,7 +102,7 @@ impl PgRequestAnalyticsRepository {
// ACK 在投影提交后使用 READ COMMITTED，行锁按最新提交版本重新比较，已改写的版本不再匹配
let mut transaction = connection.begin().await.map_err(|error| postgres_unavailable("begin analytics acknowledgement", error))?;
sqlx::query("set local statement_timeout = '10s'").execute(&mut *transaction).await.map_err(|error| postgres_unavailable("set analytics query deadline", error))?;
// 源删除会持有队列行再修改 generation，ACK 不能等待这些源事务形成反向锁依赖
// ACK 跳过源事务持有的队列行，避免等待并发更新或删除
// 被跳过的版本保留在持久队列中重放，只有精确匹配且已取得行锁的版本才确认
sqlx::query("with acknowledged as materialized (select pending.request_id,pending.change_version from request_analytics_pending_requests pending join unnest($1::text[],$2::bigint[]) job(id,version) on pending.request_id=job.id and pending.change_version=job.version for update of pending skip locked) delete from request_analytics_pending_requests pending using acknowledged where pending.request_id=acknowledged.request_id and pending.change_version=acknowledged.change_version")
.bind(ids).bind(versions).execute(&mut *transaction).await.map_err(|error| postgres_unavailable("confirm analytics request versions", error))?;
backend/crates/gateway-store/tests/postgres/analytics.rs
+48 -49
View File
@@ -300,8 +300,7 @@ async fn projection_replay_late_ops_and_deletion_preserve_exact_contributions()
let with_ops = repo.usage_summary(range, Default::default()).await.unwrap();
assert_eq!(with_ops.attempts.failure_count, 1);
assert_eq!(with_ops.attempts.rate_limited_count, 1);
sqlx::query("select enqueue_request_analytics('req_analytics_a',$1)")
.bind(start)
sqlx::query("select enqueue_request_analytics('req_analytics_a')")
.execute(&database.pool)
.await
.unwrap();
@@ -317,9 +316,12 @@ async fn projection_replay_late_ops_and_deletion_preserve_exact_contributions()
.execute(&database.pool)
.await
.unwrap();
assert!(
repo.usage_summary(range, Default::default()).await.is_err(),
"deleted contributions cannot be returned as a new generation"
let pending = repo.usage_summary(range, Default::default()).await.unwrap();
assert_eq!(pending.requests.request_count, 1);
assert_eq!(pending.attempts.failure_count, 1);
assert_eq!(
pending.analytics.state,
gateway_admin::model::observability::AnalyticsState::Lagging
);
drain_analytics(&database.pool).await;
let deleted = repo.usage_summary(range, Default::default()).await.unwrap();
@@ -450,10 +452,13 @@ async fn historical_deletion_does_not_block_disjoint_recent_statistics() {
1
);
let history = ObservabilityRange::new(now - TimeDelta::days(30), now).unwrap();
assert!(
assert_eq!(
repo.usage_summary(history, Default::default())
.await
.is_err()
.unwrap()
.requests
.request_count,
2
);
drain_analytics(&database.pool).await;
assert_eq!(
@@ -989,7 +994,7 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero()
}
#[tokio::test]
async fn source_deletions_do_not_deadlock_with_worker_publication() {
async fn source_deletions_do_not_block_worker_publication() {
use sqlx::Connection;
for backfill in [false, true] {
let Some(database) = TestDatabase::create("analytics_delete_publication").await else {
@@ -1010,26 +1015,17 @@ async fn source_deletions_do_not_deadlock_with_worker_publication() {
let mut source = sqlx::PgConnection::connect_with(&database.pool.connect_options())
.await
.unwrap();
let source_pid: i32 = sqlx::query_scalar("select pg_backend_pid()")
.fetch_one(&mut source)
.await
.unwrap();
sqlx::raw_sql("begin;delete from model_requests where id='delete_a'")
.execute(&mut source)
.await
.unwrap();
let worker =
PgRequestAnalyticsRepository::new(database.pool.clone(), observability_query_budget());
let running = tokio::spawn(async move { worker.maintain(10).await });
// 源事务先持有 generation，确认真实 Worker 到达发布或回填游标的锁等待
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let waiting: bool = sqlx::query_scalar("select exists(select 1 from pg_stat_activity where wait_event_type='Lock' and $1=any(pg_blocking_pids(pid)) and query like 'update request_analytics_build_state%')")
.bind(source_pid).fetch_one(&database.pool).await.unwrap();
if waiting { break; }
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
}).await.expect("worker reached build-state publication");
// 源删除只登记待处理版本，不持有统计发布行，Worker 可以独立完成发布
tokio::time::timeout(std::time::Duration::from_secs(2), worker.maintain(10))
.await
.expect("source deletion must not block analytics publication")
.unwrap();
tokio::time::timeout(
std::time::Duration::from_secs(2),
sqlx::query("delete from model_requests where id='delete_b'").execute(&mut source),
@@ -1038,8 +1034,6 @@ async fn source_deletions_do_not_deadlock_with_worker_publication() {
.expect("source deletion cannot wait for worker ACK")
.expect("source deletion cannot be a deadlock victim");
sqlx::query("commit").execute(&mut source).await.unwrap();
// RR 事务允许因源提交而重试，但源请求删除不能承受死锁回滚
let _ = running.await.unwrap();
drain_analytics(&database.pool).await;
let remaining: i64 = sqlx::query_scalar("select count(*) from request_analytics_facts")
.fetch_one(&database.pool)
@@ -1087,10 +1081,9 @@ async fn worker_skips_locked_versions_and_preserves_a_concurrent_source_deletion
.execute(&mut source)
.await
.unwrap();
let pending: (i64, bool) = sqlx::query_as("select change_version,deletion_pending from request_analytics_pending_requests where request_id='locked_ack'")
let pending: i64 = sqlx::query_scalar("select change_version from request_analytics_pending_requests where request_id='locked_ack'")
.fetch_one(&database.pool).await.expect("new source version cannot be acknowledged by old projection");
assert!(pending.0 > new_version);
assert!(pending.1);
assert!(pending > new_version);
drain_analytics(&database.pool).await;
assert_eq!(worker.status().await.unwrap().pending_requests, 0);
let remaining: i64 = sqlx::query_scalar("select count(*) from request_analytics_facts")
@@ -1182,7 +1175,7 @@ async fn committed_queue_versions_skip_old_ack_and_replay_without_losing_contrib
}
#[tokio::test]
async fn parent_cascades_invalidate_once_and_explicit_ops_mutations_keep_their_contribution() {
async fn parent_cascades_and_ops_deletions_publish_statistics_asynchronously() {
let Some(database) = TestDatabase::create("analytics_cascade_generation").await else {
return;
};
@@ -1206,18 +1199,20 @@ async fn parent_cascades_invalidate_once_and_explicit_ops_mutations_keep_their_c
.await
.unwrap();
assert_eq!(
after,
before + 1,
"the parent statement owns cascade invalidation"
after, before,
"source cleanup keeps the published generation"
);
drain_analytics(&database.pool).await;
sqlx::query("update ops_events set failure_kind='authentication',status_code=401 where id='event_ops_survivor'")
.execute(&database.pool).await.unwrap();
drain_analytics(&database.pool).await;
let repo = observability_repository(&database.pool);
let range =
ObservabilityRange::new(start - TimeDelta::minutes(1), start + TimeDelta::minutes(5))
.unwrap();
let published = repo.usage_summary(range, Default::default()).await.unwrap();
assert_eq!(published.requests.request_count, 3);
assert_eq!(published.attempts.failure_count, 3);
drain_analytics(&database.pool).await;
sqlx::query("update ops_events set failure_kind='authentication',status_code=401 where id='event_ops_survivor'")
.execute(&database.pool).await.unwrap();
drain_analytics(&database.pool).await;
let changed = repo.usage_summary(range, Default::default()).await.unwrap();
assert_eq!(changed.attempts.auth_failure_count, 1);
assert_eq!(changed.attempts.rate_limited_count, 0);
@@ -1233,8 +1228,10 @@ async fn parent_cascades_invalidate_once_and_explicit_ops_mutations_keep_their_c
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(after, before + 1);
assert!(repo.usage_summary(range, Default::default()).await.is_err());
assert_eq!(after, before);
let pending = repo.usage_summary(range, Default::default()).await.unwrap();
assert_eq!(pending.requests.request_count, 1);
assert_eq!(pending.attempts.auth_failure_count, 1);
drain_analytics(&database.pool).await;
let deleted = repo.usage_summary(range, Default::default()).await.unwrap();
assert_eq!(deleted.requests.request_count, 1);
@@ -1243,16 +1240,16 @@ async fn parent_cascades_invalidate_once_and_explicit_ops_mutations_keep_their_c
}
#[tokio::test]
async fn source_move_then_delete_hides_the_still_projected_old_bucket() {
assert_moved_deletion_is_hidden(false).await;
async fn source_move_then_delete_keeps_the_published_bucket_until_projection() {
assert_moved_deletion_converges(false).await;
}
#[tokio::test]
async fn unacknowledged_move_then_delete_hides_the_current_fact_bucket() {
assert_moved_deletion_is_hidden(true).await;
async fn unacknowledged_move_then_delete_keeps_the_published_bucket_until_projection() {
assert_moved_deletion_converges(true).await;
}
async fn assert_moved_deletion_is_hidden(project_without_ack: bool) {
async fn assert_moved_deletion_converges(project_without_ack: bool) {
let Some(database) = TestDatabase::create("analytics_moved_delete").await else {
return;
};
@@ -1269,7 +1266,7 @@ async fn assert_moved_deletion_is_hidden(project_without_ack: bool) {
.await;
drain_analytics(&database.pool).await;
if project_without_ack {
// 第一条待处理版本记录 A，随后源时间移到 B，队列仍保留最早的桶
// 连续源变更仍只占一个队列位置，ACK 必须比较精确版本
sqlx::query("update model_requests set total_tokens=240 where id='moved_delete'")
.execute(&database.pool)
.await
@@ -1295,11 +1292,8 @@ async fn assert_moved_deletion_is_hidden(project_without_ack: bool) {
.fetch_one(&database.pool)
.await
.unwrap();
let previous: DateTime<Utc> = sqlx::query_scalar("select previous_bucket from request_analytics_pending_requests where request_id='moved_delete'")
.fetch_one(&database.pool).await.unwrap();
let visible_bucket = if project_without_ack { moved } else { old };
assert_eq!(projected, visible_bucket + TimeDelta::minutes(1));
assert_eq!(previous, if project_without_ack { old } else { moved });
sqlx::query("delete from model_requests where id='moved_delete'")
.execute(&database.pool)
.await
@@ -1307,9 +1301,14 @@ async fn assert_moved_deletion_is_hidden(project_without_ack: bool) {
let repo = observability_repository(&database.pool);
let range =
ObservabilityRange::new(visible_bucket, visible_bucket + TimeDelta::minutes(15)).unwrap();
assert!(
repo.usage_summary(range, Default::default()).await.is_err(),
"a source deletion must hide the actually projected bucket even when the queue remembers another bucket"
assert_eq!(
repo.usage_summary(range, Default::default())
.await
.unwrap()
.requests
.request_count,
1,
"published facts stay readable until their deletion is projected"
);
drain_analytics(&database.pool).await;
for start in [old, moved] {
backend/crates/gateway-store/tests/postgres/analytics_buckets.rs
+1 -2
View File
@@ -295,8 +295,7 @@ async fn incremental_deletions_repair_extrema_last_usage_and_zero_currency_prese
.await
.unwrap();
assert_eq!(costs, serde_json::json!({}));
sqlx::query("select enqueue_request_analytics('extrema_3',$1)")
.bind(hour() + TimeDelta::minutes(3))
sqlx::query("select enqueue_request_analytics('extrema_3')")
.execute(&database.pool)
.await
.unwrap();
backend/crates/gateway-store/tests/postgres/retention.rs
+282 -3
View File
@@ -1,4 +1,4 @@
//! 验证历史清理仅按批次上限删除已过期记录
//! 验证短保留期升级、分批清理边界与统计异步收敛期间的查询可用性
use std::num::NonZeroU32;
@@ -7,9 +7,288 @@ use gateway_admin::{
model::retention::{RetentionPolicy, RetentionTarget},
ports::retention::RetentionStore as _,
};
use gateway_store::postgres::PgRetentionRepository;
use gateway_store::postgres::{
ObservabilityRange, PgRequestAnalyticsRepository, PgRetentionRepository,
};
use super::TestDatabase;
use super::{
TEST_MIGRATOR, TestDatabase, admin_observability_store, analytics::seed, drain_analytics,
observability_query_budget, observability_repository,
};
#[tokio::test]
async fn retention_upgrade_caps_request_window_and_preserves_other_settings_and_pending_work() {
let Some(database) = TestDatabase::create_through("short_retention_upgrade", 31).await else {
return;
};
sqlx::query("update runtime_settings set usage_retention_days=60,ops_event_retention_days=7,audit_retention_days=45")
.execute(&database.pool)
.await
.unwrap();
let start = Utc::now() - ChronoDuration::days(10);
seed(&database.pool, "retention_upgrade", start, &[]).await;
drain_analytics(&database.pool).await;
sqlx::query("delete from model_requests where id='retention_upgrade'")
.execute(&database.pool)
.await
.unwrap();
let before: (String, i64, chrono::DateTime<Utc>) = sqlx::query_as(
"select request_id,change_version,oldest_pending_at from request_analytics_pending_requests",
)
.fetch_one(&database.pool)
.await
.unwrap();
TEST_MIGRATOR.run(&database.pool).await.unwrap();
let repository = PgRetentionRepository::new(database.pool.clone());
assert_eq!(
repository
.load_policy()
.await
.unwrap()
.days(RetentionTarget::ModelRequests),
31
);
let policy = repository.load_policy().await.unwrap();
assert_eq!(policy.days(RetentionTarget::OpsEvents), 7);
assert_eq!(policy.days(RetentionTarget::AdminAuditEvents), 45);
let after: (String, i64, chrono::DateTime<Utc>) = sqlx::query_as(
"select request_id,change_version,oldest_pending_at from request_analytics_pending_requests",
)
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(after, before);
// 升级时遗留的删除任务继续可读并最终收敛，不重建全量统计
let range = ObservabilityRange::new(start, start + ChronoDuration::hours(1)).unwrap();
let statistics = observability_repository(&database.pool);
assert_eq!(
statistics
.usage_summary(range, Default::default())
.await
.unwrap()
.requests
.request_count,
1
);
drain_analytics(&database.pool).await;
assert_eq!(
statistics
.usage_summary(range, Default::default())
.await
.unwrap()
.requests
.request_count,
0
);
for days in [1_i64, 3, 7, 15, 30, 31] {
sqlx::query("update runtime_settings set usage_retention_days=$1")
.bind(days)
.execute(&database.pool)
.await
.unwrap();
assert_eq!(
i64::from(
repository
.load_policy()
.await
.unwrap()
.days(RetentionTarget::ModelRequests)
),
days
);
}
for (usage, ops, audit) in [
(0_i64, 30_i64, 90_i64),
(32, 30, 90),
(1, 0, 90),
(1, 30, 0),
] {
let error = sqlx::query("update runtime_settings set usage_retention_days=$1,ops_event_retention_days=$2,audit_retention_days=$3")
.bind(usage).bind(ops).bind(audit).execute(&database.pool).await.unwrap_err();
assert_eq!(
error.as_database_error().unwrap().code().as_deref(),
Some("23514")
);
}
sqlx::query("update runtime_settings set usage_retention_days=default")
.execute(&database.pool)
.await
.unwrap();
assert_eq!(
repository
.load_policy()
.await
.unwrap()
.days(RetentionTarget::ModelRequests),
31
);
database.close().await;
}
#[tokio::test]
async fn short_retention_keeps_statistics_readable_until_each_deletion_batch_is_projected() {
use gateway_admin::{
model::observability::{AnalyticsState, DashboardQuery, TimeRange},
ports::store::ObservabilityStore as _,
};
let Some(database) = TestDatabase::create("short_retention_readable").await else {
return;
};
let now = "2026-10-15T12:00:00Z"
.parse::<chrono::DateTime<Utc>>()
.unwrap();
for id in ["expired_a", "expired_b", "expired_c"] {
seed(&database.pool, id, now - ChronoDuration::days(8), &[]).await;
sqlx::query("insert into ops_events(id,model_request_id,attempt_index,level,component,operation,failure_kind,status_code,message,created_at) values($1,$2,1,'warning','routing','responses','rate_limited',429,'synthetic',$3)")
.bind(format!("event_{id}")).bind(id).bind(now - ChronoDuration::days(8))
.execute(&database.pool).await.unwrap();
}
seed(
&database.pool,
"boundary",
now - ChronoDuration::days(7) - ChronoDuration::minutes(1),
&[],
)
.await;
seed(
&database.pool,
"recent",
now - ChronoDuration::hours(1),
&[],
)
.await;
seed(
&database.pool,
"running",
now - ChronoDuration::days(40),
&[],
)
.await;
sqlx::query("update model_requests set outcome='running',completed_at=null,client_status_code=null,downstream_committed_at=null where id='running'")
.execute(&database.pool).await.unwrap();
sqlx::query("update runtime_settings set usage_retention_days=7")
.execute(&database.pool)
.await
.unwrap();
drain_analytics(&database.pool).await;
let range = ObservabilityRange::new(now - ChronoDuration::days(41), now).unwrap();
let statistics = observability_repository(&database.pool);
let dashboard = admin_observability_store(&database.pool);
let dashboard_query = DashboardQuery::new(TimeRange {
start: now - ChronoDuration::days(1),
end: now,
});
let original = statistics
.usage_summary(range, Default::default())
.await
.unwrap();
assert_eq!(original.requests.request_count, 6);
let expired_day_start = (now - ChronoDuration::days(8))
.date_naive()
.and_hms_opt(0, 0, 0)
.unwrap()
.and_utc();
// 同时验证发布前确实存在各层统计，避免清理断言因未生成汇总而空通过
const EXPIRED_ANALYTICS_ROWS: &str = "select
(select count(*) from request_analytics_facts where id in ('expired_a','expired_b','expired_c')),
(select count(*) from request_analytics_15m where bucket_start >= $1 and bucket_start < $2),
(select count(*) from request_analytics_hourly where bucket_start >= $1 and bucket_start < $2),
(select count(*) from request_analytics_daily_counts where bucket_start >= $1 and bucket_start < $2)";
let before: (i64, i64, i64, i64) = sqlx::query_as(EXPIRED_ANALYTICS_ROWS)
.bind(expired_day_start)
.bind(expired_day_start + ChronoDuration::days(1))
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(before.0, 3);
assert!(before.1 > 0 && before.2 > 0 && before.3 > 0);
let repository = PgRetentionRepository::new(database.pool.clone());
let policy = repository.load_policy().await.unwrap();
for expected_deleted in [2, 1, 0] {
let deleted = repository
.purge_batch(
RetentionTarget::ModelRequests,
now,
policy,
NonZeroU32::new(2).unwrap(),
)
.await
.unwrap();
assert_eq!(deleted, expected_deleted);
let published = statistics
.usage_summary(range, Default::default())
.await
.unwrap();
assert_eq!(published.requests.request_count, 6);
assert_eq!(published.attempts.failure_count, 3);
assert_eq!(
published.analytics.generation,
original.analytics.generation
);
assert_eq!(published.analytics.state, AnalyticsState::Lagging);
let snapshot = dashboard
.dashboard_summary(dashboard_query.clone(), now)
.await
.unwrap();
assert_eq!(snapshot.totals.request_count, 6);
}
let retained: Vec<String> = sqlx::query_scalar("select id from model_requests order by id")
.fetch_all(&database.pool)
.await
.unwrap();
assert_eq!(retained, ["boundary", "recent", "running"]);
let events: i64 = sqlx::query_scalar("select count(*) from ops_events")
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(events, 0);
let worker =
PgRequestAnalyticsRepository::new(database.pool.clone(), observability_query_budget());
for remaining_expired in [2, 1, 0] {
worker.maintain(1).await.unwrap();
let published = statistics
.usage_summary(range, Default::default())
.await
.unwrap();
assert_eq!(published.requests.request_count, 3 + remaining_expired);
assert_eq!(published.attempts.failure_count, remaining_expired);
assert_eq!(
published.requests.total_tokens,
120 * (2 + remaining_expired)
);
assert!(published.analytics.generation > original.analytics.generation);
let snapshot = dashboard
.dashboard_summary(dashboard_query.clone(), now)
.await
.unwrap();
assert_eq!(snapshot.totals.request_count, 3 + remaining_expired);
}
let status = worker.status().await.unwrap();
assert_eq!(status.pending_requests, 0);
assert_eq!(status.pending_buckets, 0);
let after: (i64, i64, i64, i64) = sqlx::query_as(EXPIRED_ANALYTICS_ROWS)
.bind(expired_day_start)
.bind(expired_day_start + ChronoDuration::days(1))
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(after, (0, 0, 0, 0));
assert_eq!(
statistics
.usage_summary(range, Default::default())
.await
.unwrap()
.analytics
.state,
AnalyticsState::Ready
);
database.close().await;
}
#[tokio::test]
async fn retention_batch_deletes_only_expired_rows_up_to_limit() {
backend/migrations/.frozen-sha256
+1
View File
@@ -29,3 +29,4 @@ d4b599066d7382a1c9cb6b3c003200046579e4fce3531f232cbd665a9d92c542 0026_codex_pri
1923f14095dac5f808a530db002c37e17168afa0778bc7f3da768c818133eec9 0029_usage_indexes_and_identity_snapshots.sql
f10058e2b4c974b6d2ffa11be78b085504a233205253948c03357ef6fe234bca 0030_analytics_and_scheduling.sql
b49832e205405de7ef078f3b65c0c19f2c030b8984d42411598738b6b6c146bf 0031_proxy_and_privacy.sql
8eade5bd4f5c8847366ca951552775605ecf4171a7f0e87fa18574bceb8a3ff1 0032_request_history_retention.sql
backend/migrations/0032_request_history_retention.sql
+54
View File
@@ -0,0 +1,54 @@
-- 请求明细与统计共用 1～31 天保留期，已有超限设置收敛到 31 天
alter table runtime_settings drop constraint runtime_settings_retention_ck;
alter table runtime_settings alter column usage_retention_days set default 31;
update runtime_settings set usage_retention_days = 31 where usage_retention_days > 31;
alter table runtime_settings add constraint runtime_settings_retention_ck check (
usage_retention_days between 1 and 31
and ops_event_retention_days > 0
and audit_retention_days > 0
);
-- 删除与新增、修改共用异步投影，读取已发布统计直到 Worker 原子撤销旧贡献
-- 缓存代次只在统计发布时推进，源清理不使仍然一致的统计快照失效
drop trigger request_analytics_delete_invalidation on model_requests;
drop trigger request_analytics_ops_delete_invalidation on ops_events;
drop function invalidate_deleted_request_analytics();
drop function invalidate_deleted_ops_analytics();
create function enqueue_request_analytics(request_ref text)
returns void language sql as $$
insert into request_analytics_pending_requests(request_id)
select request_ref where request_ref is not null
on conflict (request_id) do update
set change_version = nextval('request_analytics_change_version')
$$;
create or replace function observe_request_analytics_change() returns trigger language plpgsql as $$
begin
if TG_OP = 'DELETE' then
perform enqueue_request_analytics(OLD.id);
return OLD;
end if;
perform enqueue_request_analytics(NEW.id);
return NEW;
end
$$;
create or replace function observe_ops_analytics_change() returns trigger language plpgsql as $$
begin
if TG_OP <> 'INSERT' then
perform enqueue_request_analytics(OLD.model_request_id);
end if;
if TG_OP <> 'DELETE' then
perform enqueue_request_analytics(NEW.model_request_id);
return NEW;
end if;
return OLD;
end
$$;
-- 旧时间桶由已发布事实定位，不再维护只用于阻断读取的删除标记与索引
drop function enqueue_request_analytics(text, timestamptz, boolean);
alter table request_analytics_pending_requests
drop column previous_bucket,
drop column deletion_pending;
docs/api.md
+12 -1
View File
@@ -1405,10 +1405,21 @@ HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly
| [调度策略](#智能调度) | `rotationStrategy`、`smartScheduling` |
| [请求解压](#请求解压上限) | `responsesMaxDecompressedBodyBytes` |
| [客户端门禁](#1-鉴权与公共约定) | `minCodexDesktopVersion`、`minCodexCliVersion` |
| 历史保留 | `usageRetentionDays`、`opsEventRetentionDays`、`auditRetentionDays` |
| [历史保留](#历史保留) | `usageRetentionDays`、`opsEventRetentionDays`、`auditRetentionDays` |
| [账号自动冻结](#账号自动冻结) | `accountAutoFreezeEnabled`、`accountAutoFreezeThreshold`、`accountAutoFreezeWindowSeconds`、`accountAutoFreezeDurationSeconds`、`accountAutoFreezeProbeEnabled`、`accountAutoFreezeProbeModel`、`accountAutoFreezeAdaptiveConcurrency` |
| [账号预热](#账号预热) | `accountWarmupEnabled`、`accountWarmupScheduleTime`、`accountWarmupModel` |
### 历史保留
`usageRetentionDays` 控制请求明细及其用量统计的保留天数，默认 31，支持 1～31 的整数天数。
`opsEventRetentionDays` 与 `auditRetentionDays` 分别控制独立运维事件和管理审计，默认 30、90，均须为正整数
设置保存后由每小时运行的清理任务读取，按完成时间分批删除过期请求及关联事件，不删除运行中的请求。
缩短保留期后，已有过期数据逐批清理，清理结果不可恢复；Client Key 结算账本不受影响
统计在后台异步更新，清理期间可短暂包含已删除请求的统计贡献，投影完成后更新为剩余数据。
查询范围无需随保留期缩短，超出保留期的已清理数据不再计入结果
### 账号预热
定时账号预热默认关闭。`accountWarmupScheduleTime` 使用部署时区中的 `HH:MM`，
frontend/src/views/settings/components/RequestHistoryCard.vue
+19
View File
@@ -0,0 +1,19 @@
<script setup lang="ts">
import { ZCard, ZForm, ZFormItem, ZInput } from '@codex-proxy/ui'
const model = defineModel<string>({ required: true })
</script>
<template>
<ZCard title="历史保留" description="自动清理过期请求及对应统计">
<ZForm class="max-w-6xl sm:grid-cols-2 sm:items-end">
<ZFormItem label="保留天数" description="至少 1 天，过期数据自动分批清理，清理后无法恢复">
<ZInput v-model="model" aria-label="历史保留天数" type="number" min="1" max="31" step="1" placeholder="请输入 1–31 天">
<template #suffix>
天
</template>
</ZInput>
</ZFormItem>
</ZForm>
</ZCard>
</template>
frontend/src/views/settings/composables/useSettingsForm.ts
+10 -3
View File
@@ -44,7 +44,7 @@ export function useSettingsForm() {
rotationStrategy: '' as RotationStrategy | '',
minCodexDesktopVersion: '',
minCodexCliVersion: '',
usageRetentionDays: 31,
usageRetentionDays: 31 as number | null,
opsEventRetentionDays: 30,
auditRetentionDays: 90,
@@ -76,7 +76,7 @@ export function useSettingsForm() {
mappings.value = initial.mappings
}
function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'maxAccountRotations' | 'openaiSessionAffinityTtlHours' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') {
function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'openaiGuardianReservedConcurrency' | 'maxAccountRotations' | 'openaiSessionAffinityTtlHours' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'usageRetentionDays' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') {
return computed({
get: () => (form[key] === null ? '' : String(form[key])),
set: (value: string) => {
@@ -99,6 +99,7 @@ export function useSettingsForm() {
const maxWaitingPerKeyValue = numericModel('maxWaitingPerKey')
const maxWaitingPerAccountValue = numericModel('maxWaitingPerAccount')
const responsesMaxDecompressedBodyMiBValue = numericModel('responsesMaxDecompressedBodyMiB')
const usageRetentionDaysValue = numericModel('usageRetentionDays')
const concurrencyWaitTimeoutSecondsValue = numericModel('concurrencyWaitTimeoutSeconds')
const accountAutoFreezeThresholdValue = numericModel('accountAutoFreezeThreshold')
const accountAutoFreezeWindowSecondsValue = numericModel('accountAutoFreezeWindowSeconds')
@@ -234,6 +235,11 @@ export function useSettingsForm() {
ZNotification.warning({ message: 'Responses 解压上限应为有效的正整数（MiB）' })
return
}
const usageRetentionDays = form.usageRetentionDays
if (usageRetentionDays === null || !Number.isInteger(usageRetentionDays) || usageRetentionDays < 1 || usageRetentionDays > 31) {
ZNotification.warning({ message: '历史保留天数应为 1–31 的整数' })
return
}
if (![maxWaitingPerKey, maxWaitingPerAccount].every(value => Number.isInteger(value) && value >= 0 && value <= 1000)
|| !Number.isInteger(concurrencyWaitTimeoutSeconds) || concurrencyWaitTimeoutSeconds < 1 || concurrencyWaitTimeoutSeconds > 120) {
ZNotification.warning({ message: '队列容量应为 0～1000 的整数，排队超时应为 1～120 秒的整数' })
@@ -308,7 +314,7 @@ export function useSettingsForm() {
smartScheduling: { ...smartScheduling },
minCodexDesktopVersion: form.minCodexDesktopVersion.trim() || null,
minCodexCliVersion: form.minCodexCliVersion.trim() || null,
usageRetentionDays: form.usageRetentionDays,
usageRetentionDays,
opsEventRetentionDays: form.opsEventRetentionDays,
auditRetentionDays: form.auditRetentionDays,
accountAutoFreezeEnabled: form.accountAutoFreezeEnabled,
@@ -354,6 +360,7 @@ export function useSettingsForm() {
maxWaitingPerAccountValue,
concurrencyWaitTimeoutSecondsValue,
responsesMaxDecompressedBodyMiBValue,
usageRetentionDaysValue,
accountAutoFreezeThresholdValue,
accountAutoFreezeWindowSecondsValue,
accountAutoFreezeDurationSecondsValue,
frontend/src/views/settings/index.vue
+5 -2
View File
@@ -13,6 +13,7 @@ import ClientProfileCard from './components/ClientProfileCard.vue'
import ConcurrencyPolicyCard from './components/ConcurrencyPolicyCard.vue'
import ModelAliasesCard from './components/ModelAliasesCard.vue'
import PrivacyPolicyCard from './components/privacy/PrivacyPolicyCard.vue'
import RequestHistoryCard from './components/RequestHistoryCard.vue'
import RequestLocationCard from './components/RequestLocationCard.vue'
import RotationStrategyCard from './components/RotationStrategyCard.vue'
import SettingsAccessSection from './components/SettingsAccessSection.vue'
@@ -24,7 +25,7 @@ import PricingSection from './pricing/index.vue'
const route = useRoute()
const router = useRouter()
const sectionOptions = [
{ label: '网关调度', value: 'runtime' },
{ label: '运行设置', value: 'runtime' },
{ label: '上游配置', value: 'upstream' },
{ label: '模型定价', value: 'pricing' },
{ label: '安全与访问', value: 'access' },
@@ -69,6 +70,7 @@ const {
maxWaitingPerAccountValue,
concurrencyWaitTimeoutSecondsValue,
responsesMaxDecompressedBodyMiBValue,
usageRetentionDaysValue,
accountAutoFreezeThresholdValue,
accountAutoFreezeWindowSecondsValue,
accountAutoFreezeDurationSecondsValue,
@@ -92,7 +94,7 @@ watch(section, (value) => {
<template>
<div class="w-full" :class="section === 'pricing' ? 'flex h-[calc(100dvh-2rem)] flex-none! flex-col min-[961px]:h-[calc(100dvh-3rem)]' : undefined">
<ZPageHeader title="系统设置" description="管理网关调度、上游配置、模型定价、安全访问与数据备份" />
<ZPageHeader title="系统设置" description="配置网关调度、上游与定价，管理访问安全、历史保留与备份" />
<div class="mt-4 flex min-h-cp-control shrink-0 flex-wrap items-center justify-between gap-3">
<ZSegmented
@@ -181,6 +183,7 @@ watch(section, (value) => {
v-model:schedule-time="form.accountWarmupScheduleTime"
v-model:model="form.accountWarmupModel"
/>
<RequestHistoryCard v-model="usageRetentionDaysValue" />
</template>
<div v-if="visited.has('upstream')" v-show="section === 'upstream'" class="grid min-w-0 gap-5">