From 9f51f37a0dd5d3369f8b812918b8cc16ea23405e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Fri, 21 Aug 2026 22:04:23 +0800 Subject: [PATCH] fix(heal): make resume checkpoints crash consistent --- crates/heal/src/heal/erasure_healer.rs | 5 + crates/heal/src/heal/resume.rs | 1 + crates/heal/src/heal/resume/checkpoint.rs | 249 +++++++++++++++++++-- crates/heal/src/heal/resume/tests.rs | 258 ++++++++++++++++++++++ crates/heal/src/heal/resume/utils.rs | 3 +- 5 files changed, 497 insertions(+), 19 deletions(-) diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 920fffcf5..a0ace01b4 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -373,6 +373,11 @@ impl ErasureSetHealer { set_disk_id: &str, buckets: &[String], ) -> Result<(ResumeManager, CheckpointManager)> { + if self.replacement_task_id.is_none() && CheckpointManager::is_blocked(&self.disk, task_id).await { + return Err(Error::TaskExecutionFailed { + message: format!("Resume task {task_id} has a blocked checkpoint"), + }); + } // check if resume state exists let has_resume_state = if self.replacement_task_id.is_some() { ResumeManager::has_replacement_intent(&self.disk, task_id).await diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 291132c12..239713d3a 100644 --- a/crates/heal/src/heal/resume.rs +++ b/crates/heal/src/heal/resume.rs @@ -51,6 +51,7 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json"; const REPLACEMENT_INTENT_FILE: &str = "ahm_replacement_intent.json"; const RESUME_PROGRESS_FILE: &str = "ahm_progress.json"; pub(super) const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json"; +pub(super) const RESUME_CHECKPOINT_BLOCKED_FILE: &str = "ahm_checkpoint.blocked"; const REPLACEMENT_COMPLETION_PROOF_FILE: &str = "ahm_replacement_completion_proof.json"; const REPLACEMENT_RECOVERY_DIR: &str = "ahm-replacement"; const REPLACEMENT_INTENT_SEAL_FILE: &str = "ahm_replacement_intent_seal"; diff --git a/crates/heal/src/heal/resume/checkpoint.rs b/crates/heal/src/heal/resume/checkpoint.rs index 1b4b7ece3..584b13746 100644 --- a/crates/heal/src/heal/resume/checkpoint.rs +++ b/crates/heal/src/heal/resume/checkpoint.rs @@ -18,13 +18,14 @@ use std::collections::HashSet; use std::path::Path; use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex as AsyncMutex, RwLock}; use tracing::{debug, warn}; -use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET}; +use super::super::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes}; +use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt, RUSTFS_META_BUCKET}; use super::{ - LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_FILE, delete_resume_file, path_to_str, - validate_resume_task_id, + LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_BLOCKED_FILE, RESUME_CHECKPOINT_FILE, + delete_resume_file, path_to_str, validate_resume_task_id, }; const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state"; @@ -116,17 +117,108 @@ pub struct CheckpointManager { disk: DiskStore, checkpoint: Arc>, throttle: Mutex, + save_lock: AsyncMutex<()>, + last_saved: Mutex>, } impl CheckpointManager { + fn blocked_path(task_id: &str) -> std::path::PathBuf { + Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}")) + } + + /// Return whether a checkpoint was permanently isolated after a malformed + /// or unsupported snapshot was observed. + pub(crate) async fn is_blocked(disk: &DiskStore, task_id: &str) -> bool { + if validate_resume_task_id(task_id).is_err() { + return false; + } + let blocked_path = Self::blocked_path(task_id); + let Ok(path) = path_to_str(&blocked_path) else { + return false; + }; + match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path).await { + Ok(_) => true, + Err(crate::heal::DiskError::FileNotFound) => false, + Err(_) => true, + } + } + + /// Validate the checkpoint while enumerating resumable state. This reads + /// the checkpoint once and also isolates malformed or unsupported data. + pub(crate) async fn is_resumable(disk: &DiskStore, task_id: &str) -> bool { + if validate_resume_task_id(task_id).is_err() || Self::is_blocked(disk, task_id).await { + return false; + } + let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}")); + let Ok(path) = path_to_str(&file_path) else { + return false; + }; + match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path).await { + Ok(bytes) if bytes.is_empty() => true, + Ok(bytes) => Self::load_from_data(disk.clone(), task_id, bytes.to_vec()).await.is_ok(), + Err(crate::heal::DiskError::FileNotFound) => true, + Err(_) => false, + } + } + + async fn block_invalid_snapshot(disk: &DiskStore, task_id: &str) { + // This marker is intentionally version-agnostic: an unsupported reader + // must stop selector retries until an operator cleans up the snapshot. + let blocked_path = Self::blocked_path(task_id); + let Ok(path) = path_to_str(&blocked_path) else { + return; + }; + let result = EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + path, + None, + Some(EcstoreDiskBytes::from_static(b"blocked")), + ) + .await; + match result { + Ok(EcstoreConditionalFileUpdate::Updated | EcstoreConditionalFileUpdate::Mismatch) => {} + Ok(EcstoreConditionalFileUpdate::Missing) => warn!( + target: "rustfs::heal::resume", + event = EVENT_HEAL_CHECKPOINT_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_RESUME, + task_id, + state = "blocked_marker_write_failed", + error = "marker target disappeared", + "Heal checkpoint could not persist its blocked marker" + ), + Err(error) => warn!( + target: "rustfs::heal::resume", + event = EVENT_HEAL_CHECKPOINT_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_RESUME, + task_id, + state = "blocked_marker_write_failed", + error = %error, + "Heal checkpoint could not persist its blocked marker" + ), + } + } + /// create new checkpoint manager pub async fn new(disk: DiskStore, task_id: String) -> Result { validate_resume_task_id(&task_id)?; + let checkpoint_volume = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}"); + if let Err(error) = EcstoreDiskAPI::make_volume(disk.as_ref(), &checkpoint_volume).await + && error != crate::heal::DiskError::VolumeExists + { + return Err(Error::TaskExecutionFailed { + message: format!("Failed to create checkpoint volume: {error}"), + }); + } let checkpoint = ResumeCheckpoint::new(task_id); let manager = Self { disk, checkpoint: Arc::new(RwLock::new(checkpoint)), throttle: Mutex::new(PersistThrottle::new()), + save_lock: AsyncMutex::new(()), + last_saved: Mutex::new(None), }; // save initial checkpoint @@ -140,6 +232,7 @@ impl CheckpointManager { error = %e, "Heal checkpoint persistence failed" ); + return Err(e); } Ok(manager) } @@ -148,11 +241,22 @@ impl CheckpointManager { pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result { validate_resume_task_id(task_id)?; let checkpoint_data = Self::read_checkpoint_file(&disk, task_id).await?; - let mut checkpoint: ResumeCheckpoint = - serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed { - message: format!("Failed to deserialize checkpoint: {e}"), - })?; + Self::load_from_data(disk, task_id, checkpoint_data).await + } + + async fn load_from_data(disk: DiskStore, task_id: &str, checkpoint_data: Vec) -> Result { + validate_resume_task_id(task_id)?; + let mut checkpoint: ResumeCheckpoint = match serde_json::from_slice(&checkpoint_data) { + Ok(checkpoint) => checkpoint, + Err(error) => { + Self::block_invalid_snapshot(&disk, task_id).await; + return Err(Error::TaskExecutionFailed { + message: format!("Failed to deserialize checkpoint: {error}"), + }); + } + }; if checkpoint.task_id != task_id { + Self::block_invalid_snapshot(&disk, task_id).await; return Err(Error::TaskExecutionFailed { message: "Resume checkpoint task id does not match filename".to_string(), }); @@ -163,6 +267,7 @@ impl CheckpointManager { // identities. Discard the stale sets and position, then stamp the // current schema so the scan restarts cleanly. if checkpoint.schema_version > CURRENT_CHECKPOINT_SCHEMA { + Self::block_invalid_snapshot(&disk, task_id).await; return Err(Error::TaskExecutionFailed { message: format!( "Checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}", @@ -194,6 +299,8 @@ impl CheckpointManager { disk, checkpoint: Arc::new(RwLock::new(checkpoint)), throttle: Mutex::new(PersistThrottle::new()), + save_lock: AsyncMutex::new(()), + last_saved: Mutex::new(Some(EcstoreDiskBytes::from(checkpoint_data))), }) } @@ -204,7 +311,7 @@ impl CheckpointManager { } let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}")); match path_to_str(&file_path) { - Ok(path_str) => match disk.read_all(RUSTFS_META_BUCKET, path_str).await { + Ok(path_str) => match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path_str).await { Ok(data) => !data.is_empty(), Err(_) => false, }, @@ -292,6 +399,7 @@ impl CheckpointManager { let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}")); delete_resume_file(&self.disk, &checkpoint_file).await?; + delete_resume_file(&self.disk, &Self::blocked_path(&task_id)).await?; debug!( target: "rustfs::heal::resume", @@ -307,21 +415,126 @@ impl CheckpointManager { /// save checkpoint to disk async fn save_checkpoint(&self) -> Result<()> { - let checkpoint = self.checkpoint.read().await; + // Serialize saves and take the snapshot only after acquiring the lock: + // a slower writer must not publish a snapshot taken before a newer one. + let _save_guard = self.save_lock.lock().await; + let checkpoint = self.checkpoint.read().await.clone(); validate_resume_task_id(&checkpoint.task_id)?; - let checkpoint_data = serde_json::to_vec(&*checkpoint).map_err(|e| Error::TaskExecutionFailed { - message: format!("Failed to serialize checkpoint: {e}"), - })?; + let checkpoint_data = + EcstoreDiskBytes::from(serde_json::to_vec(&checkpoint).map_err(|e| Error::TaskExecutionFailed { + message: format!("Failed to serialize checkpoint: {e}"), + })?); let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{}_{}", checkpoint.task_id, RESUME_CHECKPOINT_FILE)); let path_str = path_to_str(&file_path)?; - self.disk - .write_all(RUSTFS_META_BUCKET, path_str, checkpoint_data.into()) + let last_saved = self + .last_saved + .lock() + .map_err(|_| Error::TaskExecutionFailed { + message: "Checkpoint save state lock is poisoned; refusing to save".to_string(), + })? + .clone(); + let update = EcstoreDiskAPI::compare_and_update_file( + self.disk.as_ref(), + RUSTFS_META_BUCKET, + path_str, + last_saved.clone(), + Some(checkpoint_data.clone()), + ) + .await + .map_err(|e| Error::TaskExecutionFailed { + message: format!("Failed to save checkpoint: {e}"), + })?; + + let expected = match update { + EcstoreConditionalFileUpdate::Updated => None, + EcstoreConditionalFileUpdate::Missing => { + return Err(Error::TaskExecutionFailed { + message: "Checkpoint was removed after this manager saved it; refusing to recreate it".to_string(), + }); + } + EcstoreConditionalFileUpdate::Mismatch => { + // A healthy manager normally completes the CAS above without + // another read or JSON parse. Inspect only after a mismatch so + // corruption and future schemas cannot be overwritten blindly. + let existing = match HealDiskExt::read_all(self.disk.as_ref(), RUSTFS_META_BUCKET, path_str).await { + Ok(existing) => existing, + Err(crate::heal::DiskError::FileNotFound) => { + return Err(Error::TaskExecutionFailed { + message: "Checkpoint was removed after this manager saved it; refusing to recreate it".to_string(), + }); + } + Err(error) => { + return Err(Error::TaskExecutionFailed { + message: format!("Failed to inspect checkpoint after CAS mismatch: {error}"), + }); + } + }; + + if existing.is_empty() && last_saved.is_none() { + Some(existing) + } else { + let current: ResumeCheckpoint = match serde_json::from_slice(&existing) { + Ok(current) => current, + Err(error) => { + Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await; + return Err(Error::TaskExecutionFailed { + message: format!("Existing checkpoint is corrupt: {error}"), + }); + } + }; + if current.task_id != checkpoint.task_id { + Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await; + return Err(Error::TaskExecutionFailed { + message: "Existing checkpoint task id does not match filename".to_string(), + }); + } + if current.schema_version > CURRENT_CHECKPOINT_SCHEMA { + Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await; + return Err(Error::TaskExecutionFailed { + message: format!( + "Existing checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}", + current.schema_version + ), + }); + } + if last_saved.as_ref().is_none_or(|saved| saved.as_ref() != existing.as_ref()) { + return Err(Error::TaskExecutionFailed { + message: "Checkpoint changed since this manager loaded it; refusing to overwrite newer progress" + .to_string(), + }); + } + Some(existing) + } + } + }; + + if let Some(expected) = expected { + match EcstoreDiskAPI::compare_and_update_file( + self.disk.as_ref(), + RUSTFS_META_BUCKET, + path_str, + Some(expected), + Some(checkpoint_data.clone()), + ) .await .map_err(|e| Error::TaskExecutionFailed { - message: format!("Failed to save checkpoint: {e}"), - })?; + message: format!("Failed to save checkpoint after CAS mismatch: {e}"), + })? { + EcstoreConditionalFileUpdate::Updated => {} + EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch => { + return Err(Error::TaskExecutionFailed { + message: "Checkpoint changed while saving; refusing to overwrite newer progress".to_string(), + }); + } + } + } + + let mut last_saved = self.last_saved.lock().map_err(|_| Error::TaskExecutionFailed { + message: "Checkpoint save state lock is poisoned after save".to_string(), + })?; + *last_saved = Some(checkpoint_data); debug!( target: "rustfs::heal::resume", @@ -341,7 +554,7 @@ impl CheckpointManager { let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}")); let path_str = path_to_str(&file_path)?; - disk.read_all(RUSTFS_META_BUCKET, path_str) + HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path_str) .await .map(|bytes| bytes.to_vec()) .map_err(|e| Error::TaskExecutionFailed { diff --git a/crates/heal/src/heal/resume/tests.rs b/crates/heal/src/heal/resume/tests.rs index 8d8e35cfa..d8331b93e 100644 --- a/crates/heal/src/heal/resume/tests.rs +++ b/crates/heal/src/heal/resume/tests.rs @@ -1675,6 +1675,264 @@ async fn future_resume_and_checkpoint_schemas_are_rejected() { temp_dir.close().expect("remove schema test directory"); } +#[tokio::test] +async fn checkpoint_save_does_not_replace_a_non_empty_truncated_snapshot() { + let (temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let manager = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create checkpoint manager"); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + let truncated = b"{\"schema_version\":5,\"task_id\":"; + disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, truncated.as_slice().into()) + .await + .expect("write truncated checkpoint fixture"); + + let error = manager + .update_position(2, 7) + .await + .expect_err("a truncated checkpoint must fail closed during save"); + assert!(error.to_string().contains("Existing checkpoint is corrupt")); + assert_eq!( + disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path) + .await + .expect("read truncated checkpoint fixture"), + truncated.as_slice() + ); + assert!(CheckpointManager::is_blocked(&disk, &task_id).await); + temp_dir.close().expect("remove checkpoint save test directory"); +} + +#[tokio::test] +async fn checkpoint_save_does_not_replace_a_future_schema_snapshot() { + let (temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let manager = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create checkpoint manager"); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + let mut future = ResumeCheckpoint::new(task_id.clone()); + future.schema_version = CURRENT_CHECKPOINT_SCHEMA + 1; + let future_bytes = serde_json::to_vec(&future).expect("serialize future checkpoint fixture"); + disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, future_bytes.clone().into()) + .await + .expect("write future checkpoint fixture"); + + let error = manager + .update_position(2, 7) + .await + .expect_err("a future schema must fail closed during save"); + assert!(error.to_string().contains("Existing checkpoint schema")); + assert_eq!( + disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path) + .await + .expect("read future checkpoint fixture"), + future_bytes + ); + assert!(CheckpointManager::is_blocked(&disk, &task_id).await); + temp_dir.close().expect("remove future schema test directory"); +} + +#[tokio::test] +async fn new_checkpoint_manager_rebuilds_an_empty_snapshot() { + let (temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, EcstoreDiskBytes::new()) + .await + .expect("write empty checkpoint fixture"); + + let manager = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("a new manager must rebuild an empty checkpoint"); + manager + .update_position(3, 11) + .await + .expect("rebuilt checkpoint must remain writable"); + assert!(CheckpointManager::has_checkpoint(&disk, &task_id).await); + temp_dir.close().expect("remove empty checkpoint test directory"); +} + +#[tokio::test] +async fn deleted_checkpoint_is_not_recreated_by_an_old_manager() { + let (temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let manager = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create checkpoint manager"); + manager.cleanup().await.expect("delete checkpoint fixture"); + + let error = manager + .update_position(1, 2) + .await + .expect_err("an old manager must not resurrect a deleted checkpoint"); + assert!(error.to_string().contains("removed after this manager saved it")); + assert!(!CheckpointManager::has_checkpoint(&disk, &task_id).await); + temp_dir.close().expect("remove deleted checkpoint test directory"); +} + +#[tokio::test] +async fn an_empty_blocked_marker_still_blocks_resume_selection() { + let (temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let manager = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create checkpoint manager"); + let blocked_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}"); + disk.write_all(RUSTFS_META_BUCKET, &blocked_path, EcstoreDiskBytes::new()) + .await + .expect("write empty blocked marker fixture"); + + assert!(CheckpointManager::is_blocked(&disk, &task_id).await); + assert!(!CheckpointManager::is_resumable(&disk, &task_id).await); + // Recovery requires replacing/cleaning the snapshot, then removing the + // marker; ordinary selector retries are intentionally not an unlock path. + manager.cleanup().await.expect("clean blocked checkpoint"); + assert!(!CheckpointManager::is_blocked(&disk, &task_id).await); + temp_dir.close().expect("remove empty blocked marker test directory"); +} + +#[tokio::test] +async fn resumable_selector_skips_healthy_tasks_with_blocked_markers() { + let (temp_dir, disk) = schema_test_disk().await; + let tasks = [ + (ResumeUtils::generate_task_id(), EcstoreDiskBytes::new()), + (ResumeUtils::generate_task_id(), EcstoreDiskBytes::from_static(b"blocked")), + ]; + for (task_id, marker) in &tasks { + ResumeManager::new( + disk.clone(), + task_id.clone(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + vec!["bucket".to_string()], + ) + .await + .expect("create healthy resume state"); + CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create healthy checkpoint"); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + let checkpoint_bytes = disk + .read_all(RUSTFS_META_BUCKET, &checkpoint_path) + .await + .expect("read healthy checkpoint before blocking"); + let marker_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}"); + disk.write_all(RUSTFS_META_BUCKET, &marker_path, marker.clone()) + .await + .expect("write blocked marker"); + + assert!( + ResumeUtils::get_resumable_tasks(&disk) + .await + .expect("filter blocked healthy task") + .is_empty() + ); + assert_eq!( + disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path) + .await + .expect("read healthy checkpoint after blocking"), + checkpoint_bytes + ); + } + temp_dir.close().expect("remove blocked selector test directory"); +} + +#[tokio::test] +async fn stale_checkpoint_manager_cannot_overwrite_newer_progress() { + let (temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let first = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create first checkpoint manager"); + let second = CheckpointManager::load_from_disk(disk.clone(), &task_id) + .await + .expect("load second checkpoint manager"); + + second + .update_position(4, 20) + .await + .expect("persist newer checkpoint progress"); + let error = first + .update_position(1, 3) + .await + .expect_err("stale checkpoint manager must not overwrite newer progress"); + assert!(error.to_string().contains("newer progress")); + + let persisted = CheckpointManager::load_from_disk(disk.clone(), &task_id) + .await + .expect("load newer checkpoint progress") + .get_checkpoint() + .await; + assert_eq!(persisted.current_bucket_index, 4); + assert_eq!(persisted.current_object_index, 20); + temp_dir.close().expect("remove stale manager test directory"); +} + +#[tokio::test] +async fn resumable_selector_isolates_future_and_corrupt_checkpoints() { + let (temp_dir, disk) = schema_test_disk().await; + let future_task = ResumeUtils::generate_task_id(); + let corrupt_task = ResumeUtils::generate_task_id(); + for task_id in [&future_task, &corrupt_task] { + ResumeManager::new( + disk.clone(), + task_id.to_string(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + vec!["bucket".to_string()], + ) + .await + .expect("create resumable state fixture"); + } + + let future_path = format!("{BUCKET_META_PREFIX}/{future_task}_{RESUME_CHECKPOINT_FILE}"); + let mut future = ResumeCheckpoint::new(future_task.clone()); + future.schema_version = CURRENT_CHECKPOINT_SCHEMA + 1; + let future_bytes = serde_json::to_vec(&future).expect("serialize future checkpoint fixture"); + disk.write_all(RUSTFS_META_BUCKET, &future_path, future_bytes.clone().into()) + .await + .expect("write future checkpoint fixture"); + let corrupt_path = format!("{BUCKET_META_PREFIX}/{corrupt_task}_{RESUME_CHECKPOINT_FILE}"); + let corrupt_bytes = b"{truncated"; + disk.write_all(RUSTFS_META_BUCKET, &corrupt_path, corrupt_bytes.as_slice().into()) + .await + .expect("write corrupt checkpoint fixture"); + + assert!( + ResumeUtils::get_resumable_tasks(&disk) + .await + .expect("filter malformed resumable tasks") + .is_empty() + ); + assert!( + ResumeUtils::get_resumable_tasks(&disk) + .await + .expect("filter blocked resumable tasks") + .is_empty() + ); + for (task_id, path, bytes) in [ + (&future_task, future_path, future_bytes), + (&corrupt_task, corrupt_path, corrupt_bytes.to_vec()), + ] { + assert_eq!( + disk.read_all(RUSTFS_META_BUCKET, &path) + .await + .expect("read isolated checkpoint bytes"), + bytes + ); + let blocked_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}"); + assert!( + !disk + .read_all(RUSTFS_META_BUCKET, &blocked_path) + .await + .expect("read checkpoint blocked marker") + .is_empty() + ); + } + temp_dir.close().expect("remove selector isolation test directory"); +} + #[test] fn test_persist_throttle_batches_until_threshold() { let mut throttle = PersistThrottle::new(); diff --git a/crates/heal/src/heal/resume/utils.rs b/crates/heal/src/heal/resume/utils.rs index 71507e7c0..2ef0c4fae 100644 --- a/crates/heal/src/heal/resume/utils.rs +++ b/crates/heal/src/heal/resume/utils.rs @@ -21,7 +21,7 @@ use uuid::Uuid; use super::super::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET}; use super::replacement::{ReplacementPhase, ReplacementRecoveryRecord}; use super::{ - EVENT_HEAL_RESUME_STATE, LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, REPLACEMENT_COMPLETION_PROOF_FILE, + CheckpointManager, EVENT_HEAL_RESUME_STATE, LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, REPLACEMENT_COMPLETION_PROOF_FILE, REPLACEMENT_INTENT_FILE, RESUME_STATE_FILE, ResumeManager, ResumeStateFile, is_replacement_intent, path_to_str, replacement_recovery_corruption_for_state_load, replacement_recovery_dir, validate_resume_task_id, }; @@ -67,6 +67,7 @@ impl ResumeUtils { // Extract task ID from filename: {task_id}_ahm_resume_state.json if let Some(task_id) = entry.strip_suffix(&format!("_{RESUME_STATE_FILE}")) && validate_resume_task_id(task_id).is_ok() + && CheckpointManager::is_resumable(disk, task_id).await { task_ids.push(task_id.to_string()); }