fix(ecstore): validate recovery evidence before mutation (#8487)

Reject causal cycles before source repair and share conservative backup
classification between discovery and inspection. Fence public raw deletion
and derive exact-key removals from successful settlement without an extra
metadata read. Preserve the existing physical phases and strict source checks.

Make outage and renewal fixtures deterministic without weakening their
convergence assertions, and complete strict TLS/RPC lint cleanup.

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/cluster/rpc/client.rs

@@ -54,7 +54,7 @@ use super::context_propagation::{inject_request_id_into_metadata, inject_trace_c
 /// 3. Subsequent calls will attempt fresh connections
 /// 4. If node is still down, connection will fail fast (3s timeout)
 pub async fn node_service_time_out_client(
-    addr: &String,
+    addr: &str,
     interceptor: TonicInterceptor,
 ) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
     // Default to the latency-sensitive control channel; bulk `bytes` RPCs opt in via the
@@ -110,7 +110,7 @@ pub async fn tier_mutation_control_time_out_client(
 /// from lock/health RPCs; everything else uses [`ChannelClass::Control`]. When isolation is
 /// disabled the two classes resolve to the same cached channel, i.e. legacy behavior.
 pub async fn node_service_time_out_client_for_class(
-    addr: &String,
+    addr: &str,
     interceptor: TonicInterceptor,
     class: ChannelClass,
 ) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
@@ -141,7 +141,7 @@ pub(crate) async fn node_service_delete_recovery_client(
 }
 
 pub async fn node_service_time_out_client_no_auth(
-    addr: &String,
+    addr: &str,
 ) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
     node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await
 }

crates/ecstore/src/cluster/rpc/network_probe.rs

@@ -289,13 +289,9 @@ async fn client(
     >,
     NetworkPeerProbeError,
 > {
-    node_service_time_out_client_for_class(
-        &address.to_owned(),
-        TonicInterceptor::Signature(gen_tonic_signature_interceptor()),
-        class,
-    )
-    .await
-    .map_err(|_| NetworkPeerProbeError::Unreachable)
+    node_service_time_out_client_for_class(address, TonicInterceptor::Signature(gen_tonic_signature_interceptor()), class)
+        .await
+        .map_err(|_| NetworkPeerProbeError::Unreachable)
 }
 
 fn ping_request(payload: &[u8]) -> PingRequest {

crates/ecstore/src/disk/local.rs

@@ -242,6 +242,10 @@ fn read_all_data_std(path: &Path) -> core::result::Result<(Vec<u8>, Option<Offse
 
 pub(crate) fn inline_metadata_rollback_dir(version_id: Uuid, meta: &FileMeta) -> Uuid {
     let used_data_dirs: HashSet<Uuid> = meta.get_data_dirs().unwrap_or_default().into_iter().flatten().collect();
+    inline_metadata_rollback_dir_with_used_dirs(version_id, &used_data_dirs)
+}
+
+fn inline_metadata_rollback_dir_with_used_dirs(version_id: Uuid, used_data_dirs: &HashSet<Uuid>) -> Uuid {
     let base = version_id.as_u128() ^ INLINE_METADATA_ROLLBACK_DIR_XOR;
     let mut salt = 0u128;
 

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -141,6 +141,73 @@ fn regular_bytes(path: &Path) -> Result<Option<Vec<u8>>> {
     regular_metadata_bytes(path)
 }
 
+/// A write/heal backup is not a delete transaction when the live metadata
+/// still owns its directory and retains every backed-up generation. Any delete
+/// evidence wins; uncertain backups remain discoverable and fail closed.
+fn live_write_backup(root: &Path, transaction: Uuid) -> Result<bool> {
+    let rollback = root.join(transaction.to_string());
+    for name in [
+        PREPARED,
+        TARGET,
+        FINISHED,
+        DELETE_MARKER_ROLLBACK_FILE,
+        DELETE_QUORUM_COMMITTED_FILE,
+    ] {
+        if regular_bytes(&rollback.join(name))?.is_some() {
+            return Ok(false);
+        }
+    }
+    for prefix in [
+        DELETE_DATA_DIR_MARKER_PREFIX,
+        RESERVED_DELETE_DATA_DIR_MARKER_PREFIX,
+        RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX,
+    ] {
+        let name = format!("{prefix}{transaction}");
+        if regular_bytes(&root.join(&name))?.is_some() || regular_bytes(&rollback.join(name))?.is_some() {
+            return Ok(false);
+        }
+    }
+    let Some(current) = regular_metadata_bytes(&root.join(STORAGE_FORMAT_FILE))? else {
+        return Ok(false);
+    };
+    let Some(backup) = regular_metadata_bytes(&rollback.join(STORAGE_FORMAT_FILE_BACKUP))? else {
+        return Ok(false);
+    };
+    live_write_backup_matches(&FileMeta::load(&current)?, &FileMeta::load(&backup)?, transaction)
+}
+
+fn live_write_backup_matches(current: &FileMeta, backup: &FileMeta, transaction: Uuid) -> Result<bool> {
+    let mut owns_directory = false;
+    let used_data_dirs: HashSet<_> = current.get_data_dirs()?.into_iter().flatten().collect();
+    let mut live = HashMap::with_capacity(current.versions.len());
+    for version in &current.versions {
+        let mut info = version.into_fileinfo("", "", true)?;
+        let id = info.version_id.unwrap_or_default();
+        owns_directory |=
+            info.data_dir == Some(transaction) || inline_metadata_rollback_dir_with_used_dirs(id, &used_data_dirs) == transaction;
+        rustfs_utils::http::metadata_compat::remove_str(&mut info.metadata, rustfs_utils::http::metadata_compat::SUFFIX_HEALING);
+        if live.insert(id, generation(info)?).is_some() {
+            return Err(DiskError::FileCorrupt);
+        }
+    }
+    if !owns_directory || backup.versions.is_empty() {
+        return Ok(false);
+    }
+    let mut seen = HashSet::with_capacity(backup.versions.len());
+    for version in &backup.versions {
+        let mut info = version.into_fileinfo("", "", true)?;
+        let id = info.version_id.unwrap_or_default();
+        if !seen.insert(id) {
+            return Err(DiskError::FileCorrupt);
+        }
+        rustfs_utils::http::metadata_compat::remove_str(&mut info.metadata, rustfs_utils::http::metadata_compat::SUFFIX_HEALING);
+        if live.get(&id) != Some(&generation(info)?) {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
 fn journal_bytes(journal: &DeleteRecoveryJournal) -> Result<Vec<u8>> {
     let bytes = serde_json::to_vec(journal)?;
     // The same journal is carried by adoption requests and decision receipts.
@@ -674,14 +741,10 @@ fn inspection(root: &Path, volume: &str, object: &str, transaction: Uuid) -> Res
         && target.is_none()
         && receipt_bytes.is_none()
         && !finished
-        && current_meta.as_ref().is_some_and(|meta| {
-            meta.get_data_dirs()
-                .is_ok_and(|dirs| dirs.into_iter().flatten().any(|dir| dir == transaction))
-        })
-        && before_meta.as_ref().is_some_and(|meta| {
-            meta.get_data_dirs()
-                .is_ok_and(|dirs| dirs.into_iter().flatten().any(|dir| dir == transaction))
-        });
+        && match (current_meta.as_ref(), before_meta.as_ref()) {
+            (Some(current), Some(backup)) => live_write_backup_matches(current, backup, transaction)?,
+            _ => false,
+        };
     let state = if live_write_backup {
         DeleteRecoveryState::Absent
     } else if journal.is_some() {
@@ -961,6 +1024,11 @@ impl LocalDisk {
                                 Some(_) => session.root.clone(),
                                 None => session.root.join(object),
                             };
+                            if name == STORAGE_FORMAT_FILE_BACKUP
+                                && live_write_backup(&physical, transaction).map_err(std::io::Error::from)?
+                            {
+                                continue;
+                            }
                             let proof_root = physical.join(transaction.to_string());
                             let prepared = regular_bytes(&proof_root.join(PREPARED)).map_err(std::io::Error::from)?;
                             let receipt =
@@ -2146,6 +2214,105 @@ mod tests {
         (temp, disk, info, bytes)
     }
 
+    #[tokio::test]
+    async fn delete_recovery_live_write_backup_preserves_delete_evidence() {
+        let (_temp, disk, info, original) = fixture().await;
+        let root = disk.io_get_object_path("recovery", "object").unwrap();
+        let meta = FileMeta::load(&original).unwrap();
+        let transaction = inline_metadata_rollback_dir(Uuid::nil(), &meta);
+        let rollback = root.join(transaction.to_string());
+        fs::create_dir_all(&rollback).await.unwrap();
+        fs::write(rollback.join(STORAGE_FORMAT_FILE_BACKUP), &original).await.unwrap();
+        let mut healing = info.clone();
+        healing.set_healing();
+        fs::write(root.join(STORAGE_FORMAT_FILE), test_meta(healing)).await.unwrap();
+        assert!(live_write_backup(&root, transaction).unwrap());
+        let page = disk
+            .delete_recovery_discover("recovery", Some("object"), None, 128)
+            .await
+            .unwrap();
+        assert!(page.transactions.is_empty());
+        assert_eq!(fs::read(rollback.join(STORAGE_FORMAT_FILE_BACKUP)).await.unwrap(), original);
+        fs::write(rollback.join(DELETE_QUORUM_COMMITTED_FILE), []).await.unwrap();
+        assert!(
+            !live_write_backup(&root, transaction).unwrap(),
+            "even an empty legacy receipt is delete evidence"
+        );
+        fs::remove_file(rollback.join(DELETE_QUORUM_COMMITTED_FILE)).await.unwrap();
+        let mut changed_parts = info.clone();
+        changed_parts.parts[0].actual_size += 1;
+        fs::write(root.join(STORAGE_FORMAT_FILE), test_meta(changed_parts))
+            .await
+            .unwrap();
+        assert!(
+            !matches!(live_write_backup(&root, transaction), Ok(true)),
+            "changed part metadata cannot hide recovery evidence"
+        );
+        let mut changed = info;
+        changed.metadata.insert("x-amz-object-lock-legal-hold".into(), "ON".into());
+        fs::write(root.join(STORAGE_FORMAT_FILE), test_meta(changed)).await.unwrap();
+        assert!(
+            !live_write_backup(&root, transaction).unwrap(),
+            "protected metadata differences remain meaningful"
+        );
+        fs::remove_file(root.join(STORAGE_FORMAT_FILE)).await.unwrap();
+        assert!(
+            !live_write_backup(&root, transaction).unwrap(),
+            "hidden legacy deletion remains discoverable"
+        );
+    }
+
+    #[tokio::test]
+    async fn delete_recovery_live_external_backup_rejects_partial_version_delete() {
+        let (_temp, disk, mut info, _) = fixture().await;
+        let root = disk.io_get_object_path("recovery", "object").unwrap();
+        let transaction = Uuid::new_v4();
+        info.data = None;
+        rustfs_utils::http::metadata_compat::remove_str(
+            &mut info.metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_INLINE_DATA,
+        );
+        info.data_dir = Some(transaction);
+        info.version_id = Some(Uuid::new_v4());
+        let mut meta = FileMeta::new();
+        meta.add_version(info.clone()).unwrap();
+        let original = meta.marshal_msg().unwrap();
+        let rollback = root.join(transaction.to_string());
+        fs::create_dir_all(&rollback).await.unwrap();
+        fs::write(rollback.join(STORAGE_FORMAT_FILE_BACKUP), &original).await.unwrap();
+        info.set_healing();
+        fs::write(root.join(STORAGE_FORMAT_FILE), test_meta(info.clone()))
+            .await
+            .unwrap();
+        assert!(live_write_backup(&root, transaction).unwrap());
+        let mut sibling = info.clone();
+        sibling.version_id = Some(Uuid::new_v4());
+        sibling.data_dir = Some(Uuid::new_v4());
+        meta.add_version(sibling).unwrap();
+        fs::write(rollback.join(STORAGE_FORMAT_FILE_BACKUP), meta.marshal_msg().unwrap())
+            .await
+            .unwrap();
+        assert!(
+            !live_write_backup(&root, transaction).unwrap(),
+            "a missing backed-up sibling retains delete recovery"
+        );
+        assert_eq!(
+            disk.inspect_delete_transaction("recovery", "object", transaction)
+                .await
+                .unwrap()
+                .state,
+            DeleteRecoveryState::LegacyAmbiguous
+        );
+        assert!(
+            disk.discover_delete_transactions("recovery", Some("object"), None, 128)
+                .await
+                .unwrap()
+                .transactions
+                .iter()
+                .any(|id| id.transaction == transaction)
+        );
+    }
+
     fn request() -> FileInfo {
         FileInfo {
             name: "object".to_owned(),

crates/ecstore/src/set_disk/ops/delete_recovery.rs

@@ -29,7 +29,7 @@ use uuid::Uuid;
 const DISCOVERY_PAGE_SIZE: u32 = 128;
 const MAX_DISCOVERED_TRANSACTIONS: usize = 1_000_000;
 
-#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub(crate) struct DeleteRecoverySummary {
     pub discovered: usize,
     pub committed: usize,
@@ -37,6 +37,8 @@ pub(crate) struct DeleteRecoverySummary {
     pub blocked: usize,
     pub pending: usize,
     pub reclaimed: usize,
+    /// Observed removals for this exact key, not aggregated across prefixes.
+    pub removed_versions: HashSet<Uuid>,
     /// Read-only evidence for ordinary missing-key S3 semantics. This never
     /// authorizes settlement while a configured participant is unavailable.
     pub no_transactions_at_quorum: bool,
@@ -191,6 +193,7 @@ async fn settle_group(
     action: impl Fn(&DeleteRecoveryInspection) -> Option<DeleteRecoveryAction> + Sync,
     can_commit: &(impl Fn() -> bool + Sync + ?Sized),
     namespace_owner: Option<&Arc<dyn Send + Sync>>,
+    removed_versions: &mut HashSet<Uuid>,
 ) -> Result<bool> {
     let results = join_all(disks.iter().zip(copies.iter()).map(|(disk, copy)| {
         let action = action(copy);
@@ -214,7 +217,22 @@ async fn settle_group(
     let mut pending = false;
     for (copy, result) in copies.iter_mut().zip(results) {
         match result {
-            Ok(Some(next)) => *copy = next,
+            Ok(Some(next)) => {
+                if copy.current_digest != next.current_digest {
+                    let retained: HashSet<_> = next
+                        .current_versions
+                        .iter()
+                        .map(|v| v.version_id.unwrap_or_default())
+                        .collect();
+                    removed_versions.extend(
+                        copy.current_versions
+                            .iter()
+                            .map(|v| v.version_id.unwrap_or_default())
+                            .filter(|id| !retained.contains(id)),
+                    );
+                }
+                *copy = next;
+            }
             Ok(None) => {}
             Err(DiskError::DiskOngoingReq) => pending = true,
             Err(error) => return Err(error),
@@ -469,6 +487,8 @@ impl SetDisks {
             }
             all_states.insert(transaction, complete);
         }
+        // Validate causal authority before prerequisite repair can publish metadata.
+        let transaction_order = causal_transaction_order(&all_states)?;
         let prerequisite = if repair_marker_source && !dry_run {
             let mut active = all_states
                 .iter()
@@ -502,7 +522,6 @@ impl SetDisks {
             }
             all_states.insert(transaction, refreshed);
         }
-        let transaction_order = causal_transaction_order(&all_states)?;
         // Legacy snapshots contain no set decision. A fresh authorized removal
         // applies only after the entire key is absent and every raw shape is safe.
         let legacy_only = explicit_delete
@@ -590,6 +609,7 @@ impl SetDisks {
                         },
                         can_commit,
                         namespace_owner.as_ref(),
+                        &mut summary.removed_versions,
                     )
                     .await?
                     {
@@ -696,6 +716,7 @@ impl SetDisks {
                     },
                     can_commit,
                     namespace_owner.as_ref(),
+                    &mut summary.removed_versions,
                 )
                 .await?;
                 if !pending {
@@ -706,6 +727,7 @@ impl SetDisks {
                         |_| Some(DeleteRecoveryAction::FinishCommit),
                         can_commit,
                         namespace_owner.as_ref(),
+                        &mut summary.removed_versions,
                     )
                     .await?;
                 }
@@ -752,6 +774,7 @@ impl SetDisks {
                     |copy| (copy.state == DeleteRecoveryState::NewPrepared).then_some(DeleteRecoveryAction::Rollback),
                     can_commit,
                     namespace_owner.as_ref(),
+                    &mut summary.removed_versions,
                 )
                 .await?
                 {
@@ -795,6 +818,7 @@ impl SetDisks {
                     |copy| (copy.state == DeleteRecoveryState::CommitFinished).then_some(DeleteRecoveryAction::FinalizeCommit),
                     can_commit,
                     namespace_owner.as_ref(),
+                    &mut summary.removed_versions,
                 )
                 .await?
                 {
@@ -1030,25 +1054,25 @@ mod tests {
         assert_eq!(body(&store, bucket, None).await, original);
     }
 
-    #[tokio::test]
-    async fn delete_recovery_marker_source_heal_preserves_foreign_history_and_damaged_data() {
-        fn snapshot(root: &std::path::Path) -> BTreeMap<std::path::PathBuf, Vec<u8>> {
-            let mut files = BTreeMap::new();
-            let mut pending = vec![root.to_path_buf()];
-            while let Some(directory) = pending.pop() {
-                for entry in std::fs::read_dir(directory).unwrap() {
-                    let path = entry.unwrap().path();
-                    if path.is_dir() {
-                        pending.push(path);
-                    } else {
-                        files.insert(path.strip_prefix(root).unwrap().to_path_buf(), std::fs::read(path).unwrap());
-                    }
+    fn snapshot(root: &std::path::Path) -> BTreeMap<std::path::PathBuf, Vec<u8>> {
+        let mut files = BTreeMap::new();
+        let mut pending = vec![root.to_path_buf()];
+        while let Some(directory) = pending.pop() {
+            for entry in std::fs::read_dir(directory).unwrap() {
+                let path = entry.unwrap().path();
+                if path.is_dir() {
+                    pending.push(path);
+                } else {
+                    files.insert(path.strip_prefix(root).unwrap().to_path_buf(), std::fs::read(path).unwrap());
                 }
             }
-            files
         }
+        files
+    }
 
-        for shape in ["live-backup", "foreign", "partial-history", "damaged-data"] {
+    #[tokio::test]
+    async fn delete_recovery_marker_source_heal_preserves_foreign_history_and_damaged_data() {
+        for shape in ["live-backup", "foreign", "partial-history", "damaged-data", "causal-cycle"] {
             let bucket = "recover-marker-source-guards";
             let (dirs, store) = isolated_store_over_temp_disks().await;
             init_bucket_metadata_sys(store.clone(), Vec::new()).await;
@@ -1135,6 +1159,33 @@ mod tests {
                         tokio::fs::write(part, bytes).await.unwrap();
                     }
                 }
+                "causal-cycle" => {
+                    for dir in dirs.iter().take(3) {
+                        let root = dir.path().join(bucket).join("object");
+                        let mut found = 0;
+                        for entry in std::fs::read_dir(root).unwrap() {
+                            let entry = entry.unwrap();
+                            let prepared = entry.path().join("delete-prepared");
+                            if !prepared.is_file() {
+                                continue;
+                            }
+                            let transaction = entry.file_name().to_str().unwrap().to_owned();
+                            let mut journal: serde_json::Value =
+                                serde_json::from_slice(&std::fs::read(&prepared).unwrap()).unwrap();
+                            journal["predecessors"] = serde_json::json!([transaction]);
+                            std::fs::write(prepared, serde_json::to_vec(&journal).unwrap()).unwrap();
+                            let receipt = entry.path().join(crate::disk::local::DELETE_QUORUM_COMMITTED_FILE);
+                            if receipt.is_file() {
+                                let mut value: serde_json::Value =
+                                    serde_json::from_slice(&std::fs::read(&receipt).unwrap()).unwrap();
+                                value["journal"] = journal;
+                                std::fs::write(receipt, serde_json::to_vec(&value).unwrap()).unwrap();
+                            }
+                            found += 1;
+                        }
+                        assert_eq!(found, 1, "one active marker journal must exercise prerequisite rejection");
+                    }
+                }
                 _ => unreachable!(),
             }
             let before: Vec<_> = dirs
@@ -1260,6 +1311,31 @@ mod tests {
             .expect("returning old replica must not leave bucket residue");
     }
 
+    #[tokio::test]
+    async fn delete_recovery_tracks_only_observed_removed_version() {
+        let bucket = "recover-actual-removal";
+        let (_dirs, store, disks) = fixture(bucket).await;
+        let target = put(&store, bucket, &[41; 4096], true).await.unwrap();
+        let retained = put(&store, bucket, &[42; 4096], true).await.unwrap();
+        let transaction = applied(&store, &disks, bucket, Some(target), 3).await;
+        publish(&disks[0], bucket, transaction).await;
+        let incarnation = store.bucket_incarnation_id_from_disk(bucket).await.unwrap();
+        let set = &store.pools[0].disk_set[0];
+        let report = set
+            .recover_delete_object(bucket, "object", DeleteRecoveryContext::new(Some(incarnation), &|| true))
+            .await
+            .unwrap();
+        assert_eq!(report.pending, 0);
+        assert_eq!(report.blocked, 0);
+        assert_eq!(report.removed_versions, HashSet::from([target]));
+        assert_eq!(body(&store, bucket, Some(retained)).await, vec![42; 4096]);
+        let replay = set
+            .recover_delete_object(bucket, "object", DeleteRecoveryContext::new(Some(incarnation), &|| true))
+            .await
+            .unwrap();
+        assert!(replay.removed_versions.is_empty(), "receipt replay cannot claim a second removal");
+    }
+
     #[tokio::test]
     async fn delete_recovery_set_retains_decisions_until_offline_peer_returns() {
         let bucket = "recover-offline";
@@ -1789,13 +1865,22 @@ mod tests {
     #[tokio::test]
     async fn delete_recovery_chained_committed_versions_finish_after_current_metadata_disappears() {
         let bucket = "recover-chained-versions";
-        let (_dirs, store, disks) = fixture(bucket).await;
+        let (dirs, store, disks) = fixture(bucket).await;
         let first = put(&store, bucket, &vec![61; 1024 * 1024], true)
             .await
             .expect("first version");
         let second = put(&store, bucket, &vec![62; 4096], true).await.expect("second version");
         let set = &store.pools[0].disk_set[0];
-        set.disks.write().await[3] = None;
+        let returning_key = dirs[3].path().join(bucket).join("object");
+        let before_outage = snapshot(&returning_key);
+        // A missing slot can be renewed by the endpoint monitor. Keep its
+        // original identity attached while rejecting I/O until explicit return.
+        disks[3].close().await.expect("stop returning drive's health monitor");
+        disks[3].force_offline_for_test();
+        set.connect_disks().await;
+        assert!(Arc::ptr_eq(set.disks.read().await[3].as_ref().unwrap(), &disks[3]));
+        assert_eq!(disks[3].runtime_state(), crate::disk::health_state::RuntimeDriveHealthState::Offline);
+        assert!(matches!(discover(&disks[3], bucket, Some("object")).await, Err(DiskError::FaultyDisk)));
         for version in [first, second] {
             store
                 .delete_object(
@@ -1816,7 +1901,9 @@ mod tests {
                 .await
                 .is_err()
         );
-        set.disks.write().await[3] = Some(disks[3].clone());
+        assert_eq!(snapshot(&returning_key), before_outage, "offline peer retains both deleted versions");
+        assert_eq!(disks[3].runtime_state(), crate::disk::health_state::RuntimeDriveHealthState::Offline);
+        disks[3].reset_health_for_store_init_retry();
         let report = store
             .recover_delete_transactions(bucket, &HealOpts::default())
             .await

crates/ecstore/src/set_disk/ops/heal.rs

@@ -3485,18 +3485,6 @@ impl SetDisks {
             && let Some(scope) = crate::store::bucket_heal_scope_for_object(bucket, object)
             && !crate::bucket::utils::is_meta_bucketname(bucket)
         {
-            let requested = Uuid::parse_str(version_id).ok();
-            if !opts.dry_run && (version_id.is_empty() || requested.is_some()) {
-                let disks = self.get_disks_internal().await;
-                let before = Self::read_metadata(&disks, "", bucket, object, version_id, false, false, false)
-                    .await
-                    .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
-                recovery_removed_version = before
-                    .slots
-                    .iter()
-                    .filter_map(|slot| slot.file_info())
-                    .any(|info| version_id.is_empty() || Some(info.version_id.unwrap_or_default()) == requested);
-            }
             let explicit_delete = opts.remove
                 && version_id.is_empty()
                 && crate::bucket::metadata_sys::has_authoritative_never_versioned_state_in(&self.ctx, bucket).await?
@@ -3522,6 +3510,16 @@ impl SetDisks {
                     },
                 )
                 .await?;
+            recovery_removed_version = if version_id.is_empty() {
+                !recovery.removed_versions.is_empty()
+            } else {
+                let target = if version_id == "null" {
+                    Some(Uuid::nil())
+                } else {
+                    Uuid::parse_str(version_id).ok()
+                };
+                target.is_some_and(|id| recovery.removed_versions.contains(&id))
+            };
             if opts.dry_run && recovery.discovered > 0 {
                 return Ok((
                     HealResultItem {

crates/ecstore/src/set_disk/ops/object.rs

@@ -8342,18 +8342,22 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         } else {
             self.bucket_incarnation_id_from_disk(bucket).await.ok()
         };
-        self.delete_object_version_with_purge(
-            bucket,
-            object,
-            fi,
-            force_del_marker,
-            None,
-            DeleteCommitContext {
-                incarnation: source_bucket_incarnation_id,
-                ..Default::default()
-            },
-        )
-        .await
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        let result = self
+            .delete_object_version_with_purge(
+                bucket,
+                object,
+                fi,
+                force_del_marker,
+                None,
+                DeleteCommitContext {
+                    incarnation: source_bucket_incarnation_id,
+                    ..Default::default()
+                },
+            )
+            .await;
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        result
     }
 
     #[tracing::instrument(skip(self, objects, opts))]
@@ -14241,6 +14245,25 @@ mod metadata_mutation_generation_tests {
         assert_retired(&set_disks, &delete_key).await;
         drop(delete_probe);
 
+        let raw_object = "raw-delete-object";
+        let (_, raw_key) = put_and_prime(&set_disks, bucket, raw_object, b"raw DELETE body").await;
+        let raw_probe = MetadataCacheInvalidationProbe::install(bucket, raw_object);
+        set_disks
+            .delete_object_version(
+                bucket,
+                raw_object,
+                &FileInfo {
+                    name: raw_object.to_owned(),
+                    ..Default::default()
+                },
+                false,
+            )
+            .await
+            .expect("raw DELETE should succeed");
+        assert_eq!(raw_probe.count(), 2, "raw DELETE must fence the cache before and after mutation");
+        assert_retired(&set_disks, &raw_key).await;
+        drop(raw_probe);
+
         let copy_object = "copy-object";
         let (mut copy_info, copy_key) = put_and_prime(&set_disks, bucket, copy_object, b"COPY body").await;
         copy_info.metadata_only = true;
@@ -24413,8 +24436,7 @@ mod delete_recovery_inventory_tests {
         });
         barrier.wait_until_paused().await;
         let extra_gate = set.disks.read().await;
-        let renewal = set.renew_disk(&endpoint);
-        tokio::pin!(renewal);
+        let mut renewal = Box::pin(set.renew_disk(&endpoint));
         tokio::time::timeout(
             Duration::from_secs(5),
             futures::future::poll_fn(|cx| {

crates/tls-runtime/src/material.rs

@@ -225,10 +225,10 @@ fn load_server_material(source: &TlsSource, base_dir: &Path) -> Result<Option<Se
     if cert_key_pairs.is_empty() {
         return Ok(None);
     }
-    if cert_key_pairs.len() == 1 {
-        if let Some((certs, key)) = cert_key_pairs.remove("default") {
-            return Ok(Some(ServerTlsMaterial::SingleCert { certs, key }));
-        }
+    if cert_key_pairs.len() == 1
+        && let Some((certs, key)) = cert_key_pairs.remove("default")
+    {
+        return Ok(Some(ServerTlsMaterial::SingleCert { certs, key }));
     }
     Ok(Some(ServerTlsMaterial::MultiCert { cert_key_pairs }))
 }

rustfs/src/server/tls_material.rs

@@ -512,10 +512,7 @@ async fn load_cert_file(path: &Path, pem_data: &mut Vec<u8>, desc: &str) -> bool
     }
 }
 
-/// Search for and load certificate files matching `cert_name` in the directory
-/// and one level of subdirectories.
-/// Returns `true` if at least one matching file was loaded.
-
+/// Validity windows indexed by leaf certificate DER.
 type CertificateValidities = HashMap<Vec<u8>, (i64, i64)>;
 
 fn certificate_validities<'a>(

rustfs/src/server/tls_publish.rs

@@ -1148,7 +1148,7 @@ mod tests {
                 let snapshot = generation
                     .into_snapshot(&TlsSource::from_directory(base.path()).snapshot().expect("source"))
                     .expect("the pinned operation consumes its already verified material without a second file read");
-                assert_eq!(snapshot_leaf(&snapshot), Some(receipt.certificate_fingerprint.clone()));
+                assert_eq!(snapshot_leaf(&snapshot), Some(receipt.certificate_fingerprint));
                 assert!(
                     matches!(state_at(root.path(), path), Err(PublishError::Invalid)),
                     "the next operation must revalidate disk"