diff --git a/benchmarks/aft-search/slice-descriptors/land-retention-census-backoff.json b/benchmarks/aft-search/slice-descriptors/land-retention-census-backoff.json new file mode 100644 index 000000000..f65eac5af --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-retention-census-backoff.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "retention-census-backoff" + ], + "note": "Storage retention backoff and scale test; no changed path matches RANKING_FENCE_PREFIXES." +} diff --git a/crates/aft/src/storage_retention.rs b/crates/aft/src/storage_retention.rs index 3dd128247..7e0dfeef3 100644 --- a/crates/aft/src/storage_retention.rs +++ b/crates/aft/src/storage_retention.rs @@ -132,6 +132,8 @@ pub(crate) struct TrashReport { /// Work examined by the pass's read-only durable-directory census. pub protection_census: crate::bash_background::persistence::IdentityCensus, pub protection_failure: Option, + /// A failed protection decision remains visible while reaping backs off. + pub protection_retry_at_ms: Option, /// The durable-storage census (directory identities no trash removal may /// touch), taken at most once per reaper pass and shared by every trash /// container that pass visits. Taking it again for each container spent @@ -248,6 +250,34 @@ struct State { reaping: bool, trash: TrashReport, reaper_cancelled: Option bool + Send + Sync>>, + protection_failures: u32, + protection_retry: Option, +} + +impl State { + fn record_trash(&mut self, mut report: TrashReport, now: Instant) { + if report.protection_failure.is_some() { + self.protection_failures = self.protection_failures.saturating_add(1); + // Skip at least one scheduled pass after a failed census. Repeated + // failures reach an hourly retry without spending CPU every interval. + let delay = (INTERVAL * 2u32.saturating_pow(self.protection_failures.min(3))) + .min(Duration::from_secs(60 * 60)); + self.protection_retry = Some(now + delay); + report.protection_retry_at_ms = + Some(crate::pins::now_ms().saturating_add(delay.as_millis() as u64)); + } else { + self.protection_failures = 0; + self.protection_retry = None; + } + self.trash = report.clone(); + if let Some(sweep) = &mut self.report { + sweep.trash = report; + } + } + + fn reaper_ready(&self, now: Instant) -> bool { + !self.reaping && self.protection_retry.is_none_or(|retry| now >= retry) + } } struct TrashScan { @@ -3393,19 +3423,19 @@ fn reap_retired(storage: &StoragePaths) -> io::Result<()> { report.keep(&storage.artifacts, "reaper_pass", error); } } + let report = { + let mut states = states() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let state = states.entry(storage.artifacts.clone()).or_default(); + state.record_trash(report, Instant::now()); + state.trash.clone() + }; crate::slog_info!( "storage retention trash root={} report={}", storage.artifacts.display(), serde_json::to_string(&report).unwrap_or_default() ); - let mut states = states() - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let state = states.entry(storage.artifacts.clone()).or_default(); - state.trash = report.clone(); - if let Some(sweep) = &mut state.report { - sweep.trash = report; - } result } @@ -3518,18 +3548,18 @@ fn reap_retired_with_budget( } fn schedule_reaper(storage: &StoragePaths) { - #[cfg(test)] - test_hook(&storage.artifacts, "reaper-launch"); { let mut states = states() .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let state = states.entry(storage.artifacts.clone()).or_default(); - if state.reaping { + if !state.reaper_ready(Instant::now()) { return; } state.reaping = true; } + #[cfg(test)] + test_hook(&storage.artifacts, "reaper-launch"); let storage = storage.clone(); let worker_storage = storage.clone(); let spawned = std::thread::Builder::new() @@ -5688,8 +5718,44 @@ fn record_v3_removal(domain: &mut DomainReport, bytes: Option) { #[cfg(test)] mod storage_retention_tests { + #[test] + fn retention_protection_failure_backs_off_and_stays_in_health() { + let now = Instant::now(); + let mut state = State { + report: Some(SweepReport::default()), + ..Default::default() + }; + for attempt in 1..=5 { + state.record_trash( + TrashReport { + incomplete: true, + protection_failure: Some(ProtectionFailure::Budget), + ..Default::default() + }, + now, + ); + assert!(!state.reaper_ready(now + INTERVAL)); + let retry = state.protection_retry.unwrap(); + assert!(retry <= now + Duration::from_secs(60 * 60)); + assert!(state.reaper_ready(retry)); + assert_eq!(state.protection_failures, attempt); + let health = serde_json::to_value(state.report.as_ref().unwrap()).unwrap(); + assert_eq!(health["trash"]["incomplete"], true); + assert_eq!(health["trash"]["protection_failure"], "budget"); + assert!(health["trash"]["protection_retry_at_ms"].as_u64().is_some()); + } + state.record_trash(TrashReport::default(), now); + assert!(state.reaper_ready(now)); + assert_eq!(state.protection_failures, 0); + assert!(state.trash.protection_retry_at_ms.is_none()); + } + #[test] fn retention_protection_large_task_store_fixture() { + let stress = std::env::var("AFT_STRESS").as_deref() == Ok("1"); + let tasks = if stress { 30_000 } else { 300 }; + let sessions = if stress { 1_372 } else { 14 }; + let files = tasks * 11 + tasks * 2 / 3; let temp = tempfile::tempdir().unwrap(); let storage = StoragePaths::same(temp.path()); fs::create_dir_all(temp.path().join("retention/bindings")).unwrap(); @@ -5701,8 +5767,10 @@ mod storage_retention_tests { ) .unwrap(); } - for task in 0..5_000 { - let directory = temp.path().join(format!("tasks/session/{task:032x}")); + for task in 0..tasks { + let directory = temp + .path() + .join(format!("tasks/session-{:04}/{task:032x}", task % sessions)); fs::create_dir_all(directory.join("control")).unwrap(); fs::create_dir_all(directory.join("io")).unwrap(); for name in crate::bash_background::persistence::readable_task_file_names( @@ -5710,6 +5778,9 @@ mod storage_retention_tests { ) { fs::write(directory.join(name), b"finished task artifact").unwrap(); } + if task < tasks * 2 / 3 { + fs::write(directory.join("control/environment.bin"), b"environment").unwrap(); + } } for entry in 0..50 { let domain = if entry % 2 == 0 { "index" } else { "inspect" }; @@ -5718,6 +5789,41 @@ mod storage_retention_tests { fs::write(source.join("payload"), b"retired cache").unwrap(); retire_directory(temp.path(), &source).unwrap(); } + // Even the default fixture exceeds a small census allowance. An + // incomplete protection set must authorize no deletion, then a fresh + // pass must be able to reclaim the same prefetched trash entries. + let mut cutoff = TrashReport::default(); + assert!(reap_trash_with_snapshot_budget( + &storage, + &temp.path().join("index/.retention-trash"), + false, + Instant::now() + Duration::from_secs(60), + &mut crate::bash_background::persistence::RepairBudget::with_entries(500), + &mut cutoff, + NO_TIME_BOUND, + ) + .unwrap()); + assert_eq!(cutoff.protection_failure, Some(ProtectionFailure::Budget)); + assert!(cutoff.incomplete); + assert_eq!(cutoff.removed_entries, 0); + let mut state = State::default(); + let failed_at = Instant::now(); + state.record_trash(cutoff, failed_at); + assert!(!state.reaper_ready(failed_at + INTERVAL)); + assert!(state.reaper_ready(state.protection_retry.unwrap())); + for domain in ["index", "inspect"] { + let entries = fs::read_dir(temp.path().join(domain).join(".retention-trash")) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(entries.len(), 25); + for entry in entries { + assert_eq!( + fs::read(entry.path().join("payload")).unwrap(), + b"retired cache" + ); + } + } let deadline = Instant::now() + Duration::from_secs(60); let mut budget = crate::bash_background::persistence::RepairBudget::new(); let mut report = TrashReport::default(); @@ -5732,6 +5838,8 @@ mod storage_retention_tests { }), ); let started = Instant::now(); + #[cfg(any(target_os = "linux", target_os = "macos"))] + let cpu_started = thread_cpu_seconds(); for domain in ["index", "inspect"] { reap_trash_with_snapshot_budget( &storage, @@ -5745,14 +5853,36 @@ mod storage_retention_tests { .unwrap(); } eprintln!( - "retention fixture: roots=200 finished_tasks=5000 trash=50 elapsed_ms={:.3} report={}", + "retention fixture: roots=200 finished_tasks={tasks} sessions={sessions} files={files} trash=50 elapsed_ms={:.3} report={}", started.elapsed().as_secs_f64() * 1_000.0, serde_json::to_string(&report).unwrap(), ); + #[cfg(any(target_os = "linux", target_os = "macos"))] + eprintln!( + "retention fixture: thread_cpu_ms={:.3}", + (thread_cpu_seconds() - cpu_started) * 1_000.0 + ); assert_eq!(report.removed_entries, 50, "{report:?}"); assert!(!report.incomplete, "{report:?}"); assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); - assert!((15_000..15_010).contains(&report.protection_census.directories)); + assert!(report.protection_census.directories >= tasks * 3); + state.record_trash(report, Instant::now()); + assert!(state.reaper_ready(Instant::now())); + assert!(state.protection_retry.is_none()); + assert!(state.trash.protection_retry_at_ms.is_none()); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + fn thread_cpu_seconds() -> f64 { + let mut time = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + assert_eq!( + unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut time) }, + 0 + ); + time.tv_sec as f64 + time.tv_nsec as f64 / 1_000_000_000.0 } #[test] @@ -5822,9 +5952,29 @@ mod storage_retention_tests { assert_eq!(report.kept_by_reason.get("protections_budget"), Some(&1)); assert_eq!(report.removed_entries, 0); assert!(report.incomplete); - for entry in entries { + for entry in &entries { assert!(entry.join("payload").exists()); } + let storage = StoragePaths::same(temp.path()); + states() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entry(storage.artifacts.clone()) + .or_default() + .record_trash(report, Instant::now()); + let launches = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let observed = launches.clone(); + let _hook = observe_test_hook( + temp.path(), + Arc::new(move |step| { + if step == "reaper-launch" { + observed.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + }), + ); + schedule_reaper(&storage); + schedule_reaper(&storage); + assert_eq!(launches.load(std::sync::atomic::Ordering::SeqCst), 0); } #[test] @@ -5887,11 +6037,43 @@ mod storage_retention_tests { #[test] fn retention_protection_keeps_a_nested_live_task_alias() { - use crate::bash_background::persistence::{PinnedDir, REPAIR_IDENTITIES}; + assert_nested_task_alias_protected(false); + } + + #[test] + fn retention_protection_keeps_a_nested_finished_task_alias() { + assert_nested_task_alias_protected(true); + } + + fn assert_nested_task_alias_protected(finished: bool) { + use crate::bash_background::persistence::{PersistedTask, PinnedDir, REPAIR_IDENTITIES}; + use crate::bash_background::BgTaskStatus; let temp = tempfile::tempdir().unwrap(); - let task = temp.path().join("tasks/session/live-task"); - fs::create_dir_all(&task).unwrap(); - fs::write(task.join("task.json"), b"running task").unwrap(); + let task = temp.path().join("tasks/session").join("a".repeat(32)); + fs::create_dir_all(task.join("control")).unwrap(); + fs::create_dir_all(task.join("io")).unwrap(); + let mut metadata = PersistedTask::starting( + "a".repeat(32), + "session".into(), + "echo output".into(), + temp.path().into(), + None, + None, + false, + false, + ); + metadata.status = if finished { + BgTaskStatus::Completed + } else { + BgTaskStatus::Running + }; + if finished { + metadata.finished_at = Some(crate::pins::now_ms()); + metadata.exit_code = Some(0); + } + let metadata = serde_json::to_vec(&metadata).unwrap(); + fs::write(task.join("control/metadata.json"), &metadata).unwrap(); + fs::write(task.join("io/stdout"), b"task output").unwrap(); let source = temp.path().join("views/0123456789abcdef"); fs::create_dir_all(source.join("nested")).unwrap(); fs::write(source.join("nested/output"), b"task output").unwrap(); @@ -5910,7 +6092,7 @@ mod storage_retention_tests { entry.join("nested"), fs::canonicalize(entry.join("nested")).unwrap(), ]; - let task_id = PinnedDir::open(&task) + let task_id = PinnedDir::open(&task.join("io")) .unwrap() .directory_identity() .unwrap(); @@ -5945,7 +6127,11 @@ mod storage_retention_tests { fs::read(entry.join("nested/output")).unwrap(), b"task output" ); - assert_eq!(fs::read(task.join("task.json")).unwrap(), b"running task"); + assert_eq!( + fs::read(task.join("control/metadata.json")).unwrap(), + metadata + ); + assert_eq!(fs::read(task.join("io/stdout")).unwrap(), b"task output"); } fn followup_trash_fixture(count: usize) -> (tempfile::TempDir, Vec) {