diff --git a/backend/crates/gateway-store/src/lib.rs b/backend/crates/gateway-store/src/lib.rs
index 6859cdb9f..f85ddfcda 100644
--- a/backend/crates/gateway-store/src/lib.rs
+++ b/backend/crates/gateway-store/src/lib.rs
@@ -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};
diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/incremental.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/incremental.rs
index 89f0bbcb6..3fe8653d0 100644
--- a/backend/crates/gateway-store/src/postgres/observability/analytics/incremental.rs
+++ b/backend/crates/gateway-store/src/postgres/observability/analytics/incremental.rs
@@ -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)
 }
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 e3af9edb1..e14228306 100644
--- a/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs
+++ b/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs
@@ -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 {
diff --git a/backend/crates/gateway-store/src/workers.rs b/backend/crates/gateway-store/src/workers.rs
index 82938e8e0..489735712 100644
--- a/backend/crates/gateway-store/src/workers.rs
+++ b/backend/crates/gateway-store/src/workers.rs
@@ -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"))
+}
diff --git a/backend/crates/gateway-store/tests/postgres/retention.rs b/backend/crates/gateway-store/tests/postgres/retention.rs
index ca0577e8d..1c8df7ac1 100644
--- a/backend/crates/gateway-store/tests/postgres/retention.rs
+++ b/backend/crates/gateway-store/tests/postgres/retention.rs
@@ -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;
+}
diff --git a/docs/architecture.md b/docs/architecture.md
index 5fc8f326f..99e88a503 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -926,6 +926,10 @@ Store 从请求源事实维护窄事实及十五分钟、小时统计状态，
 完整时间桶在数据库内合并，边界与不能由汇总表达的筛选在数据库内精确聚合；向应用返回的状态按输出分组组织。
 计数、金额与分布状态的旧贡献撤销、新贡献合并和事实替换共用事务，构建期间只调整已扫描前缀，
 不可相减的极值按受影响分组重算。同步账号用量查询在同一快照补齐待投影源版本，读取结果包含尚未完成异步投影的用量。
+空统计分组删除和极值重算在事务内按集合分批执行，避免历史清理产生逐组 SQL 往返。
+请求队列为仍存在的源请求预留部分批量，其余名额按最早入队消费，清理积压期间持续发布新请求统计，同时推进旧贡献撤销。
+统计 Worker 跨周期保留消费批量上限，整轮无进展超时或存储操作超时后缩小批量，满批快速完成时恢复吞吐；
+维护锁竞争造成的空轮不重置批量，未提交事务与未确认队列项由后续周期重试。
 
 ### 额度预测