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 {