From 0e1c609cd92d98dd5cbffa7ba1bb187f83fb26d6 Mon Sep 17 00:00:00 2001 From: Yuqi Xiong <116958528+Mr-xiongyq@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:43:11 +0800 Subject: [PATCH] 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 --- .../crates/gateway-admin/src/backup/policy.rs | 10 +-- .../crates/gateway-admin/src/backup/task.rs | 59 +++++++++++----- .../crates/gateway-admin/src/ports/backup.rs | 4 +- .../gateway-admin/tests/backup/policy.rs | 8 +-- .../gateway-admin/tests/backup/support.rs | 9 ++- .../crates/gateway-admin/tests/backup/task.rs | 69 +++++++++++++++++++ .../gateway-store/src/postgres/backup.rs | 7 +- .../gateway-store/tests/postgres/backup.rs | 57 +++++++++++++++ 8 files changed, 195 insertions(+), 28 deletions(-) diff --git a/backend/crates/gateway-admin/src/backup/policy.rs b/backend/crates/gateway-admin/src/backup/policy.rs index ead47eb9d..2acadbca4 100644 --- a/backend/crates/gateway-admin/src/backup/policy.rs +++ b/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, + offset: u64, scheduled_completed_desc: &[BackupRecord], ) -> Vec { 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(), diff --git a/backend/crates/gateway-admin/src/backup/task.rs b/backend/crates/gateway-admin/src/backup/task.rs index 85284fa85..7eed3f312 100644 --- a/backend/crates/gateway-admin/src/backup/task.rs +++ b/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) -> Result<(), WorkerTaskError> { + async fn run_retention_batch( + &self, + now: DateTime, + 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) diff --git a/backend/crates/gateway-admin/src/ports/backup.rs b/backend/crates/gateway-admin/src/ports/backup.rs index 7444af434..b42b5fe50 100644 --- a/backend/crates/gateway-admin/src/ports/backup.rs +++ b/backend/crates/gateway-admin/src/ports/backup.rs @@ -136,9 +136,10 @@ pub trait BackupRepository: Send + Sync { timezone: &str, ) -> AdminStoreResult; - /// 按完成时间倒序读取 completed 计划备份，供保留策略扫描（最多 `limit` 条） + /// 按完成时间及 ID 倒序分页读取 completed 计划备份，offset 相对完整集合 async fn list_scheduled_completed_desc( &self, + offset: u64, limit: u32, ) -> AdminStoreResult>; } @@ -388,6 +389,7 @@ impl BackupRepository for UnavailableBackupRepository { } async fn list_scheduled_completed_desc( &self, + _offset: u64, _limit: u32, ) -> AdminStoreResult> { Err(disabled_store()) diff --git a/backend/crates/gateway-admin/tests/backup/policy.rs b/backend/crates/gateway-admin/tests/backup/policy.rs index 0f7a155e7..ffe0f250c 100644 --- a/backend/crates/gateway-admin/tests/backup/policy.rs +++ b/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] diff --git a/backend/crates/gateway-admin/tests/backup/support.rs b/backend/crates/gateway-admin/tests/backup/support.rs index 1c5b0c177..79edb3b02 100644 --- a/backend/crates/gateway-admin/tests/backup/support.rs +++ b/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> { 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()) } } diff --git a/backend/crates/gateway-admin/tests/backup/task.rs b/backend/crates/gateway-admin/tests/backup/task.rs index ed2f809f9..5fd1d2469 100644 --- a/backend/crates/gateway-admin/tests/backup/task.rs +++ b/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(); diff --git a/backend/crates/gateway-store/src/postgres/backup.rs b/backend/crates/gateway-store/src/postgres/backup.rs index f1979de76..e2b11b749 100644 --- a/backend/crates/gateway-store/src/postgres/backup.rs +++ b/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> { 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| { diff --git a/backend/crates/gateway-store/tests/postgres/backup.rs b/backend/crates/gateway-store/tests/postgres/backup.rs index f3dd77368..ac65520da 100644 --- a/backend/crates/gateway-store/tests/postgres/backup.rs +++ b/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 {