From 57c8ef358f20d09538d4071c69d35d32fbe8105e Mon Sep 17 00:00:00 2001 From: Zyy <0x@ainz.cc> Date: Thu, 15 Oct 2026 22:38:53 +0800 Subject: [PATCH] feat(settings): configure request history retention (#435) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Allow 1–31 day retention and keep published analytics readable during cleanup. Co-authored-by: Zyy Co-authored-by: Codex --- .../gateway-admin/src/model/retention.rs | 4 +- .../src/use_case/observability/store_cache.rs | 4 +- .../gateway-admin/tests/model/retention.rs | 18 +- .../gateway-api/tests/admin/settings.rs | 56 +++- .../postgres/observability/analytics/query.rs | 37 +-- .../observability/analytics/worker.rs | 2 +- .../gateway-store/tests/postgres/analytics.rs | 97 +++--- .../tests/postgres/analytics_buckets.rs | 3 +- .../gateway-store/tests/postgres/retention.rs | 285 +++++++++++++++++- backend/migrations/.frozen-sha256 | 1 + .../0032_request_history_retention.sql | 54 ++++ docs/api.md | 13 +- .../components/RequestHistoryCard.vue | 19 ++ .../settings/composables/useSettingsForm.ts | 13 +- frontend/src/views/settings/index.vue | 7 +- 15 files changed, 512 insertions(+), 101 deletions(-) create mode 100644 backend/migrations/0032_request_history_retention.sql create mode 100644 frontend/src/views/settings/components/RequestHistoryCard.vue diff --git a/backend/crates/gateway-admin/src/model/retention.rs b/backend/crates/gateway-admin/src/model/retention.rs index d89664260..c7ffd16e1 100644 --- a/backend/crates/gateway-admin/src/model/retention.rs +++ b/backend/crates/gateway-admin/src/model/retention.rs @@ -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 { if Self::validate_values(usage_days, ops_days, audit_days).is_err() { return Err(AdminError::invalid( - "请求保留期至少 31 天，事件与审计保留期必须为正数", + "请求保留期必须为 1～31 天，事件与审计保留期必须为正数", )); } Ok(Self { diff --git a/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs b/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs index cebe2f15a..44dbf4b02 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs @@ -1,4 +1,4 @@ -//! 有界统计缓存：身份和完整筛选参与键，删除版本阻止旧数据复活 +//! 有界统计缓存：身份和完整筛选参与键，统计发布代次使旧快照失效 use crate::model::observability::*; use crate::ports::store::{ @@ -121,7 +121,7 @@ impl Cache { }; // 同键刷新只执行一次；失败不会缓存成成功，也不延长旧结果寿命 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 diff --git a/backend/crates/gateway-admin/tests/model/retention.rs b/backend/crates/gateway-admin/tests/model/retention.rs index 4b384212a..3618a9a17 100644 --- a/backend/crates/gateway-admin/tests/model/retention.rs +++ b/backend/crates/gateway-admin/tests/model/retention.rs @@ -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); + } +} diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index 3a65926a0..c7019c46d 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -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, diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs index d41443659..4d1375168 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs @@ -55,7 +55,7 @@ impl<'a> AnalyticsSnapshot<'a> { }) } - pub async fn require_visible(&mut self, range: Option) -> StoreResult<()> { + pub fn require_visible(&self, range: Option) -> 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 { - 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)> { 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)?) diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs index 1a5fd265e..e3af9edb1 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs @@ -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))?; diff --git a/backend/crates/gateway-store/tests/postgres/analytics.rs b/backend/crates/gateway-store/tests/postgres/analytics.rs index 58031e786..90c6ec449 100644 --- a/backend/crates/gateway-store/tests/postgres/analytics.rs +++ b/backend/crates/gateway-store/tests/postgres/analytics.rs @@ -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 = 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] { diff --git a/backend/crates/gateway-store/tests/postgres/analytics_buckets.rs b/backend/crates/gateway-store/tests/postgres/analytics_buckets.rs index d0049263b..d5339c6a8 100644 --- a/backend/crates/gateway-store/tests/postgres/analytics_buckets.rs +++ b/backend/crates/gateway-store/tests/postgres/analytics_buckets.rs @@ -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(); diff --git a/backend/crates/gateway-store/tests/postgres/retention.rs b/backend/crates/gateway-store/tests/postgres/retention.rs index 580c69065..0c41c6bb6 100644 --- a/backend/crates/gateway-store/tests/postgres/retention.rs +++ b/backend/crates/gateway-store/tests/postgres/retention.rs @@ -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) = 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) = 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::>() + .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 = 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() { diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 4d69e34ad..5ff672ae9 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -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 diff --git a/backend/migrations/0032_request_history_retention.sql b/backend/migrations/0032_request_history_retention.sql new file mode 100644 index 000000000..c76642ea7 --- /dev/null +++ b/backend/migrations/0032_request_history_retention.sql @@ -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; diff --git a/docs/api.md b/docs/api.md index 5e4d09a45..1a5691c86 100644 --- a/docs/api.md +++ b/docs/api.md @@ -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`， diff --git a/frontend/src/views/settings/components/RequestHistoryCard.vue b/frontend/src/views/settings/components/RequestHistoryCard.vue new file mode 100644 index 000000000..5184b050d --- /dev/null +++ b/frontend/src/views/settings/components/RequestHistoryCard.vue @@ -0,0 +1,19 @@ + + + diff --git a/frontend/src/views/settings/composables/useSettingsForm.ts b/frontend/src/views/settings/composables/useSettingsForm.ts index 3c48cdb79..0d09c5511 100644 --- a/frontend/src/views/settings/composables/useSettingsForm.ts +++ b/frontend/src/views/settings/composables/useSettingsForm.ts @@ -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, diff --git a/frontend/src/views/settings/index.vue b/frontend/src/views/settings/index.vue index 1988f3f3f..a8bd8969a 100644 --- a/frontend/src/views/settings/index.vue +++ b/frontend/src/views/settings/index.vue @@ -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) => {