mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-13 16:11:56 +00:00
fix(heal): fence canonical repair receipts (#7909)
* fix(heal): fence canonical repair receipts * fix(heal): preserve unversioned receipt identity
This commit is contained in:
@@ -2710,6 +2710,29 @@ impl SetDisks {
|
||||
}
|
||||
}
|
||||
|
||||
fn finalize_object_heal_result(
|
||||
mut result: HealResultItem,
|
||||
mut error: Option<Error>,
|
||||
absence: &mut Option<HealedObjectAbsence>,
|
||||
lock_lost: bool,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
) -> (HealResultItem, Option<Error>) {
|
||||
if lock_lost {
|
||||
result.integrity_verified = false;
|
||||
result.repair_verified = false;
|
||||
*absence = None;
|
||||
error = Some(Error::NamespaceLockQuorumUnavailable {
|
||||
mode: "write",
|
||||
bucket: bucket.to_owned(),
|
||||
object: object.to_owned(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
(result, error)
|
||||
}
|
||||
|
||||
impl SetDisks {
|
||||
#[cfg(test)]
|
||||
pub(crate) async fn heal_replacement_format(
|
||||
@@ -3095,18 +3118,26 @@ impl SetDisks {
|
||||
)
|
||||
.await
|
||||
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
|
||||
if _write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
|
||||
*absence = None;
|
||||
}
|
||||
return Ok((result, err.map(|e| e.into())));
|
||||
return Ok(finalize_object_heal_result(
|
||||
result,
|
||||
err.map(Into::into),
|
||||
absence,
|
||||
_write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()),
|
||||
bucket,
|
||||
object,
|
||||
));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
if _write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
|
||||
*absence = None;
|
||||
}
|
||||
Ok((result, err.map(|e| e.into())))
|
||||
Ok(finalize_object_heal_result(
|
||||
result,
|
||||
err.map(Into::into),
|
||||
absence,
|
||||
_write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()),
|
||||
bucket,
|
||||
object,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3162,6 +3193,36 @@ mod heal_result_report_tests {
|
||||
use tracing_subscriber::fmt::MakeWriter;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[test]
|
||||
fn lost_object_lock_revokes_storage_proofs_and_returns_retryable_error() {
|
||||
let mut absence = Some(super::HealedObjectAbsence {
|
||||
pool_index: 2,
|
||||
set_index: 3,
|
||||
removed: true,
|
||||
});
|
||||
let result = rustfs_madmin::heal_commands::HealResultItem {
|
||||
integrity_verified: true,
|
||||
repair_verified: true,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let (result, error) = super::finalize_object_heal_result(result, None, &mut absence, true, "bucket", "object");
|
||||
|
||||
assert!(!result.integrity_verified);
|
||||
assert!(!result.repair_verified);
|
||||
assert!(absence.is_none());
|
||||
assert!(matches!(
|
||||
error,
|
||||
Some(Error::NamespaceLockQuorumUnavailable {
|
||||
mode: "write",
|
||||
bucket,
|
||||
object,
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
}) if bucket == "bucket" && object == "object"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_less_part_file_accepts_positive_part_numbers_only() {
|
||||
assert!(super::metadata_less_part_file("part.1"));
|
||||
|
||||
@@ -123,12 +123,17 @@ fn verified_object_receipt(
|
||||
if !item.after.drives.iter().all(|drive| drive.state == ok_drive_state) {
|
||||
return None;
|
||||
}
|
||||
let receipt_version_id = if resolved_version.is_nil() {
|
||||
version_id.filter(|version| !version.is_empty()).map(ToOwned::to_owned)
|
||||
} else {
|
||||
Some(resolved_version.to_string())
|
||||
};
|
||||
Some(HealObjectReceipt {
|
||||
identity: HealObjectIdentity {
|
||||
kind: HealObjectKind::Object,
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
version_id: version_id.map(ToOwned::to_owned),
|
||||
version_id: receipt_version_id,
|
||||
bucket_incarnation_id: Some(bucket_incarnation_id),
|
||||
pool_index: opts.pool,
|
||||
set_index: opts.set,
|
||||
@@ -1807,6 +1812,7 @@ mod tests {
|
||||
let incarnation = Uuid::new_v4();
|
||||
let null = Uuid::nil().to_string();
|
||||
let latest = Uuid::new_v4();
|
||||
let latest_version = latest.to_string();
|
||||
let options = HealOpts::default();
|
||||
let mut item = HealResultItem {
|
||||
integrity_verified: true,
|
||||
@@ -1824,18 +1830,28 @@ mod tests {
|
||||
verified_object_receipt("bucket", "object", Some(&null), &options, &item, incarnation).is_none(),
|
||||
"echoing null cannot certify a different resolved version"
|
||||
);
|
||||
assert!(
|
||||
verified_object_receipt("bucket", "object", None, &options, &item, incarnation).is_some(),
|
||||
"an omitted selector still means latest"
|
||||
);
|
||||
assert!(verified_object_receipt("bucket", "object", Some(""), &options, &item, incarnation).is_some());
|
||||
let receipt = verified_object_receipt("bucket", "object", None, &options, &item, incarnation)
|
||||
.expect("an omitted selector still means latest");
|
||||
assert_eq!(receipt.identity.version_id.as_deref(), Some(latest_version.as_str()));
|
||||
let receipt = verified_object_receipt("bucket", "object", Some(""), &options, &item, incarnation)
|
||||
.expect("an empty selector still means latest");
|
||||
assert_eq!(receipt.identity.version_id.as_deref(), Some(latest_version.as_str()));
|
||||
assert!(
|
||||
verified_object_receipt("bucket", "object", Some("null"), &options, &item, incarnation).is_none(),
|
||||
"the internal boundary requires a UUID, not an S3 spelling"
|
||||
);
|
||||
assert!(verified_object_receipt("bucket", "object", Some(&latest.to_string()), &options, &item, incarnation).is_some());
|
||||
let requested_latest = latest_version.to_uppercase();
|
||||
let receipt = verified_object_receipt("bucket", "object", Some(&requested_latest), &options, &item, incarnation)
|
||||
.expect("the exact UUID selector should be certifiable");
|
||||
assert_eq!(receipt.identity.version_id.as_deref(), Some(latest_version.as_str()));
|
||||
|
||||
item.resolved_version_id = Some([0; 16]);
|
||||
let receipt = verified_object_receipt("bucket", "object", None, &options, &item, incarnation)
|
||||
.expect("an omitted selector should preserve unversioned identity");
|
||||
assert_eq!(receipt.identity.version_id, None);
|
||||
let receipt = verified_object_receipt("bucket", "object", Some(""), &options, &item, incarnation)
|
||||
.expect("an empty selector should preserve unversioned identity");
|
||||
assert_eq!(receipt.identity.version_id, None);
|
||||
let receipt = verified_object_receipt("bucket", "object", Some(&null), &options, &item, incarnation)
|
||||
.expect("the exact healthy null version should be certifiable");
|
||||
assert_eq!(receipt.identity.version_id.as_deref(), Some(null.as_str()));
|
||||
|
||||
@@ -180,6 +180,12 @@ impl HealTask {
|
||||
|
||||
match heal_result {
|
||||
Ok(storage_result) => {
|
||||
if let Some(resolved_version_id) = storage_result.item.resolved_version_id {
|
||||
let resolved_version = Uuid::from_bytes(resolved_version_id);
|
||||
if !resolved_version.is_nil() {
|
||||
expected_identity.version_id = Some(resolved_version.to_string());
|
||||
}
|
||||
}
|
||||
let result = storage_result.item;
|
||||
let error = storage_result.error;
|
||||
if let Some(e) = error {
|
||||
|
||||
@@ -1566,6 +1566,35 @@ async fn object_heal_records_matching_positive_storage_receipt() {
|
||||
assert_eq!(object.disposition, HealObjectDisposition::Repaired);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn object_heal_binds_omitted_selector_to_storage_resolved_version() {
|
||||
let incarnation = Uuid::new_v4();
|
||||
let resolved_version = Uuid::new_v4();
|
||||
let resolved_version_id = resolved_version.to_string();
|
||||
let storage = Arc::new(MockStorage {
|
||||
heal_object_receipts: Mutex::new(HashMap::from([(
|
||||
"object-a".to_string(),
|
||||
VecDeque::from([object_receipt(
|
||||
"object-a",
|
||||
Some(&resolved_version_id),
|
||||
HealObjectDisposition::Repaired,
|
||||
incarnation,
|
||||
)]),
|
||||
)])),
|
||||
bucket_incarnation_id: Mutex::new(Some(incarnation)),
|
||||
..Default::default()
|
||||
});
|
||||
let task = HealTask::from_request(HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None), storage);
|
||||
|
||||
task.execute().await.expect("latest object heal should complete");
|
||||
|
||||
let outcome = task.get_outcome().await;
|
||||
assert_eq!(outcome.counters.healed, 1);
|
||||
assert_eq!(outcome.counters.unknown, 0);
|
||||
let object = outcome.objects.front().expect("resolved latest receipt should be recorded");
|
||||
assert_eq!(object.identity.version_id.as_deref(), Some(resolved_version_id.as_str()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancelled_object_heal_rejects_matching_positive_storage_receipt() {
|
||||
let incarnation = Uuid::new_v4();
|
||||
@@ -2110,13 +2139,21 @@ impl HealStorageAPI for MockStorage {
|
||||
version_id: Option<&str>,
|
||||
opts: &HealOpts,
|
||||
) -> Result<HealStorageObjectResult> {
|
||||
let (item, error) = self.heal_object(bucket, object, version_id, opts).await?;
|
||||
let (mut item, error) = self.heal_object(bucket, object, version_id, opts).await?;
|
||||
let receipt = self
|
||||
.heal_object_receipts
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get_mut(object)
|
||||
.and_then(VecDeque::pop_front);
|
||||
if let Some(resolved_version_id) = receipt
|
||||
.as_ref()
|
||||
.and_then(|receipt| receipt.identity.version_id.as_deref())
|
||||
.and_then(|version| Uuid::parse_str(version).ok())
|
||||
.map(|version| *version.as_bytes())
|
||||
{
|
||||
item.resolved_version_id = Some(resolved_version_id);
|
||||
}
|
||||
Ok(HealStorageObjectResult { item, error, receipt })
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user