From 276eea1fba7b4cbbc0c6ea88dcf61e628971ba46 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 10 Aug 2026 14:50:55 +0800 Subject: [PATCH] test(heal): cover replacement target evidence failures (#5919) Co-authored-by: heihutu --- crates/ecstore/src/set_disk/ops/heal.rs | 90 +++++++++++++++++++++++++ crates/heal/src/heal/erasure_healer.rs | 77 +++++++++++++++++++-- 2 files changed, 163 insertions(+), 4 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index cf7f6c1e4..10d159781 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -2558,6 +2558,96 @@ mod heal_result_report_tests { ); } + #[tokio::test] + async fn replacement_target_readback_checks_the_requested_historical_version() { + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "replacement-target-readback-versioned"; + let object = "object.bin"; + set.make_bucket( + bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("versioned bucket should be created"); + + let mut old_reader = PutObjReader::from_vec(vec![0x5a; 1024 * 1024]); + let old_info = set + .put_object( + bucket, + object, + &mut old_reader, + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("old object version should be written"); + let old_version = old_info + .version_id + .expect("versioned put should return the old version id") + .to_string(); + let mut latest_reader = PutObjReader::from_vec(vec![0x33; 1024 * 1024]); + let latest_info = set + .put_object( + bucket, + object, + &mut latest_reader, + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("latest object version should be written"); + let latest_version = latest_info + .version_id + .expect("versioned put should return the latest version id") + .to_string(); + let old_source = disks[2] + .read_version("", bucket, object, &old_version, &ReadOptions::default()) + .await + .expect("old version metadata should be readable"); + let old_data_dir = old_source.data_dir.expect("old version should have a data directory"); + let targets = vec![set.set_endpoints[0].to_string(), set.set_endpoints[1].to_string()]; + + assert!( + set.replacement_targets_have_version(bucket, object, &old_version, &targets) + .await + .expect("healthy historical target shards should be readable") + ); + assert!( + set.replacement_targets_have_version(bucket, object, &latest_version, &targets) + .await + .expect("healthy latest target shards should be readable") + ); + + tokio::fs::remove_file( + temp_dirs[1] + .path() + .join(bucket) + .join(object) + .join(old_data_dir.to_string()) + .join("part.1"), + ) + .await + .expect("old target shard should be removed after the initial commit"); + + assert!( + !set.replacement_targets_have_version(bucket, object, &old_version, &targets) + .await + .expect("missing old target shard should be observable") + ); + assert!( + set.replacement_targets_have_version(bucket, object, &latest_version, &targets) + .await + .expect("latest target evidence should remain independent") + ); + } + #[tokio::test] async fn format_heal_cached_layout_rejects_a_disk_from_another_slot() { let mut _temp_dirs = Vec::new(); diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 376a8d717..8f1810bb6 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -1229,6 +1229,12 @@ mod resume_loop_tests { Timeout, } + #[derive(Clone)] + enum ReplacementCommitEvidence { + Confirmed(bool), + Error(String), + } + #[derive(Default)] struct FakeStorage { /// page keyed by the *incoming* continuation token @@ -1239,7 +1245,7 @@ mod resume_loop_tests { results: Mutex>, /// Target-specific physical readback evidence per `compose_key`; the /// fake models a healthy backend unless a test explicitly revokes it. - replacement_commit_evidence: Mutex>, + replacement_commit_evidence: Mutex>, /// every heal_object call recorded as (name, version_id) heal_calls: Mutex)>>, replacement_target_identity_sequences: Mutex>>, @@ -1260,7 +1266,13 @@ mod resume_loop_tests { self.replacement_commit_evidence .lock() .unwrap() - .insert(compose_key(name, version), committed); + .insert(compose_key(name, version), ReplacementCommitEvidence::Confirmed(committed)); + } + fn set_replacement_commit_evidence_error(&self, name: &str, version: Option<&str>, message: &str) { + self.replacement_commit_evidence + .lock() + .unwrap() + .insert(compose_key(name, version), ReplacementCommitEvidence::Error(message.to_string())); } fn calls(&self) -> Vec<(String, Option)> { self.heal_calls.lock().unwrap().clone() @@ -1354,12 +1366,17 @@ mod resume_loop_tests { _opts: &HealOpts, _targets: &[String], ) -> Result { - Ok(*self + match self .replacement_commit_evidence .lock() .unwrap() .get(&compose_key(object, version_id)) - .unwrap_or(&true)) + .cloned() + .unwrap_or(ReplacementCommitEvidence::Confirmed(true)) + { + ReplacementCommitEvidence::Confirmed(committed) => Ok(committed), + ReplacementCommitEvidence::Error(message) => Err(Error::other(message)), + } } async fn list_objects_for_heal(&self, _b: &str, _p: &str) -> Result> { Ok(Vec::new()) @@ -2142,6 +2159,58 @@ mod resume_loop_tests { assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty()); } + #[tokio::test] + async fn replacement_delete_marker_readback_error_keeps_auto_heal_resumable() { + let env = make_env_with_targets(vec!["replacement-a".to_string()]).await; + let healer = ErasureSetHealer::new( + env.storage.clone(), + Arc::new(RwLock::new(HealProgress::new())), + CancellationToken::new(), + env.healer.disk.clone(), + HealOpts::default(), + HealRequestSource::AutoHeal, + ) + .with_replacement_targets(vec!["replacement-a".to_string()], Some("generation-a".to_string())); + env.storage.set_page( + None, + Page { + items: vec![item("object", Some("dm-v1"), true)], + next: None, + truncated: false, + }, + ); + env.storage.set_result( + "object", + Some("dm-v1"), + HealResultItem { + after: Infos { + drives: vec![HealDriveInfo { + endpoint: "replacement-a".to_string(), + state: "ok".to_string(), + ..Default::default() + }], + }, + ..Default::default() + }, + ); + env.storage + .set_replacement_commit_evidence_error("object", Some("dm-v1"), "injected target readback failure"); + + let result = healer + .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint) + .await; + + let error = result.expect_err("a target readback error must not complete automatic replacement"); + let error = error.to_string(); + assert!(error.contains("Transient heal skip"), "unexpected error: {error}"); + assert!(error.contains("retry scheduled"), "unexpected error: {error}"); + let state = env.resume.get_state().await; + assert!(!state.completed, "readback errors must leave the replacement task incomplete"); + assert_eq!(state.retry_count, 1, "readback errors must arm the bounded retry path"); + assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty()); + assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("dm-v1".to_string()))]); + } + #[tokio::test] async fn manual_targeted_heal_keeps_existing_best_effort_result_semantics() { let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;