fix(backup): paginate scheduled retention cleanup (#552)

Scan completed scheduled backups in bounded pages and evaluate retention
against their global positions. Collect a deletion batch before deleting
records so cleanup does not shift its own later page offsets.

Cover count and age thresholds beyond 1000 records and PostgreSQL page
ordering with tied completion times.

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-admin/src/backup/policy.rs

@@ -126,30 +126,32 @@ pub struct RetentionDecision {
 /// 自动保留规则：
 /// - `retentionDays = 0` 时不按天数清理；`retentionCount = 0` 时不按份数清理
 /// - 自动计划备份满足任一启用阈值即可进入删除
-/// - 无论阈值如何，至少保留最近一个成功的计划备份（倒序下标 0）
+/// - 无论阈值如何，至少保留最近一个成功的计划备份（全局倒序下标 0）
 /// - 手动备份不进入本规则；手工/计划记录各自的 `expires_at` 由到期扫描处理
+/// - `offset` 是当前页在全部 completed 计划备份中的起始位置
 #[must_use]
 pub fn decide_retention(
     retention_days: u32,
     retention_count: u32,
     now: DateTime<Utc>,
+    offset: u64,
     scheduled_completed_desc: &[BackupRecord],
 ) -> Vec<RetentionDecision> {
     let mut decisions = Vec::new();
     for (index, record) in scheduled_completed_desc.iter().enumerate() {
+        let position = offset.saturating_add(index as u64);
         if record.status != BackupStatus::Completed {
             continue;
         }
         // 最近一个成功的计划备份永远保留
-        if index == 0 {
+        if position == 0 {
             continue;
         }
         let expired_by_days = retention_days > 0
             && record.completed_at.is_some_and(|completed_at| {
                 completed_at <= now - Duration::days(i64::from(retention_days))
             });
-        let exceeded_count =
-            retention_count > 0 && index >= usize::try_from(retention_count).unwrap_or(usize::MAX);
+        let exceeded_count = retention_count > 0 && position >= u64::from(retention_count);
         if expired_by_days || exceeded_count {
             decisions.push(RetentionDecision {
                 record_id: record.id.clone(),

backend/crates/gateway-admin/src/backup/task.rs

@@ -38,8 +38,8 @@ use super::policy::{BackupSchedule, decide_retention};
 const BACKUP_LOOP_INTERVAL: Duration = Duration::from_secs(30);
 /// 每个循环最多执行的保留删除数量
 const RETENTION_BATCH_SIZE: usize = 10;
-/// 保留扫描一次读取的 completed 计划备份上限
-const RETENTION_SCAN_LIMIT: u32 = 1000;
+/// 保留扫描每页读取的 completed 计划备份数量
+const RETENTION_PAGE_SIZE: u32 = 1000;
 /// 每个循环最多完成的待删除记录数量
 const PENDING_DELETION_BATCH: u32 = 20;
 
@@ -112,7 +112,7 @@ impl BackupTask {
         self.recover_intermediate(cancellation).await?;
         self.finish_pending_deletions().await?;
         self.process_next_task(cancellation).await?;
-        self.run_retention_batch(Utc::now()).await?;
+        self.run_retention_batch(Utc::now(), cancellation).await?;
         Ok(())
     }
 
@@ -637,7 +637,11 @@ impl BackupTask {
     }
 
     /// 执行一小批到期保留清理：先处理 `expires_at` 已到期的记录，再按天数/份数清理计划备份
-    async fn run_retention_batch(&self, now: DateTime<Utc>) -> Result<(), WorkerTaskError> {
+    async fn run_retention_batch(
+        &self,
+        now: DateTime<Utc>,
+        cancellation: &CancellationToken,
+    ) -> Result<(), WorkerTaskError> {
         let settings = self.repository.load_settings().await.map_err(repo_error)?;
 
         // 1. expires_at 已到期的记录（手动或计划）无条件进入删除
@@ -659,18 +663,41 @@ impl BackupTask {
         }
 
         // 2. 计划备份按 retentionDays / retentionCount 清理
-        let records = self
-            .repository
-            .list_scheduled_completed_desc(RETENTION_SCAN_LIMIT)
-            .await
-            .map_err(repo_error)?;
-        let decisions = decide_retention(
-            settings.retention_days,
-            settings.retention_count,
-            now,
-            &records,
-        );
-        for decision in decisions.into_iter().take(RETENTION_BATCH_SIZE) {
+        if settings.retention_days == 0 && settings.retention_count == 0 {
+            return Ok(());
+        }
+        let mut offset = 0;
+        let mut decisions = Vec::new();
+        loop {
+            if cancellation.is_cancelled() {
+                return Ok(());
+            }
+            let records = self
+                .repository
+                .list_scheduled_completed_desc(offset, RETENTION_PAGE_SIZE)
+                .await
+                .map_err(repo_error)?;
+            let remaining = RETENTION_BATCH_SIZE - decisions.len();
+            decisions.extend(
+                decide_retention(
+                    settings.retention_days,
+                    settings.retention_count,
+                    now,
+                    offset,
+                    &records,
+                )
+                .into_iter()
+                .take(remaining),
+            );
+            if decisions.len() == RETENTION_BATCH_SIZE
+                || records.len() < RETENTION_PAGE_SIZE as usize
+            {
+                break;
+            }
+            offset += records.len() as u64;
+        }
+        // 收齐本批候选后再删除，避免删除前页记录导致后续分页跳过数据
+        for decision in decisions {
             let Some(record) = self
                 .repository
                 .transition_to_deleting(&decision.record_id, now, None)

backend/crates/gateway-admin/src/ports/backup.rs

@@ -136,9 +136,10 @@ pub trait BackupRepository: Send + Sync {
         timezone: &str,
     ) -> AdminStoreResult<bool>;
 
-    /// 按完成时间倒序读取 completed 计划备份，供保留策略扫描（最多 `limit` 条）
+    /// 按完成时间及 ID 倒序分页读取 completed 计划备份，offset 相对完整集合
     async fn list_scheduled_completed_desc(
         &self,
+        offset: u64,
         limit: u32,
     ) -> AdminStoreResult<Vec<BackupRecord>>;
 }
@@ -388,6 +389,7 @@ impl BackupRepository for UnavailableBackupRepository {
     }
     async fn list_scheduled_completed_desc(
         &self,
+        _offset: u64,
         _limit: u32,
     ) -> AdminStoreResult<Vec<BackupRecord>> {
         Err(disabled_store())

backend/crates/gateway-admin/tests/backup/policy.rs

@@ -106,7 +106,7 @@ fn retention_count_keeps_newest_n() {
             Some(base),
         ),
     ];
-    let decisions = decide_retention(0, 2, now, &records);
+    let decisions = decide_retention(0, 2, now, 0, &records);
     let ids: Vec<_> = decisions.iter().map(|d| d.record_id.as_str()).collect();
     assert_eq!(ids, vec!["oldest"]);
     assert_eq!(decisions[0].reason, RetentionReason::Count);
@@ -131,7 +131,7 @@ fn retention_days_deletes_old_scheduled_but_keeps_newest() {
             Some(now - Duration::days(40)),
         ),
     ];
-    let decisions = decide_retention(30, 0, now, &records);
+    let decisions = decide_retention(30, 0, now, 0, &records);
     let ids: Vec<_> = decisions.iter().map(|d| d.record_id.as_str()).collect();
     assert_eq!(ids, vec!["ancient"]);
 }
@@ -146,7 +146,7 @@ fn newest_scheduled_never_auto_deleted_even_when_ancient() {
         now - Duration::days(400),
         Some(now - Duration::days(400)),
     )];
-    assert!(decide_retention(30, 1, now, &records).is_empty());
+    assert!(decide_retention(30, 1, now, 0, &records).is_empty());
 }
 
 #[test]
@@ -168,7 +168,7 @@ fn disabled_thresholds_delete_nothing() {
             Some(now - Duration::days(90)),
         ),
     ];
-    assert!(decide_retention(0, 0, now, &records).is_empty());
+    assert!(decide_retention(0, 0, now, 0, &records).is_empty());
 }
 
 #[test]

backend/crates/gateway-admin/tests/backup/support.rs

@@ -470,7 +470,8 @@ impl BackupRepository for FakeBackupRepository {
 
     async fn list_scheduled_completed_desc(
         &self,
-        _limit: u32,
+        offset: u64,
+        limit: u32,
     ) -> AdminStoreResult<Vec<BackupRecord>> {
         let mut records = self
             .records
@@ -488,7 +489,11 @@ impl BackupRepository for FakeBackupRepository {
                 .cmp(&a.completed_at)
                 .then_with(|| b.id.cmp(&a.id))
         });
-        Ok(records)
+        Ok(records
+            .into_iter()
+            .skip(usize::try_from(offset).unwrap_or(usize::MAX))
+            .take(limit as usize)
+            .collect())
     }
 }
 

backend/crates/gateway-admin/tests/backup/task.rs

@@ -239,6 +239,75 @@ async fn retention_cleans_expired_scheduled_backups() {
     assert!(object_store.object(&new_seed.object_key).is_some());
 }
 
+#[tokio::test]
+async fn retention_scans_all_pages_and_deletes_only_one_batch_per_cycle() {
+    for (retention_days, retention_count, retained) in [
+        (0, 999, 999),
+        (0, 1000, 1000),
+        (0, 1001, 1001),
+        (30, 0, 1000),
+    ] {
+        let mut settings = configured_settings();
+        settings.retention_days = retention_days;
+        settings.retention_count = retention_count;
+        let repository = Arc::new(FakeBackupRepository::new(settings));
+        let object_store = Arc::new(FakeObjectStore::new());
+        let now = Utc::now();
+        let mut seeds = Vec::new();
+        for index in 0..retained + 15 {
+            let completed_at = if retention_days > 0 && index >= retained {
+                now - Duration::days(40) - Duration::minutes(index)
+            } else {
+                now - Duration::minutes(index)
+            };
+            let seed = BackupRecordSeed {
+                id: backup_id(&format!("retention-{index}")),
+                trigger_kind: BackupTriggerKind::Scheduled,
+                scheduled_at: Some(completed_at),
+                object_key: format!("codex/retention-{index}.dump"),
+                expires_at: None,
+            };
+            repository
+                .insert_backup_record(seed.clone(), &system_context())
+                .await
+                .expect("insert scheduled backup");
+            repository.set_completed(&seed.id, completed_at);
+            object_store
+                .objects
+                .lock()
+                .expect("objects")
+                .insert(seed.object_key.clone(), DUMP_CONTENT.to_vec());
+            seeds.push(seed);
+        }
+        let task = BackupTask::new(
+            repository.clone(),
+            Arc::new(FakeDumpPort::new()),
+            object_store.clone(),
+        );
+
+        for deleted in [10, 15] {
+            task.run_cycle(&CancellationToken::new())
+                .await
+                .expect("run retention cycle");
+            let records = repository.all_records();
+            assert_eq!(records.len(), seeds.len() - deleted);
+            for (index, seed) in seeds.iter().enumerate() {
+                let retained = usize::try_from(retained).unwrap();
+                let should_remain = index < retained || index >= retained + deleted;
+                assert_eq!(
+                    records.iter().any(|record| record.id == seed.id),
+                    should_remain,
+                    "record {index}, retention days {retention_days}, count {retention_count}"
+                );
+                assert_eq!(
+                    object_store.object(&seed.object_key).is_some(),
+                    should_remain
+                );
+            }
+        }
+    }
+}
+
 #[tokio::test]
 async fn timezone_change_rebases_future_cursor_without_running_old_due_backup() {
     let mut settings = configured_settings();

backend/crates/gateway-store/src/postgres/backup.rs

@@ -657,6 +657,7 @@ impl BackupRepository for PgBackupRepository {
 
     async fn list_scheduled_completed_desc(
         &self,
+        offset: u64,
         limit: u32,
     ) -> AdminStoreResult<Vec<BackupRecord>> {
         let limit = limit.clamp(1, 10000);
@@ -667,9 +668,13 @@ impl BackupRepository for PgBackupRepository {
                from backup_records
               where trigger_kind = 'scheduled' and status = 'completed'
               order by completed_at desc, id desc
-              limit $1",
+              limit $1 offset $2",
         )
         .bind(i64::from(limit))
+        .bind(
+            i64::try_from(offset)
+                .map_err(|source| store_invalid("page offset").with_source(source))?,
+        )
         .fetch_all(&self.pool)
         .await
         .map_err(|source| {

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

@@ -78,6 +78,63 @@ async fn enable_verified_schedule(repository: &PgBackupRepository) -> BackupSett
         .expect("enable schedule")
 }
 
+#[tokio::test]
+async fn scheduled_completed_pages_keep_stable_order_and_exclude_other_records() {
+    let Some(db) = TestDatabase::create("backup_retention_pages").await else {
+        return;
+    };
+    let repository = PgBackupRepository::new(db.pool.clone());
+    let now = Utc::now();
+    sqlx::query(
+        "insert into backup_records (
+            id, trigger_kind, status, scheduled_at, object_key, size_bytes, sha256,
+            error_code, error_message, started_at, completed_at, created_at, updated_at
+         )
+         select 'backup_' || lpad(to_hex(n), 32, '0'),
+                case when n = 1006 then 'manual' else 'scheduled' end,
+                case when n = 1007 then 'failed' else 'completed' end,
+                case when n = 1006 then null else $1 - n * interval '1 minute' end,
+                'codex/' || n || '.dump', 1, repeat('a', 64),
+                case when n = 1007 then 'backup.pg_dump_failed' end,
+                case when n = 1007 then '导出失败' end,
+                $1 - interval '2 days', $1, $1 - interval '2 days', $1
+           from generate_series(1, 1007) as n",
+    )
+    .bind(now)
+    .execute(&db.pool)
+    .await
+    .expect("seed completed backups with tied completion times");
+
+    let first = repository
+        .list_scheduled_completed_desc(0, 1000)
+        .await
+        .expect("first page");
+    let second = repository
+        .list_scheduled_completed_desc(1000, 1000)
+        .await
+        .expect("second page");
+    assert_eq!(first.len(), 1000);
+    assert_eq!(second.len(), 5);
+    let ids: Vec<_> = first
+        .iter()
+        .chain(&second)
+        .map(|record| record.id.clone())
+        .collect();
+    let expected: Vec<_> = (1..=1005)
+        .rev()
+        .map(|index| format!("backup_{index:032x}"))
+        .collect();
+    assert_eq!(ids, expected);
+    assert!(
+        repository
+            .list_scheduled_completed_desc(1005, 1000)
+            .await
+            .expect("past final page")
+            .is_empty()
+    );
+    db.close().await;
+}
+
 #[tokio::test]
 async fn storage_rotation_pauses_schedule_and_rejects_stale_verification() {
     let Some(db) = TestDatabase::create("backup_rotation").await else {