diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs index c8b48da6d..933c53d09 100644 --- a/crates/heal/src/heal/outcome.rs +++ b/crates/heal/src/heal/outcome.rs @@ -110,7 +110,8 @@ impl HealObjectReceipt { && self.identity.version_id == expected.version_id && self.identity.pool_index == expected.pool_index && self.identity.set_index == expected.set_index - && self.identity.bucket_incarnation_id.is_some() + && self.identity.bucket_incarnation_id == expected.bucket_incarnation_id + && expected.bucket_incarnation_id.is_some() } } @@ -496,21 +497,31 @@ mod canonical_outcome_tests { #[test] fn positive_receipt_requires_exact_identity_and_bucket_incarnation() { - let expected = item(HealObjectDisposition::Unknown).identity; + let incarnation = Uuid::new_v4(); + let expected = HealObjectIdentity { + bucket_incarnation_id: Some(incarnation), + ..item(HealObjectDisposition::Unknown).identity + }; let mut receipt = HealObjectReceipt { identity: expected.clone(), disposition: HealObjectDisposition::Repaired, }; + receipt.identity.bucket_incarnation_id = None; assert!( !receipt.verified_for(&expected), "a positive storage receipt without bucket incarnation must remain untrusted" ); - let incarnation = Uuid::new_v4(); receipt.identity.bucket_incarnation_id = Some(incarnation); assert!(receipt.verified_for(&expected)); + receipt.identity.bucket_incarnation_id = Some(Uuid::new_v4()); + assert!( + !receipt.verified_for(&expected), + "a storage receipt for a different bucket incarnation must not clear the requested responsibility" + ); + receipt.identity.version_id = Some("older-version".to_string()); assert!( !receipt.verified_for(&expected), diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index c4554b327..962ee84b9 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -20,6 +20,7 @@ use rustfs_madmin::heal_commands::HealResultItem; use serde::{Deserialize, Serialize}; use std::sync::Arc; use tracing::{debug, error, warn}; +use uuid::Uuid; use super::outcome::{HealObjectDisposition, HealObjectIdentity, HealObjectKind, HealObjectReceipt}; use super::progress::stable_generation; @@ -383,6 +384,11 @@ pub trait HealStorageAPI: Send + Sync { /// Check object exists async fn object_exists(&self, bucket: &str, object: &str) -> Result; + /// Stable bucket incarnation observed before an object heal starts. + async fn bucket_incarnation_id(&self, _bucket: &str) -> Result> { + Ok(None) + } + /// Heal object using ecstore async fn heal_object( &self, @@ -1028,6 +1034,14 @@ impl HealStorageAPI for ECStoreHealStorage { } } + async fn bucket_incarnation_id(&self, bucket: &str) -> Result> { + self.ecstore + .bucket_incarnation_id(bucket) + .await + .map(Some) + .map_err(Error::Storage) + } + async fn heal_object( &self, bucket: &str, diff --git a/crates/heal/src/heal/task/heal_object.rs b/crates/heal/src/heal/task/heal_object.rs index 9e2c859e7..2fdbc4a8f 100644 --- a/crates/heal/src/heal/task/heal_object.rs +++ b/crates/heal/src/heal/task/heal_object.rs @@ -266,8 +266,10 @@ impl HealTask { let mut progress = self.progress.write().await; progress.update_object_progress(1, 1, 0, 0, object_size); } - let expected_identity = + let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?; + let mut expected_identity = self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index); + expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id; self.record_verified_storage_receipt(expected_identity, storage_result.receipt) .await; self.record_result_item(result).await; diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 01616f087..a026f6f35 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -1116,6 +1116,7 @@ struct MockStorage { heal_object_outcome: Mutex>, heal_object_outcomes: Mutex>>, heal_object_receipts: Mutex>>, + bucket_incarnation_id: Mutex>, format_no_heal_required: Mutex, format_error: Mutex>, global_format_calls: Mutex, @@ -1219,14 +1220,19 @@ async fn execute_emits_heal_trace_task_state() { assert_eq!(trace_attr_string(&completed, "state").as_deref(), Some("completed")); } -fn object_receipt(object: &str, version_id: Option<&str>, disposition: HealObjectDisposition) -> HealObjectReceipt { +fn object_receipt( + object: &str, + version_id: Option<&str>, + disposition: HealObjectDisposition, + bucket_incarnation_id: Uuid, +) -> HealObjectReceipt { HealObjectReceipt { identity: HealObjectIdentity { kind: HealObjectKind::Object, bucket: "bucket-a".to_string(), object: object.to_string(), version_id: version_id.map(ToOwned::to_owned), - bucket_incarnation_id: Some(Uuid::new_v4()), + bucket_incarnation_id: Some(bucket_incarnation_id), pool_index: None, set_index: None, }, @@ -1236,11 +1242,18 @@ fn object_receipt(object: &str, version_id: Option<&str>, disposition: HealObjec #[tokio::test] async fn object_heal_records_matching_positive_storage_receipt() { + let incarnation = Uuid::new_v4(); let storage = Arc::new(MockStorage { heal_object_receipts: Mutex::new(HashMap::from([( "object-a".to_string(), - VecDeque::from([object_receipt("object-a", Some("version-a"), HealObjectDisposition::Repaired)]), + VecDeque::from([object_receipt( + "object-a", + Some("version-a"), + HealObjectDisposition::Repaired, + incarnation, + )]), )])), + bucket_incarnation_id: Mutex::new(Some(incarnation)), ..Default::default() }); let task = HealTask::from_request( @@ -1262,15 +1275,18 @@ async fn object_heal_records_matching_positive_storage_receipt() { #[tokio::test] async fn object_heal_rejects_mismatched_or_legacy_storage_receipts() { + let expected_incarnation = Uuid::new_v4(); let storage = Arc::new(MockStorage { heal_object_receipts: Mutex::new(HashMap::from([( "object-a".to_string(), VecDeque::from([object_receipt( "object-a", - Some("old-version"), + Some("version-a"), HealObjectDisposition::Repaired, + Uuid::new_v4(), )]), )])), + bucket_incarnation_id: Mutex::new(Some(expected_incarnation)), ..Default::default() }); let task = HealTask::from_request( @@ -1456,6 +1472,10 @@ impl HealStorageAPI for MockStorage { Ok(self.object_exists.lock().unwrap().unwrap_or(true)) } + async fn bucket_incarnation_id(&self, _bucket: &str) -> Result> { + Ok(*self.bucket_incarnation_id.lock().unwrap()) + } + async fn heal_object( &self, bucket: &str,