mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
refactor(store): consolidate unreleased database migrations
Merge migrations 0030 through 0032 after the v3.25.0 release baseline. Queue analytics bucket rebuilds once and verify upgrade preservation and repeated startup. Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -0,0 +1,87 @@
|
||||
//! 验证统计升级保留历史事实与回填进度,并重建已有时间桶
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde_json::Value;
|
||||
|
||||
use super::{TEST_MIGRATOR, TestDatabase};
|
||||
|
||||
#[tokio::test]
|
||||
async fn analytics_upgrade_preserves_facts_and_backfill_progress_and_requeues_buckets() {
|
||||
let Some(database) = TestDatabase::create_through("analytics_upgrade", 29).await else {
|
||||
return;
|
||||
};
|
||||
sqlx::raw_sql(
|
||||
"insert into request_analytics_facts(id,started_at,source_version) values
|
||||
('first','2026-10-01 10:01:00+00',1),
|
||||
('same_bucket','2026-10-01 10:02:00+00',2),
|
||||
('next_bucket','2026-10-01 10:16:00+00',3);
|
||||
insert into request_analytics_pending_buckets(bucket_start,oldest_pending_at)
|
||||
values ('2026-10-01 10:00:00+00','2026-10-01 09:00:00+00');
|
||||
update request_analytics_build_state set
|
||||
backfill_complete=true, backfill_cursor_id='next_bucket',
|
||||
backfill_cursor_started_at='2026-10-01 10:16:00+00',
|
||||
coverage_start='2026-10-01 10:00:00+00',
|
||||
coverage_end='2026-10-01 10:30:00+00', generation=7;",
|
||||
)
|
||||
.execute(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let facts_before: Vec<Value> =
|
||||
sqlx::query_scalar("select to_jsonb(f) from request_analytics_facts f order by id")
|
||||
.fetch_all(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let progress_before: Value =
|
||||
sqlx::query_scalar("select to_jsonb(s) from request_analytics_build_state s")
|
||||
.fetch_one(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let pending_before: (i64, DateTime<Utc>) = sqlx::query_as(
|
||||
"select change_version,oldest_pending_at from request_analytics_pending_buckets",
|
||||
)
|
||||
.fetch_one(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
TEST_MIGRATOR.run(&database.pool).await.unwrap();
|
||||
|
||||
let facts_after: Vec<Value> =
|
||||
sqlx::query_scalar("select to_jsonb(f) from request_analytics_facts f order by id")
|
||||
.fetch_all(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let progress_after: Value =
|
||||
sqlx::query_scalar("select to_jsonb(s) from request_analytics_build_state s")
|
||||
.fetch_one(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(facts_after, facts_before);
|
||||
assert_eq!(progress_after, progress_before);
|
||||
let pending: Vec<(DateTime<Utc>, i64, DateTime<Utc>)> = sqlx::query_as(
|
||||
"select bucket_start,change_version,oldest_pending_at
|
||||
from request_analytics_pending_buckets order by bucket_start",
|
||||
)
|
||||
.fetch_all(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let expected_buckets = ["2026-10-01T10:00:00Z", "2026-10-01T10:15:00Z"]
|
||||
.map(|value| value.parse::<DateTime<Utc>>().unwrap());
|
||||
assert_eq!(
|
||||
pending.iter().map(|row| row.0).collect::<Vec<_>>(),
|
||||
expected_buckets
|
||||
);
|
||||
assert!(pending[0].1 > pending_before.0);
|
||||
assert_eq!(pending[0].2, pending_before.1);
|
||||
|
||||
// 重启不会再次入队,也不会覆盖最早待处理时间
|
||||
TEST_MIGRATOR.run(&database.pool).await.unwrap();
|
||||
let pending_after_restart: Vec<(DateTime<Utc>, i64, DateTime<Utc>)> = sqlx::query_as(
|
||||
"select bucket_start,change_version,oldest_pending_at
|
||||
from request_analytics_pending_buckets order by bucket_start",
|
||||
)
|
||||
.fetch_all(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(pending_after_restart, pending);
|
||||
database.close().await;
|
||||
}
|
||||
@@ -18,6 +18,7 @@ mod admission_recovery;
|
||||
mod analytics;
|
||||
mod analytics_buckets;
|
||||
mod analytics_reset;
|
||||
mod analytics_upgrade;
|
||||
mod backup;
|
||||
mod client_budgets;
|
||||
mod client_keys;
|
||||
|
||||
@@ -61,7 +61,7 @@ fn runtime_settings_keep_account_rotation_global() {
|
||||
#[tokio::test]
|
||||
async fn removing_request_interval_preserves_settings_and_snapshot() {
|
||||
use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository};
|
||||
let Some(database) = TestDatabase::create_through("drop_interval", 31).await else {
|
||||
let Some(database) = TestDatabase::create_through("drop_interval", 29).await else {
|
||||
return;
|
||||
};
|
||||
sqlx::query("update runtime_settings set request_interval_ms = 60000, max_concurrent_per_account = 7, refresh_concurrency = 12, refresh_margin_seconds = 1234, config_revision = 70 where id = 1")
|
||||
|
||||
@@ -27,6 +27,4 @@ d4b599066d7382a1c9cb6b3c003200046579e4fce3531f232cbd665a9d92c542 0026_codex_pri
|
||||
8c9bbecd8315a26e928213a82a2ec651c35976e0643b82caef1d1bcfeee2916f 0027_request_observation_projection.sql
|
||||
31d6668707186d7c19cdd507e20162a08c7fbb8ba41ff4a4439d0a5e08369cad 0028_request_analytics.sql
|
||||
1923f14095dac5f808a530db002c37e17168afa0778bc7f3da768c818133eec9 0029_usage_indexes_and_identity_snapshots.sql
|
||||
ab964fcdf56aa1f595fffaf02556aeebce5037f7bfa1ea6f4fef594112eaba6a 0030_analytics_usage_dimensions.sql
|
||||
9d6e5254ee4a7aec40c4fc5f1904300228c57826941b962fa78cc6e550fa99f8 0031_analytics_forecast_completion.sql
|
||||
a3bc8fee6fe567a0c15ddb0c50b3189e27fb89c4dc43105a4608df7aa33d2dac 0032_remove_request_interval.sql
|
||||
f10058e2b4c974b6d2ffa11be78b085504a233205253948c03357ef6fe234bca 0030_analytics_and_scheduling.sql
|
||||
|
||||
+22
-7
@@ -2,13 +2,6 @@
|
||||
alter table request_analytics_15m add column last_used_at timestamptz;
|
||||
alter table request_analytics_hourly add column last_used_at timestamptz;
|
||||
|
||||
-- 新指标只重建时间桶,不清空窄事实,也不重置已经完成的历史投影游标
|
||||
insert into request_analytics_pending_buckets(bucket_start, oldest_pending_at)
|
||||
select distinct date_bin(interval '15 minutes', started_at, timestamptz '2000-01-01 UTC'), now()
|
||||
from request_analytics_facts
|
||||
on conflict (bucket_start) do update
|
||||
set change_version = nextval('request_analytics_change_version');
|
||||
|
||||
-- 撤销最近一次模型贡献后按真实源定位极值,避免扫描整个账号历史
|
||||
create index model_requests_delivered_account_model on model_requests
|
||||
(provider_account_ref,(coalesce(upstream_model_id,requested_model_id,'')),started_at desc,id desc)
|
||||
@@ -31,3 +24,25 @@ create index request_analytics_facts_build_cursor on request_analytics_facts
|
||||
(started_at,id collate "C");
|
||||
create index request_analytics_15m_build_cursor on request_analytics_15m
|
||||
(bucket_start,collection collate "C",dimensions collate "C",outcome collate "C");
|
||||
|
||||
-- 完成时间边界只回读五分钟内的窄事实,Provider 文档继续使用源表采样索引
|
||||
create index request_analytics_facts_account_completion on request_analytics_facts
|
||||
(provider_account_ref,completed_at,started_at)
|
||||
where completed_at is not null and outcome <> 'running';
|
||||
|
||||
-- 新指标只重建时间桶,不清空窄事实,也不重置已经完成的历史投影游标
|
||||
insert into request_analytics_pending_buckets(bucket_start, oldest_pending_at)
|
||||
select distinct date_bin(interval '15 minutes', started_at, timestamptz '2000-01-01 UTC'), now()
|
||||
from request_analytics_facts
|
||||
on conflict (bucket_start) do update
|
||||
set change_version = nextval('request_analytics_change_version');
|
||||
|
||||
-- 移除账号请求间隔,保留刷新与并发配置的约束
|
||||
alter table runtime_settings
|
||||
drop constraint runtime_settings_refresh_ck,
|
||||
drop column request_interval_ms,
|
||||
add constraint runtime_settings_refresh_ck check (
|
||||
refresh_margin_seconds > 0
|
||||
and refresh_concurrency > 0
|
||||
and max_concurrent_per_account >= 0
|
||||
);
|
||||
@@ -1,11 +0,0 @@
|
||||
-- 完成时间边界只回读五分钟内的窄事实,Provider 文档继续使用源表采样索引
|
||||
create index request_analytics_facts_account_completion on request_analytics_facts
|
||||
(provider_account_ref,completed_at,started_at)
|
||||
where completed_at is not null and outcome <> 'running';
|
||||
|
||||
-- 新完成时间维度与既有统计共用可恢复构建,旧事实无需再次投影
|
||||
insert into request_analytics_pending_buckets(bucket_start, oldest_pending_at)
|
||||
select distinct date_bin(interval '15 minutes', started_at, timestamptz '2000-01-01 UTC'), now()
|
||||
from request_analytics_facts
|
||||
on conflict (bucket_start) do update
|
||||
set change_version = nextval('request_analytics_change_version');
|
||||
@@ -1,8 +0,0 @@
|
||||
alter table runtime_settings
|
||||
drop constraint runtime_settings_refresh_ck,
|
||||
drop column request_interval_ms,
|
||||
add constraint runtime_settings_refresh_ck check (
|
||||
refresh_margin_seconds > 0
|
||||
and refresh_concurrency > 0
|
||||
and max_concurrent_per_account >= 0
|
||||
);
|
||||
Reference in New Issue
Block a user