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

fix(store): prevent analytics stalls during history cleanup

Browse Source 
Batch empty-group deletion and extrema rebuilds, and reserve queue capacity
for live requests while historical contributions are removed.

Keep adaptive batch sizes across worker cycles so timed-out batches can
shrink and make progress without changing transaction or version ACK rules.

Add regression coverage for rollback, fair processing, and timeout recovery.
Validated with 452 Store tests, strict Clippy, and delayed PostgreSQL stress
with 30,000 historical requests plus 12,000 new requests at 200 per second.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-19 12:57:08 +08:00
co-authored by Codex
parent d678f350c1
commit b49b4528e7
6 changed files with 511 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-store/src/lib.rs
+1 -1
View File
@@ -46,4 +46,4 @@ pub use bundle::*;
pub use config::*;
pub use value::*;
pub(crate) use workers::*;
pub use workers::{CommandStoreDrainError, PostgresHealthProbe, run_request_analytics_cycle};
pub use workers::{CommandStoreDrainError, PostgresHealthProbe, RequestAnalyticsCycle};
backend/crates/gateway-store/src/postgres/observability/analytics/incremental.rs
+110 -64
View File
@@ -192,6 +192,8 @@ async fn apply_changes(
}
let mut buckets =
BTreeMap::<DateTime<Utc>, BTreeMap<(String, Vec<String>, String), MetricState>>::new();
let mut empty_groups = Vec::new();
let mut rebuild = Vec::new();
for (key, change) in changes {
if change.previous.encode()? == change.current.encode()?
&& change.previous.last_used_at() == change.current.last_used_at()
@@ -202,21 +204,17 @@ async fn apply_changes(
let needs_rebuild = subtract(&mut state, &change.previous, &change.current)?;
state.merge(&change.current)?;
if state.count("request_count") == 0 {
let statement = format!(
"delete from {table} where bucket_start=$1 and collection=$2 and dimensions=$3 and outcome=$4"
);
sqlx::query(sqlx::AssertSqlSafe(statement))
.bind(key.0)
.bind(&key.1)
.bind(&key.2)
.bind(&key.3)
.execute(&mut **tx)
.await
.map_err(|error| postgres_unavailable("remove empty analytics group", error))?;
empty_groups.push(serde_json::json!({
"bucket_start": key.0,
"collection": key.1,
"dimensions": key.2,
"outcome": key.3,
}));
} else {
// 预测状态只有可加计数和美元总额，零金额无需修复币种存在性
if needs_rebuild && key.1 != forecast::COLLECTION {
state = rebuild_group(tx, hourly, progress, &key).await?;
rebuild.push(key);
continue;
}
buckets
.entry(key.0)
@@ -224,6 +222,24 @@ async fn apply_changes(
.insert((key.1, key.2, key.3), state);
}
}
for (key, state) in rebuild_groups(tx, hourly, progress, rebuild).await? {
buckets
.entry(key.0)
.or_default()
.insert((key.1, key.2, key.3), state);
}
// 保留期缩短会同时清空大量分组，集合删除避免 SQL 往返随分组数放大
for groups in empty_groups.chunks(1_000) {
let statement = format!(
"delete from {table} a using jsonb_to_recordset($1::jsonb) k(bucket_start timestamptz,collection text,dimensions text[],outcome text)
where a.bucket_start=k.bucket_start and a.collection=k.collection and a.dimensions=k.dimensions and a.outcome=k.outcome"
);
sqlx::query(sqlx::AssertSqlSafe(statement))
.bind(sqlx::types::Json(groups))
.execute(&mut **tx)
.await
.map_err(|error| postgres_unavailable("remove empty analytics groups", error))?;
}
write_bucket_groups(tx, table, buckets).await
}
@@ -311,61 +327,91 @@ pub(super) fn subtract(
Ok(rebuild)
}
async fn rebuild_group(
async fn rebuild_groups(
tx: &mut sqlx::Transaction<'_, Postgres>,
hourly: bool,
progress: &BTreeMap<DateTime<Utc>, Progress>,
key: &GroupKey,
) -> StoreResult<MetricState> {
let mut end = key.0 + TimeDelta::seconds(if hourly { 3600 } else { 900 });
let mut cursor = None;
if hourly {
if let Some((_, owner)) = progress.range(key.0..end).find(|(_, p)| p.phase == 4)
&& let Some(time) = owner.started_at
{
let candidate = (time, key.1.clone(), key.2.clone(), key.3.clone());
end = if owner.contains_quarter(&candidate) {
time + TimeDelta::minutes(15)
} else {
time
};
keys: Vec<GroupKey>,
) -> StoreResult<BTreeMap<GroupKey, MetricState>> {
let mut by_collection = BTreeMap::<&str, Vec<&GroupKey>>::new();
for key in &keys {
by_collection.entry(&key.1).or_default().push(key);
}
let mut rebuilt = BTreeMap::new();
for (collection, keys) in by_collection {
let columns = worker::COLLECTION_COLUMNS
.iter()
.find_map(|(name, columns)| (*name == collection).then_some(*columns))
.ok_or_else(|| invalid("unknown incremental analytics collection"))?;
// 同类分组共用维度匹配和聚合 SQL，各组仍独立遵守已发布的构建前缀
for keys in keys.chunks(1_000) {
let mut input = Vec::with_capacity(keys.len());
for (index, key) in keys.iter().enumerate() {
let mut end = key.0 + TimeDelta::seconds(if hourly { 3600 } else { 900 });
let mut cursor = None;
if hourly {
if let Some((_, owner)) = progress.range(key.0..end).find(|(_, p)| p.phase == 4)
&& let Some(time) = owner.started_at
{
let candidate = (time, key.1.clone(), key.2.clone(), key.3.clone());
end = if owner.contains_quarter(&candidate) {
time + TimeDelta::minutes(15)
} else {
time
};
}
} else if let Some(stage) = progress.get(&key.0)
&& stage.phase == 1
{
cursor = stage.started_at.zip(stage.id.as_deref());
}
input.push(serde_json::json!({
"bucket_start": key.0,
"end_at": end,
"cursor_time": cursor.map(|(time, _)| time),
"cursor_id": cursor.map(|(_, id)| id),
"dimensions": key.2,
"outcome": key.3,
"name": index.to_string(),
}));
rebuilt.insert((*key).clone(), MetricState::default());
}
// 等值桶键避免范围连接反复比较其他时间桶，范围和游标仍限定已发布前缀
let mut query = QueryBuilder::<Postgres>::new(
"with matching as materialized (select mr.*,k.bucket_start as output_bucket,k.name,coalesce((",
);
query.push(completed_usage_fact_predicate("mr"))
.push("),false) as is_completed_usage from jsonb_to_recordset(")
.push_bind(serde_json::Value::Array(input))
.push("::jsonb) k(bucket_start timestamptz,end_at timestamptz,cursor_time timestamptz,cursor_id text,dimensions text[],outcome text,name text)
join request_analytics_facts mr on date_bin(make_interval(secs => ")
.push_bind(if hourly { 3600_f64 } else { 900_f64 })
.push("),mr.started_at,timestamptz '2000-01-01 UTC')=k.bucket_start
and mr.started_at >= k.bucket_start and mr.started_at < k.end_at
and mr.recovered_at is null and mr.outcome=k.outcome
and (k.cursor_time is null or (mr.started_at,mr.id collate \"C\") <= (k.cursor_time,k.cursor_id collate \"C\"))");
for (index, column) in columns.iter().enumerate() {
query.push(format!(
" and coalesce(mr.{column}::text,'')=k.dimensions[{}]",
index + 1
));
}
query.push(")");
aggregate::push_fact_aggregation(&mut query, false, true);
let rows = query.build().fetch_all(&mut **tx).await.map_err(|error| {
postgres_unavailable("rebuild changed analytics extrema", error)
})?;
for row in rows {
let name: String = get(&row, "name")?;
let index = name
.parse::<usize>()
.map_err(|_| invalid("invalid analytics rebuild group"))?;
let key = keys
.get(index)
.ok_or_else(|| invalid("unknown analytics rebuild group"))?;
rebuilt.insert((*key).clone(), MetricState::decode(&row)?);
}
}
} else if let Some(stage) = progress.get(&key.0)
&& stage.phase == 1
{
cursor = stage.started_at.zip(stage.id.as_deref());
}
let mut query = QueryBuilder::<Postgres>::new("with matching as materialized (select mr.*,");
query.push_bind(key.0).push("::timestamptz as output_bucket,''::text as name,coalesce((")
.push(completed_usage_fact_predicate("mr")).push("),false) as is_completed_usage from request_analytics_facts mr where mr.started_at >= ")
.push_bind(key.0).push(" and mr.started_at < ").push_bind(end)
.push(" and mr.recovered_at is null and mr.outcome=").push_bind(&key.3);
if let Some((time, id)) = cursor {
query
.push(" and (mr.started_at,mr.id collate \"C\") <= (")
.push_bind(time)
.push(",")
.push_bind(id)
.push(")");
}
let columns = worker::COLLECTION_COLUMNS
.iter()
.find_map(|(collection, columns)| (*collection == key.1).then_some(*columns))
.ok_or_else(|| invalid("unknown incremental analytics collection"))?;
for (column, value) in columns.iter().zip(&key.2) {
query
.push(format!(" and coalesce(mr.{column}::text,'')="))
.push_bind(value);
}
query.push(")");
aggregate::push_fact_aggregation(&mut query, false, true);
let row = query
.build()
.fetch_optional(&mut **tx)
.await
.map_err(|error| postgres_unavailable("rebuild changed analytics extrema", error))?;
row.as_ref()
.map(MetricState::decode)
.transpose()
.map(Option::unwrap_or_default)
Ok(rebuilt)
}
backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs
+11 -1
View File
@@ -83,7 +83,17 @@ impl PgRequestAnalyticsRepository {
return Ok(AnalyticsMaintenance { reset_progress: true, ..Default::default() });
}
let mut result = AnalyticsMaintenance::default();
let rows = sqlx::query("select request_id, change_version from request_analytics_pending_requests order by oldest_pending_at, request_id limit $1")
// 历史删除可能一次排入大量旧任务，给仍存在的请求预留半批名额，避免实时统计饥饿
// 其余名额仍按最早入队消费，两类任务都能前进；同一快照内选版本，沿用精确版本 ACK
let rows = sqlx::query("with live as materialized (
select pending.request_id, pending.change_version from request_analytics_pending_requests pending
join model_requests source on source.id=pending.request_id
order by pending.oldest_pending_at, pending.request_id limit ($1::bigint / 2)
), oldest as (
select pending.request_id, pending.change_version from request_analytics_pending_requests pending
where not exists (select 1 from live where live.request_id=pending.request_id)
order by pending.oldest_pending_at, pending.request_id limit ($1::bigint - (select count(*) from live))
) select request_id, change_version from live union all select request_id, change_version from oldest")
.bind(i64::from(batch_size)).fetch_all(&mut *transaction).await.map_err(|error| postgres_unavailable("load analytics request queue", error))?;
let mut jobs = Vec::with_capacity(rows.len());
for row in rows {
backend/crates/gateway-store/src/workers.rs
+108 -39
View File
@@ -2,6 +2,7 @@
use super::*;
use gateway_core::task::DaemonTask;
use std::sync::atomic::{AtomicU32, Ordering};
const COMMAND_DRAIN_TIMEOUT: Duration = Duration::from_secs(4);
@@ -325,12 +326,16 @@ pub(crate) fn request_analytics_worker(
Ok(WorkerContribution::Registration(scheduled_worker(
id,
Duration::from_secs(5),
Box::new(RequestAnalyticsTask { repository }),
Box::new(RequestAnalyticsTask {
repository,
cycle: RequestAnalyticsCycle::default(),
}),
)?))
}
struct RequestAnalyticsTask {
repository: postgres::PgRequestAnalyticsRepository,
cycle: RequestAnalyticsCycle,
}
impl ScheduledTask for RequestAnalyticsTask {
@@ -341,7 +346,7 @@ impl ScheduledTask for RequestAnalyticsTask {
Box::pin(async move {
tokio::select! {
() = context.cancellation().cancelled() => Ok(()),
result = run_request_analytics_cycle(&self.repository) => result.map(|_| ()).map_err(|source| {
result = self.cycle.run(&self.repository) => result.map(|_| ()).map_err(|source| {
WorkerTaskError::safe("request analytics maintenance failed").with_source(source)
}),
}
@@ -349,43 +354,107 @@ impl ScheduledTask for RequestAnalyticsTask {
}
}
/// 生产与隔离压测共用的单周期推进预算；各批独立提交并归还观测槽位
pub async fn run_request_analytics_cycle(
repository: &postgres::PgRequestAnalyticsRepository,
) -> StoreResult<postgres::AnalyticsMaintenance> {
let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
let mut total = postgres::AnalyticsMaintenance::default();
loop {
let result = match tokio::time::timeout_at(deadline, repository.maintain(1_000)).await {
Ok(result) => result?,
Err(source) => {
if total.projected_requests == 0
&& total.backfilled_requests == 0
&& total.rebuilt_buckets == 0
&& total.advanced_buckets == 0
&& !total.reset_progress
{
return Err(StoreError::Unavailable {
backend: StoreBackend::PostgreSql,
message: "request analytics cycle timed out without progress".to_owned(),
source: Some(gateway_core::error::ErrorSource::new(source)),
});
}
return Ok(total);
}
};
let idle = result.projected_requests == 0
&& result.backfilled_requests == 0
&& result.rebuilt_buckets == 0
&& result.advanced_buckets == 0
&& !result.reset_progress;
total.projected_requests += result.projected_requests;
total.backfilled_requests += result.backfilled_requests;
total.rebuilt_buckets += result.rebuilt_buckets;
total.advanced_buckets += result.advanced_buckets;
total.reset_progress |= result.reset_progress;
if idle || tokio::time::Instant::now() >= deadline {
return Ok(total);
/// 统计消费周期预算，跨轮保留批量上限以便超时后的下一轮提交进展
pub struct RequestAnalyticsCycle {
batch_size: AtomicU32,
}
impl Default for RequestAnalyticsCycle {
fn default() -> Self {
Self {
batch_size: AtomicU32::new(1_000),
}
}
}
impl RequestAnalyticsCycle {
/// 生产与隔离压测共用的推进路径；各批独立提交并归还观测槽位
pub async fn run(
&self,
repository: &postgres::PgRequestAnalyticsRepository,
) -> StoreResult<postgres::AnalyticsMaintenance> {
let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
let mut total = postgres::AnalyticsMaintenance::default();
loop {
let batch_size = self.batch_size.load(Ordering::Relaxed);
let started = tokio::time::Instant::now();
let result =
match tokio::time::timeout_at(deadline, repository.maintain(batch_size)).await {
Ok(Ok(result)) => result,
Ok(Err(error)) => {
if analytics_maintenance_timed_out(&error) {
self.reduce_batch_size(batch_size);
}
return Err(error);
}
Err(source) => {
if total.projected_requests == 0
&& total.backfilled_requests == 0
&& total.rebuilt_buckets == 0
&& total.advanced_buckets == 0
&& !total.reset_progress
{
// 只在整轮没有返回进展时退让，避免尾批剩余时间短导致持续缩批
self.reduce_batch_size(batch_size);
return Err(StoreError::Unavailable {
backend: StoreBackend::PostgreSql,
message: "request analytics cycle timed out without progress"
.to_owned(),
source: Some(gateway_core::error::ErrorSource::new(source)),
});
}
return Ok(total);
}
};
let idle = result.projected_requests == 0
&& result.backfilled_requests == 0
&& result.rebuilt_buckets == 0
&& result.advanced_buckets == 0
&& !result.reset_progress;
total.projected_requests += result.projected_requests;
total.backfilled_requests += result.backfilled_requests;
total.rebuilt_buckets += result.rebuilt_buckets;
total.advanced_buckets += result.advanced_buckets;
total.reset_progress |= result.reset_progress;
// 满批快速完成才恢复吞吐，空结果也可能只是其他实例持有维护锁
if (result.projected_requests >= u64::from(batch_size)
|| result.backfilled_requests >= u64::from(batch_size))
&& started.elapsed() < Duration::from_secs(2)
{
self.batch_size
.store((batch_size * 2).min(1_000), Ordering::Relaxed);
}
if idle {
return Ok(total);
}
if tokio::time::Instant::now() >= deadline {
return Ok(total);
}
}
}
fn reduce_batch_size(&self, batch_size: u32) {
let next_batch_size = (batch_size / 2).max(1);
self.batch_size.store(next_batch_size, Ordering::Relaxed);
tracing::warn!(
batch_size,
next_batch_size,
"request analytics batch timed out"
);
}
}
fn analytics_maintenance_timed_out(error: &StoreError) -> bool {
let StoreError::Unavailable {
source: Some(source),
..
} = error
else {
return false;
};
source.is::<tokio::time::error::Elapsed>()
|| source
.downcast_ref::<sqlx::Error>()
.and_then(sqlx::Error::as_database_error)
.is_some_and(|error| error.code().as_deref() == Some("57014"))
}
backend/crates/gateway-store/tests/postgres/retention.rs
+277 -8
View File
@@ -1,14 +1,18 @@
//! 验证短保留期升级、分批清理边界与统计异步收敛期间的查询可用性
use std::num::NonZeroU32;
use std::{num::NonZeroU32, time::Duration};
use chrono::{Duration as ChronoDuration, Utc};
use gateway_admin::{
model::retention::{RetentionPolicy, RetentionTarget},
ports::retention::RetentionStore as _,
ports::{retention::RetentionStore as _, store::ObservabilityStore as _},
};
use gateway_store::postgres::{
ObservabilityRange, PgRequestAnalyticsRepository, PgRetentionRepository,
use gateway_store::{
RequestAnalyticsCycle,
postgres::{
ObservabilityQueryBudget, ObservabilityRange, PgRequestAnalyticsRepository,
PgRetentionRepository,
},
};
use super::{
@@ -18,10 +22,7 @@ use super::{
#[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 _,
};
use gateway_admin::model::observability::{AnalyticsState, DashboardQuery, TimeRange};
let Some(database) = TestDatabase::create("short_retention_readable").await else {
return;
@@ -280,3 +281,271 @@ async fn empty_request_cleanup_does_not_conflict_with_analytics_publication() {
publication.commit().await.unwrap();
database.close().await;
}
#[tokio::test]
async fn scattered_retention_deletions_commit_atomically_and_allow_new_requests_to_advance() {
let Some(database) = TestDatabase::create("retention_scattered_groups").await else {
return;
};
let now = Utc::now();
for number in 0..600 {
seed(
&database.pool,
&format!("expired_{number:04}"),
now - ChronoDuration::days(30) + ChronoDuration::minutes(number * 20),
&[],
)
.await;
}
drain_analytics(&database.pool).await;
let worker = PgRequestAnalyticsRepository::new(
database.pool.clone(),
ObservabilityQueryBudget::try_new(2, Duration::from_secs(5), Duration::from_secs(120))
.unwrap(),
);
let before = worker.status().await.unwrap();
let retention = PgRetentionRepository::new(database.pool.clone());
assert_eq!(
retention
.purge_batch(
RetentionTarget::ModelRequests,
now,
RetentionPolicy::try_new(10, 30, 90).unwrap(),
NonZeroU32::new(1000).unwrap(),
)
.await
.unwrap(),
600
);
seed(
&database.pool,
"new_request",
now - ChronoDuration::minutes(2),
&[],
)
.await;
// 小时层失败时，已经执行的多个刻钟删除批次与事实删除也必须一起回滚
sqlx::raw_sql("create function reject_hour_deletion() returns trigger language plpgsql as $$begin raise exception 'test hour deletion failure'; end$$;
create trigger reject_hour_deletion before delete on request_analytics_1h for each statement execute function reject_hour_deletion()")
.execute(&database.pool).await.unwrap();
let cycle = RequestAnalyticsCycle::default();
assert!(cycle.run(&worker).await.is_err());
let failed = worker.status().await.unwrap();
assert_eq!(failed.generation, before.generation);
assert_eq!(failed.pending_requests, 601);
let statistics = observability_repository(&database.pool);
let range = ObservabilityRange::new(now - ChronoDuration::days(31), now).unwrap();
assert_eq!(
statistics
.usage_summary(range, Default::default())
.await
.unwrap()
.requests
.request_count,
600
);
sqlx::raw_sql("drop trigger reject_hour_deletion on request_analytics_1h; drop function reject_hour_deletion();
create function slow_summary_statement() returns trigger language plpgsql as $$begin perform pg_sleep(0.002); return null; end$$;
create trigger slow_quarter_delete before delete on request_analytics_15m for each statement execute function slow_summary_statement();
create trigger slow_hour_delete before delete on request_analytics_1h for each statement execute function slow_summary_statement()")
.execute(&database.pool).await.unwrap();
// 按语句注入延迟，覆盖大量分组的 SQL 往返成本及多个集合删除批次
cycle
.run(&worker)
.await
.expect("retention backlog must advance within the cycle budget");
let after = worker.status().await.unwrap();
assert_eq!(after.pending_requests, 0);
assert_eq!(after.pending_buckets, 0);
assert!(after.generation > before.generation);
let actual = statistics
.usage_summary(range, Default::default())
.await
.unwrap();
assert_eq!(actual.requests.request_count, 1);
assert_eq!(actual.requests.total_tokens, 120);
let timeline = admin_observability_store(&database.pool)
.dashboard_summary(
gateway_admin::model::observability::DashboardQuery::new(
gateway_admin::model::observability::TimeRange {
start: now - ChronoDuration::days(1),
end: now,
},
Default::default(),
)
.unwrap(),
now,
)
.await
.unwrap();
assert_eq!(
timeline
.trend
.iter()
.map(|point| point.metrics.request_count)
.sum::<u64>(),
1
);
let remaining: (i64, i64, i64, i64) = sqlx::query_as(
"select
(select count(*) from request_analytics_facts where id like 'expired_%'),
(select count(*) from request_analytics_15m where bucket_start < $1),
(select count(*) from request_analytics_1h where bucket_start < $1),
(select count(*) from request_analytics_1d_counts where bucket_start < $1)",
)
.bind(now - ChronoDuration::days(10))
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(remaining, (0, 0, 0, 0));
database.close().await;
}
#[tokio::test]
async fn fresh_requests_and_retention_deletions_both_advance_within_each_batch() {
let Some(database) = TestDatabase::create("retention_fair_projection").await else {
return;
};
let now = Utc::now();
for number in 0..20 {
seed(
&database.pool,
&format!("expired_{number:04}"),
now - ChronoDuration::days(30),
&[],
)
.await;
}
drain_analytics(&database.pool).await;
PgRetentionRepository::new(database.pool.clone())
.purge_batch(
RetentionTarget::ModelRequests,
now,
RetentionPolicy::try_new(10, 30, 90).unwrap(),
NonZeroU32::new(1000).unwrap(),
)
.await
.unwrap();
let worker =
PgRequestAnalyticsRepository::new(database.pool.clone(), observability_query_budget());
for wave in 1..=3 {
for number in 0..2 {
seed(
&database.pool,
&format!("new_{wave}_{number}"),
now - ChronoDuration::minutes(2),
&[],
)
.await;
}
assert_eq!(worker.maintain(4).await.unwrap().projected_requests, 4);
let facts: (i64, i64) = sqlx::query_as(
"select count(*) filter (where id like 'expired_%'), count(*) filter (where id like 'new_%') from request_analytics_facts",
)
.fetch_one(&database.pool)
.await
.unwrap();
// 历史队列尚未排空时，每批都须撤销旧贡献并发布本轮的新请求
assert_eq!(facts, (20 - wave * 2, wave * 2));
let current = observability_repository(&database.pool)
.usage_summary(
ObservabilityRange::new(now - ChronoDuration::days(1), now).unwrap(),
Default::default(),
)
.await
.unwrap();
assert_eq!(current.requests.request_count, wave as u64 * 2);
}
drain_analytics(&database.pool).await;
let final_summary = observability_repository(&database.pool)
.usage_summary(
ObservabilityRange::new(now - ChronoDuration::days(31), now).unwrap(),
Default::default(),
)
.await
.unwrap();
assert_eq!(final_summary.requests.request_count, 6);
assert_eq!(final_summary.requests.total_tokens, 720);
database.close().await;
}
#[tokio::test]
async fn timed_out_retention_batches_shrink_across_cycles_and_lock_contention() {
let Some(database) = TestDatabase::create("retention_batch_timeout").await else {
return;
};
let now = Utc::now();
for number in 0..600 {
seed(
&database.pool,
&format!("expired_{number:04}"),
now - ChronoDuration::days(30),
&[],
)
.await;
}
drain_analytics(&database.pool).await;
sqlx::query("delete from model_requests")
.execute(&database.pool)
.await
.unwrap();
seed(
&database.pool,
"new_request",
now - ChronoDuration::minutes(2),
&[],
)
.await;
// 数据库延迟与实际批量相关，小批量无需等待，以可观察进展验证跨轮退让
sqlx::raw_sql("create function slow_large_fact_deletion() returns trigger language plpgsql as $$begin
if (select count(*) from removed_facts) > 250 then perform pg_sleep(2); end if;
return null; end$$;
create trigger slow_large_fact_deletion after delete on request_analytics_facts referencing old table as removed_facts for each statement execute function slow_large_fact_deletion()")
.execute(&database.pool).await.unwrap();
let worker = PgRequestAnalyticsRepository::new(
database.pool.clone(),
ObservabilityQueryBudget::try_new(2, Duration::from_secs(1), Duration::from_secs(1))
.unwrap(),
);
let cycle = RequestAnalyticsCycle::default();
for _ in 0..2 {
assert!(cycle.run(&worker).await.is_err());
// 空轮可能来自别的实例持锁，不能因此恢复会反复超时的大批量
let mut lock = database.pool.begin().await.unwrap();
sqlx::query(
"select pg_advisory_xact_lock(hashtextextended(current_schema()::text,382032))",
)
.execute(&mut *lock)
.await
.unwrap();
let idle = cycle.run(&worker).await.unwrap();
assert_eq!(idle.projected_requests, 0);
lock.rollback().await.unwrap();
}
let pending = worker.status().await.unwrap().pending_requests;
assert_eq!(pending, 601);
// 此轮可以在后续批次超时前已有提交，下一轮重放仍须保持计数正确
let _ = cycle.run(&worker).await;
assert!(worker.status().await.unwrap().pending_requests < pending);
for _ in 0..5 {
// 与调度器错误重试一样留出退避时间，让已取消语句完成回滚并释放维护锁
tokio::time::sleep(Duration::from_secs(1)).await;
let _ = cycle.run(&worker).await;
if worker.status().await.unwrap().pending_requests == 0 {
break;
}
}
assert_eq!(worker.status().await.unwrap().pending_requests, 0);
let actual = observability_repository(&database.pool)
.usage_summary(
ObservabilityRange::new(now - ChronoDuration::days(31), now).unwrap(),
Default::default(),
)
.await
.unwrap();
assert_eq!(actual.requests.request_count, 1);
assert_eq!(actual.requests.total_tokens, 120);
database.close().await;
}
docs/architecture.md
+4
View File
@@ -926,6 +926,10 @@ Store 从请求源事实维护窄事实及十五分钟、小时统计状态，
完整时间桶在数据库内合并，边界与不能由汇总表达的筛选在数据库内精确聚合；向应用返回的状态按输出分组组织。
计数、金额与分布状态的旧贡献撤销、新贡献合并和事实替换共用事务，构建期间只调整已扫描前缀，
不可相减的极值按受影响分组重算。同步账号用量查询在同一快照补齐待投影源版本，读取结果包含尚未完成异步投影的用量。
空统计分组删除和极值重算在事务内按集合分批执行，避免历史清理产生逐组 SQL 往返。
请求队列为仍存在的源请求预留部分批量，其余名额按最早入队消费，清理积压期间持续发布新请求统计，同时推进旧贡献撤销。
统计 Worker 跨周期保留消费批量上限，整轮无进展超时或存储操作超时后缩小批量，满批快速完成时恢复吞吐；
维护锁竞争造成的空轮不重置批量，未提交事务与未确认队列项由后续周期重试。
### 额度预测