diff --git a/crates/ecstore/src/cluster/rpc/client.rs b/crates/ecstore/src/cluster/rpc/client.rs index 75706254c..ca781d6be 100644 --- a/crates/ecstore/src/cluster/rpc/client.rs +++ b/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>, Box> { // 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>, Box> { @@ -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>, Box> { node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await } diff --git a/crates/ecstore/src/cluster/rpc/network_probe.rs b/crates/ecstore/src/cluster/rpc/network_probe.rs index 75d9049ea..765cc7322 100644 --- a/crates/ecstore/src/cluster/rpc/network_probe.rs +++ b/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 { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 11134035f..8b53add0b 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -242,6 +242,10 @@ fn read_all_data_std(path: &Path) -> core::result::Result<(Vec, Option Uuid { let used_data_dirs: HashSet = 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 { let base = version_id.as_u128() ^ INLINE_METADATA_ROLLBACK_DIR_XOR; let mut salt = 0u128; diff --git a/crates/ecstore/src/disk/local/delete_recovery.rs b/crates/ecstore/src/disk/local/delete_recovery.rs index c1b90219e..f8f4e3485 100644 --- a/crates/ecstore/src/disk/local/delete_recovery.rs +++ b/crates/ecstore/src/disk/local/delete_recovery.rs @@ -141,6 +141,73 @@ fn regular_bytes(path: &Path) -> Result>> { 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 { + 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(¤t)?, &FileMeta::load(&backup)?, transaction) +} + +fn live_write_backup_matches(current: &FileMeta, backup: &FileMeta, transaction: Uuid) -> Result { + 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 ¤t.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> { 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(), diff --git a/crates/ecstore/src/set_disk/ops/delete_recovery.rs b/crates/ecstore/src/set_disk/ops/delete_recovery.rs index 71d76f8b8..2aef4e65b 100644 --- a/crates/ecstore/src/set_disk/ops/delete_recovery.rs +++ b/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, /// 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 + Sync, can_commit: &(impl Fn() -> bool + Sync + ?Sized), namespace_owner: Option<&Arc>, + removed_versions: &mut HashSet, ) -> Result { 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> { - 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> { + 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 diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 200c13690..0cde84a0a 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/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 { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index b35eb4cb7..4c707eeea 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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| { diff --git a/crates/tls-runtime/src/material.rs b/crates/tls-runtime/src/material.rs index 5f8f200e5..7046f0174 100644 --- a/crates/tls-runtime/src/material.rs +++ b/crates/tls-runtime/src/material.rs @@ -225,10 +225,10 @@ fn load_server_material(source: &TlsSource, base_dir: &Path) -> Result, 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, (i64, i64)>; fn certificate_validities<'a>( diff --git a/rustfs/src/server/tls_publish.rs b/rustfs/src/server/tls_publish.rs index ab11787e9..afd6dc2fa 100644 --- a/rustfs/src/server/tls_publish.rs +++ b/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"