fix(store): bound extrema rebuilds to their analytics bucket (#477)

Group extrema rebuilds by collection and analytics bucket, with explicit started_at bounds so large fact histories can use the time-range index. Preserve per-group build prefixes and aggregate results.

Keep the scan regression test on a single PostgreSQL connection so forced statistics flushes include the worker's reads. The updated test passes with the bounded query and rejects the previous query after reading 6,000,040 rows.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-store/src/postgres/observability/analytics/incremental.rs

@@ -333,16 +333,18 @@ async fn rebuild_groups(
     progress: &BTreeMap<DateTime<Utc>, Progress>,
     keys: Vec<GroupKey>,
 ) -> StoreResult<BTreeMap<GroupKey, MetricState>> {
-    let mut by_collection = BTreeMap::<&str, Vec<&GroupKey>>::new();
+    // 同一统计桶的分组共享时间范围，按桶查询才能让事实表范围扫描命中 started_at 索引
+    let mut by_bucket = BTreeMap::<(&str, DateTime<Utc>), Vec<&GroupKey>>::new();
     for key in &keys {
-        by_collection.entry(&key.1).or_default().push(key);
+        by_bucket.entry((&key.1, key.0)).or_default().push(key);
     }
     let mut rebuilt = BTreeMap::new();
-    for (collection, keys) in by_collection {
+    for ((collection, bucket_start), keys) in by_bucket {
         let columns = worker::COLLECTION_COLUMNS
             .iter()
             .find_map(|(name, columns)| (*name == collection).then_some(*columns))
             .ok_or_else(|| invalid("unknown incremental analytics collection"))?;
+        let bucket_end = bucket_start + TimeDelta::seconds(if hourly { 3600 } else { 900 });
         // 同类分组共用维度匹配和聚合 SQL，各组仍独立遵守已发布的构建前缀
         for keys in keys.chunks(1_000) {
             let mut input = Vec::with_capacity(keys.len());
@@ -389,7 +391,11 @@ async fn rebuild_groups(
                 .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\"))");
+                    and (k.cursor_time is null or (mr.started_at,mr.id collate \"C\") <= (k.cursor_time,k.cursor_id collate \"C\"))
+                    and mr.started_at >= ")
+                .push_bind(bucket_start)
+                .push(" and mr.started_at < ")
+                .push_bind(bucket_end);
             for (index, column) in columns.iter().enumerate() {
                 query.push(format!(
                     " and coalesce(mr.{column}::text,'')=k.dimensions[{}]",

backend/crates/gateway-store/tests/postgres/analytics_buckets.rs

@@ -1,7 +1,10 @@
 //! 小时重建合并、分段快照失效与阻塞桶调度的真实数据库验收
 
 use chrono::{DateTime, TimeDelta, Utc};
-use gateway_store::postgres::{ObservabilityRange, PgRequestAnalyticsRepository};
+use gateway_store::{
+    RequestAnalyticsCycle,
+    postgres::{ObservabilityRange, PgRequestAnalyticsRepository},
+};
 
 use super::{
     TestDatabase, analytics::seed, drain_analytics, observability_query_budget,
@@ -352,3 +355,64 @@ async fn incremental_summary_write_failure_rolls_back_facts_and_preserves_queue(
     assert_eq!(actual.requests.request_count, 1);
     database.close().await;
 }
+
+#[tokio::test]
+async fn extrema_rebuild_reads_only_its_bucket_when_history_is_large() {
+    let Some(database) = TestDatabase::create("analytics_rebuild_bucket_range").await else {
+        return;
+    };
+    // 强制刷新只作用于当前连接，Worker 与计数读取必须复用同一条连接
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(1)
+        .connect_with(database.pool.connect_options().as_ref().clone())
+        .await
+        .unwrap();
+    for number in 1..=3_i64 {
+        let id = format!("rebuild_range_{number}");
+        seed(&pool, &id, hour() + TimeDelta::minutes(number), &[]).await;
+        sqlx::query("update model_requests set provider_account_ref='rebuild_range_account',request_observation_json=jsonb_set(request_observation_json,'{timings,local}',jsonb_build_object('latencyMs',$2::bigint,'firstTokenMs',$2::bigint)) where id=$1")
+            .bind(id).bind(number).execute(&pool).await.unwrap();
+    }
+    drain_analytics(&pool).await;
+    // 只扩充其他时间桶的事实，规划器拿到真实行数后不能再为一个分组扫描全部历史
+    sqlx::query("insert into request_analytics_facts select ('history_'||n)::text as id,(f.started_at-make_interval(hours=>n/1000+2))::timestamptz,f.completed_at-make_interval(hours=>n/1000+2),f.client_api_key_ref,f.provider_account_ref,f.routing_group_refs,f.operation,f.provider_kind,f.requested_model_id,f.upstream_model_id,f.outcome,f.client_status_code,f.upstream_status_code,f.client_transport,f.upstream_transport,f.attempt_count,null,null,null,f.recovered_at,f.downstream_committed_at,f.request_kind,f.image_generation_requested,f.image_generation_succeeded,f.error_kind,f.input_tokens,f.output_tokens,f.cached_input_tokens,f.cache_write_input_tokens,f.reasoning_output_tokens,f.image_input_tokens,f.image_output_tokens,f.total_tokens,f.cost_source,f.cost_currency,f.cost_amount,f.first_token_ms,f.latency_ms,f.admission_decision_ms,f.account_selection_wait_ms,f.upstream_response_ms,f.capacity_used_slots,f.capacity_total_slots,f.standard_cost_amount,f.no_cache_cost_amount,f.source_version,f.schema_version,f.ops_failure_count,f.ops_rate_limited_count,f.ops_auth_failure_count,f.ops_provider_5xx_count from request_analytics_facts f cross join generate_series(1,300000) n where f.id='rebuild_range_2'")
+        .execute(&pool).await.unwrap();
+    sqlx::query("analyze request_analytics_facts")
+        .execute(&pool)
+        .await
+        .unwrap();
+    sqlx::query("delete from model_requests where id='rebuild_range_1'")
+        .execute(&pool)
+        .await
+        .unwrap();
+    let worker = PgRequestAnalyticsRepository::new(pool.clone(), observability_query_budget());
+    let scanned_before = sequentially_read_facts(&pool).await;
+    RequestAnalyticsCycle::default()
+        .run(&worker)
+        .await
+        .expect("extrema rebuild must finish within the cycle budget");
+    // 耗时随机器和数据量变化，用顺序扫描读取的行数判断重建是否越出所在统计桶
+    let scanned = sequentially_read_facts(&pool).await - scanned_before;
+    assert!(
+        scanned < 300_000,
+        "extrema rebuild scanned {scanned} fact rows"
+    );
+    let status = worker.status().await.unwrap();
+    assert_eq!((status.pending_requests, status.pending_buckets), (0, 0));
+    let row: (i64, f64, f64) = sqlx::query_as("select (counts->>'request_count')::bigint,(distributions->'latency_ms'->>'minimum')::double precision,(distributions->'latency_ms'->>'maximum')::double precision from request_analytics_15m where collection='global' and bucket_start=$1")
+        .bind(hour()).fetch_one(&pool).await.unwrap();
+    assert_eq!(row, (2, 2.0, 3.0));
+    pool.close().await;
+    database.close().await;
+}
+
+async fn sequentially_read_facts(pool: &sqlx::PgPool) -> i64 {
+    sqlx::query("select pg_stat_force_next_flush()")
+        .execute(pool)
+        .await
+        .unwrap();
+    sqlx::query_scalar("select coalesce(seq_tup_read,0) from pg_stat_user_tables where relid='request_analytics_facts'::regclass")
+        .fetch_one(pool)
+        .await
+        .unwrap()
+}