From 30a0937a7d058bcde7555975c0ad652937f9afa6 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 6 Sep 2026 16:56:24 +0800 Subject: [PATCH] fix(ecstore): retain single-delete physical namespace ownership (#7287) --- crates/ecstore/src/disk/disk_store.rs | 40 ++ crates/ecstore/src/disk/local.rs | 566 +++++++++++++++++----- crates/ecstore/src/disk/mod.rs | 40 ++ crates/ecstore/src/disk/os.rs | 80 ++- crates/ecstore/src/set_disk/ops/object.rs | 427 +++++++++++++++- 5 files changed, 1041 insertions(+), 112 deletions(-) diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index a11594997..2035b9ee3 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -324,6 +324,46 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper { } impl LocalDiskWrapper { + pub(in crate::disk) async fn delete_version_with_namespace_owner( + &self, + volume: &str, + path: &str, + fi: FileInfo, + force_del_marker: bool, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + self.track_disk_health_mutation( + "delete_version", + DiskMetricMutation::Delete, + || async { + Box::pin( + self.disk + .delete_version_with_namespace_owner(volume, path, fi, force_del_marker, opts, namespace_owner), + ) + .await + }, + get_max_timeout_duration(), + ) + .await + } + + pub(in crate::disk) async fn delete_with_namespace_owner( + &self, + volume: &str, + path: &str, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + self.track_disk_health_mutation( + "delete", + DiskMetricMutation::Delete, + || async { Box::pin(self.disk.delete_with_namespace_owner(volume, path, opts, namespace_owner)).await }, + get_max_timeout_duration(), + ) + .await + } + pub(in crate::disk) async fn undo_write_with_namespace_owner( &self, volume: &str, diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index c24c90724..58bef8df2 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -191,11 +191,33 @@ fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, r } async fn write_metadata_rollback_backup(object_dir: &Path, rollback_dir: Uuid, data: &[u8]) -> Result<()> { + write_delete_rollback_file(object_dir, rollback_dir, STORAGE_FORMAT_FILE_BACKUP, data, None).await +} + +async fn write_delete_rollback_file( + object_dir: &Path, + rollback_dir: Uuid, + name: &str, + data: &[u8], + namespace_owner: Option>, +) -> Result<()> { let backup_dir = object_dir.join(rollback_dir.to_string()); - fs::create_dir_all(&backup_dir).await.map_err(to_file_error)?; - fs::write(backup_dir.join(STORAGE_FORMAT_FILE_BACKUP), data) - .await - .map_err(to_file_error)?; + let path = backup_dir.join(name); + if namespace_owner.is_none() { + fs::create_dir_all(&backup_dir).await.map_err(to_file_error)?; + fs::write(path, data).await.map_err(to_file_error)?; + return Ok(()); + } + let lease = os::acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await; + let data = data.to_vec(); + os::run_blocking_namespace_operation(lease, move || { + std::fs::create_dir_all(&backup_dir)?; + #[cfg(test)] + run_owned_file_write_before_open(&path); + std::fs::write(path, data) + }) + .await + .map_err(to_file_error)?; Ok(()) } @@ -342,6 +364,7 @@ struct DeleteVersionMutation { struct DeleteRollbackFailure { stage: &'static str, error: DiskError, + namespace_owner: Option>, } async fn restore_delete_rollback_after_error( @@ -353,12 +376,18 @@ async fn restore_delete_rollback_after_error( failure: DeleteRollbackFailure, publication_root: &os::PublicationRoot, ) -> DiskError { - let DeleteRollbackFailure { stage, error } = failure; + let DeleteRollbackFailure { + stage, + error, + namespace_owner, + } = failure; let Some(rollback_dir) = rollback_dir else { return error; }; - if let Err(restore_err) = restore_delete_rollback(object_dir, xl_path, rollback_dir, publication_root).await { + if let Err(restore_err) = + restore_delete_rollback_with_namespace_owner(object_dir, xl_path, rollback_dir, publication_root, namespace_owner).await + { warn!( volume, path, @@ -5654,6 +5683,123 @@ impl LocalDisk { // }) // } + #[tracing::instrument(name = "delete_version", level = "trace", skip_all)] + pub(in crate::disk) async fn delete_version_with_namespace_owner( + &self, + volume: &str, + path: &str, + fi: FileInfo, + force_del_marker: bool, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + self.delete_version_inner( + volume, + path, + fi, + DeleteVersionMutation { + force_del_marker, + opts, + namespace_owner, + }, + ) + .await + } + + #[tracing::instrument(name = "write_metadata", level = "trace", skip_all)] + async fn write_metadata_with_namespace_owner( + &self, + volume: &str, + path: &str, + fi: FileInfo, + namespace_owner: Option>, + ) -> Result<()> { + crate::hp_guard!("LocalDisk::write_metadata"); + fi.validate_for_metadata_read()?; + let p = self.io_get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?; + + let mut meta = FileMeta::new(); + if !fi.fresh { + let (buf, _) = read_file_exists(&p).await?; + if !buf.is_empty() { + let _ = meta.unmarshal_msg(&buf).map_err(|_| { + meta = FileMeta::new(); + }); + } + } + + meta.add_version(fi)?; + + let fm_data = meta.marshal_msg()?; + + // Atomic temp+rename: this path also rewrites live xl.meta (delete markers, + // decommission), where an in-place truncate would expose torn metadata. + self.write_all_meta_with_namespace_owner( + volume, + format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), + &fm_data, + true, + namespace_owner, + ) + .await?; + + Ok(()) + } + + async fn delete_data_dir_with_namespace_owner( + &self, + volume: &str, + path: &str, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result { + let key = SnapshotLeaseKey { + volume: volume.to_string(), + path: path.to_string(), + }; + { + let mut registry = self.snapshot_leases.lock().await; + if let Some(entry) = registry.entries.get_mut(&key) { + if !entry.tokens.is_empty() { + entry.pending_delete.get_or_insert_with(|| opts.clone()); + return Ok(DataDirDeleteStatus::Deferred); + } + if entry.deleting { + entry.pending_delete.get_or_insert_with(|| opts.clone()); + return Ok(DataDirDeleteStatus::Deferred); + } + entry.deleting = true; + entry.pending_delete.get_or_insert_with(|| opts.clone()); + } else { + registry.entries.insert( + key.clone(), + SnapshotLeaseEntry { + pending_delete: Some(opts.clone()), + deleting: true, + ..Default::default() + }, + ); + } + } + + let result = self + .delete_unleased_with_namespace_owner(volume, path, &opts, namespace_owner) + .await; + let mut registry = self.snapshot_leases.lock().await; + match result { + Ok(()) => { + registry.entries.remove(&key); + Ok(DataDirDeleteStatus::Deleted) + } + Err(err) => { + if let Some(entry) = registry.entries.get_mut(&key) { + entry.deleting = false; + } + Err(err) + } + } + } + async fn delete_version_inner(&self, volume: &str, path: &str, fi: FileInfo, mutation: DeleteVersionMutation) -> Result<()> { let DeleteVersionMutation { force_del_marker, @@ -5696,7 +5842,7 @@ impl LocalDisk { if fi.deleted && force_del_marker { return self - .write_missing_delete_marker(volume, path, fi, file_path.as_path(), &xl_path, rollback_dir) + .write_missing_delete_marker(volume, path, fi, file_path.as_path(), rollback_dir, namespace_owner.clone()) .await; } @@ -5712,7 +5858,14 @@ impl LocalDisk { let old_dir = meta.delete_version(&fi)?; let mut reserved_version_delete = false; if let Some(rollback_dir) = rollback_dir { - write_metadata_rollback_backup(file_path.as_path(), rollback_dir, &buf).await?; + write_delete_rollback_file( + file_path.as_path(), + rollback_dir, + STORAGE_FORMAT_FILE_BACKUP, + &buf, + namespace_owner.clone(), + ) + .await?; } if let Some(uuid) = old_dir { @@ -5728,6 +5881,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_metadata_update", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5745,6 +5899,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_data_path", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5753,7 +5908,7 @@ impl LocalDisk { if let Some(rollback_dir) = rollback_dir { let rollback_path = file_path.join(rollback_dir.to_string()); - if let Err(err) = fs::create_dir_all(&rollback_path).await { + if let Err(err) = os::create_dir_all_with_namespace_owner(&rollback_path, namespace_owner.clone()).await { let err: DiskError = to_file_error(err).into(); return Err(restore_delete_rollback_after_error( file_path.as_path(), @@ -5764,12 +5919,16 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_rollback_dir", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) .await); } - reserved_version_delete = match self.reserve_version_delete(volume, path, uuid, rollback_dir).await { + reserved_version_delete = match self + .reserve_version_delete_with_namespace_owner(volume, path, uuid, rollback_dir, namespace_owner.clone()) + .await + { Ok(reserved) => reserved, Err(err) => { return Err(restore_delete_rollback_after_error( @@ -5781,6 +5940,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_reserve_data", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5789,9 +5949,14 @@ impl LocalDisk { }; let rollback_data_path = rollback_path.join(uuid.to_string()); if !reserved_version_delete - && let Err(err) = - rename_all_ignore_missing_source(&old_path, &rollback_data_path, &rollback_path, &self.publication_root) - .await + && let Err(err) = os::rename_all_ignore_missing_source_with_owner( + &old_path, + &rollback_data_path, + &rollback_path, + &self.publication_root, + namespace_owner.clone(), + ) + .await { return Err(restore_delete_rollback_after_error( file_path.as_path(), @@ -5802,6 +5967,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_stage_data", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5810,13 +5976,16 @@ impl LocalDisk { if should_fail_after_delete_data_staged(path) { if reserved_version_delete { return Err(self - .abort_reserved_version_delete( + .abort_reserved_version_delete_with_failure( file_path.as_path(), rollback_dir, volume, path, - "delete_version_test_after_stage", - DiskError::Unexpected, + DeleteRollbackFailure { + stage: "delete_version_test_after_stage", + error: DiskError::Unexpected, + namespace_owner: namespace_owner.clone(), + }, ) .await); } @@ -5829,6 +5998,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_test_after_stage", error: DiskError::Unexpected, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5858,13 +6028,16 @@ impl LocalDisk { let err: DiskError = err.into(); if reserved_version_delete && let Some(rollback_dir) = rollback_dir { return Err(self - .abort_reserved_version_delete( + .abort_reserved_version_delete_with_failure( file_path.as_path(), rollback_dir, volume, path, - "delete_version_metadata_encode", - err, + DeleteRollbackFailure { + stage: "delete_version_metadata_encode", + error: err, + namespace_owner: namespace_owner.clone(), + }, ) .await); } @@ -5877,6 +6050,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_metadata_encode", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5899,7 +6073,17 @@ impl LocalDisk { if let Err(err) = commit_result { if reserved_version_delete && let Some(rollback_dir) = rollback_dir { return Err(self - .abort_reserved_version_delete(file_path.as_path(), rollback_dir, volume, path, "delete_version_commit", err) + .abort_reserved_version_delete_with_failure( + file_path.as_path(), + rollback_dir, + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_commit", + error: err, + namespace_owner: namespace_owner.clone(), + }, + ) .await); } return Err(restore_delete_rollback_after_error( @@ -5911,6 +6095,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_version_commit", error: err, + namespace_owner: namespace_owner.clone(), }, &self.publication_root, ) @@ -5919,16 +6104,21 @@ impl LocalDisk { if reserved_version_delete && let Some(rollback_dir) = rollback_dir - && let Err(err) = self.commit_reserved_version_delete(volume, path, rollback_dir).await + && let Err(err) = self + .commit_reserved_version_delete_with_namespace_owner(volume, path, rollback_dir, namespace_owner.clone()) + .await { return Err(self - .abort_reserved_version_delete( + .abort_reserved_version_delete_with_failure( file_path.as_path(), rollback_dir, volume, path, - "delete_version_commit_intent", - err, + DeleteRollbackFailure { + stage: "delete_version_commit_intent", + error: err, + namespace_owner: namespace_owner.clone(), + }, ) .await); } @@ -6098,7 +6288,7 @@ impl LocalDisk { } #[tracing::instrument(name = "delete", level = "trace", skip_all)] - async fn delete_with_namespace_owner( + pub(in crate::disk) async fn delete_with_namespace_owner( &self, volume: &str, path: &str, @@ -6111,7 +6301,8 @@ impl LocalDisk { && let Some((object, transaction_id)) = path.rsplit_once('/') && let Ok(transaction_id) = Uuid::parse_str(transaction_id) { - self.finish_version_delete(volume, object, transaction_id).await? + self.finish_version_delete(volume, object, transaction_id, namespace_owner.clone()) + .await? } else { false }; @@ -6453,19 +6644,27 @@ impl LocalDisk { path: &str, fi: FileInfo, object_dir: &Path, - xl_path: &Path, rollback_dir: Option, + namespace_owner: Option>, ) -> Result<()> { + let xl_path = object_dir.join(STORAGE_FORMAT_FILE); if let Some(rollback_dir) = rollback_dir { - let rollback_path = object_dir.join(rollback_dir.to_string()); - fs::create_dir_all(&rollback_path).await.map_err(to_file_error)?; - fs::write(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE), []) - .await - .map_err(to_file_error)?; + write_delete_rollback_file(object_dir, rollback_dir, DELETE_MARKER_ROLLBACK_FILE, &[], namespace_owner.clone()) + .await?; } - if let Err(err) = self.write_metadata("", volume, path, fi).await { + if let Err(err) = self + .write_metadata_with_namespace_owner(volume, path, fi, namespace_owner.clone()) + .await + { if let Some(rollback_dir) = rollback_dir - && let Err(restore_err) = restore_delete_rollback(object_dir, xl_path, rollback_dir, &self.publication_root).await + && let Err(restore_err) = restore_delete_rollback_with_namespace_owner( + object_dir, + &xl_path, + rollback_dir, + &self.publication_root, + namespace_owner, + ) + .await { warn!( event = EVENT_DISK_LOCAL_DELETE_ROLLBACK_FAILED, @@ -6510,7 +6709,7 @@ impl LocalDisk { return Err(DiskError::FileNotFound); }; return self - .write_missing_delete_marker(volume, path, delete_marker, object_dir, &xlpath, opts.old_data_dir) + .write_missing_delete_marker(volume, path, delete_marker, object_dir, opts.old_data_dir, None) .await; } Err(err) => return Err(err), @@ -6559,6 +6758,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_metadata_update", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -6594,6 +6794,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_data_path", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -6625,6 +6826,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_rollback_dir", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -6665,6 +6867,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_stage_data", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -6692,6 +6895,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_test_after_stage", error: DiskError::Unexpected, + namespace_owner: None, }, &self.publication_root, ) @@ -6734,6 +6938,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_commit_delete", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -6780,6 +6985,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_metadata_encode", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -6805,6 +7011,7 @@ impl LocalDisk { DeleteRollbackFailure { stage: "delete_versions_commit_write", error: err, + namespace_owner: None, }, &self.publication_root, ) @@ -8015,6 +8222,18 @@ impl LocalDisk { } async fn reserve_version_delete(&self, volume: &str, object: &str, data_dir: Uuid, rollback_dir: Uuid) -> Result { + self.reserve_version_delete_with_namespace_owner(volume, object, data_dir, rollback_dir, None) + .await + } + + async fn reserve_version_delete_with_namespace_owner( + &self, + volume: &str, + object: &str, + data_dir: Uuid, + rollback_dir: Uuid, + namespace_owner: Option>, + ) -> Result { let path = format!("{object}/{data_dir}"); let data_path = self.io_get_object_path(volume, &path)?; match fs::metadata(&data_path).await { @@ -8024,6 +8243,28 @@ impl LocalDisk { Err(err) => return Err(to_file_error(err).into()), } let marker_path = data_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}")); + if namespace_owner.is_some() { + let lease = os::acquire_namespace_mutation_lease_with_owner(&marker_path, namespace_owner.clone()).await; + let volume = volume.to_string(); + let sync = os::run_blocking_namespace_operation(lease, move || { + #[cfg(test)] + run_owned_file_write_before_open(&marker_path); + let marker = std::fs::File::create(marker_path)?; + let sync = effective_durability(&volume).syncs_commit_metadata(); + if sync { + marker.sync_all()?; + } + Ok(sync) + }) + .await + .map_err(to_file_error)?; + if sync { + os::fsync_dir_with_owner(&data_path, namespace_owner) + .await + .map_err(to_file_error)?; + } + return Ok(true); + } let marker = File::create(marker_path).await.map_err(to_file_error)?; if effective_durability(volume).syncs_commit_metadata() { marker.sync_all().await.map_err(to_file_error)?; @@ -8033,6 +8274,17 @@ impl LocalDisk { } async fn commit_reserved_version_delete(&self, volume: &str, object: &str, rollback_dir: Uuid) -> Result<()> { + self.commit_reserved_version_delete_with_namespace_owner(volume, object, rollback_dir, None) + .await + } + + async fn commit_reserved_version_delete_with_namespace_owner( + &self, + volume: &str, + object: &str, + rollback_dir: Uuid, + namespace_owner: Option>, + ) -> Result<()> { let object_path = self.io_get_object_path(volume, object)?; let mut entries = match fs::read_dir(object_path).await { Ok(entries) => entries, @@ -8048,10 +8300,14 @@ impl LocalDisk { continue; } let reserved_path = entry.path().join(&reserved_name); - match fs::rename(&reserved_path, entry.path().join(&committed_name)).await { + match os::rename_with_namespace_owner(&reserved_path, &entry.path().join(&committed_name), namespace_owner.clone()) + .await + { Ok(()) => { if effective_durability(volume).syncs_commit_metadata() { - os::fsync_dir(&entry.path()).await.map_err(to_file_error)?; + os::fsync_dir_with_owner(&entry.path(), namespace_owner.clone()) + .await + .map_err(to_file_error)?; } } Err(err) if err.kind() == ErrorKind::NotFound => {} @@ -8061,7 +8317,13 @@ impl LocalDisk { Ok(()) } - async fn finish_version_delete(&self, volume: &str, object: &str, rollback_dir: Uuid) -> Result { + async fn finish_version_delete( + &self, + volume: &str, + object: &str, + rollback_dir: Uuid, + namespace_owner: Option>, + ) -> Result { let object_path = self.io_get_object_path(volume, object)?; let mut entries = match fs::read_dir(object_path).await { Ok(entries) => entries, @@ -8082,13 +8344,14 @@ impl LocalDisk { Err(err) => return Err(to_file_error(err).into()), } if let Err(err) = self - .delete_data_dir( + .delete_data_dir_with_namespace_owner( volume, &format!("{object}/{data_dir}"), DeleteOptions { recursive: true, ..Default::default() }, + namespace_owner.clone(), ) .await && first_err.is_none() @@ -8109,6 +8372,28 @@ impl LocalDisk { object: &str, stage: &'static str, err: DiskError, + ) -> DiskError { + self.abort_reserved_version_delete_with_failure( + object_dir, + rollback_dir, + volume, + object, + DeleteRollbackFailure { + stage, + error: err, + namespace_owner: None, + }, + ) + .await + } + + async fn abort_reserved_version_delete_with_failure( + &self, + object_dir: &Path, + rollback_dir: Uuid, + volume: &str, + object: &str, + failure: DeleteRollbackFailure, ) -> DiskError { let xl_path = object_dir.join(STORAGE_FORMAT_FILE); restore_delete_rollback_after_error( @@ -8117,7 +8402,7 @@ impl LocalDisk { Some(rollback_dir), volume, object, - DeleteRollbackFailure { stage, error: err }, + failure, &self.publication_root, ) .await @@ -9608,49 +9893,7 @@ impl DiskAPI for LocalDisk { } async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result { - let key = SnapshotLeaseKey { - volume: volume.to_string(), - path: path.to_string(), - }; - { - let mut registry = self.snapshot_leases.lock().await; - if let Some(entry) = registry.entries.get_mut(&key) { - if !entry.tokens.is_empty() { - entry.pending_delete.get_or_insert_with(|| opts.clone()); - return Ok(DataDirDeleteStatus::Deferred); - } - if entry.deleting { - entry.pending_delete.get_or_insert_with(|| opts.clone()); - return Ok(DataDirDeleteStatus::Deferred); - } - entry.deleting = true; - entry.pending_delete.get_or_insert_with(|| opts.clone()); - } else { - registry.entries.insert( - key.clone(), - SnapshotLeaseEntry { - pending_delete: Some(opts.clone()), - deleting: true, - ..Default::default() - }, - ); - } - } - - let result = self.delete_unleased(volume, path, &opts).await; - let mut registry = self.snapshot_leases.lock().await; - match result { - Ok(()) => { - registry.entries.remove(&key); - Ok(DataDirDeleteStatus::Deleted) - } - Err(err) => { - if let Some(entry) = registry.entries.get_mut(&key) { - entry.deleting = false; - } - Err(err) - } - } + self.delete_data_dir_with_namespace_owner(volume, path, opts, None).await } #[tracing::instrument(level = "trace", skip_all)] @@ -9689,32 +9932,8 @@ impl DiskAPI for LocalDisk { Err(Error::other("Invalid Argument")) } - #[tracing::instrument(level = "trace", skip_all)] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { - crate::hp_guard!("LocalDisk::write_metadata"); - fi.validate_for_metadata_read()?; - let p = self.io_get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?; - - let mut meta = FileMeta::new(); - if !fi.fresh { - let (buf, _) = read_file_exists(&p).await?; - if !buf.is_empty() { - let _ = meta.unmarshal_msg(&buf).map_err(|_| { - meta = FileMeta::new(); - }); - } - } - - meta.add_version(fi)?; - - let fm_data = meta.marshal_msg()?; - - // Atomic temp+rename: this path also rewrites live xl.meta (delete markers, - // decommission), where an in-place truncate would expose torn metadata. - self.write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &fm_data, true) - .await?; - - Ok(()) + self.write_metadata_with_namespace_owner(volume, path, fi, None).await } #[tracing::instrument(level = "trace", skip_all)] @@ -22177,4 +22396,135 @@ mod test { assert_eq!(mount_id_from_mountinfo_contents(mountinfo, Path::new("/mnt/replacement disk")), Some(202)); assert_eq!(mount_id_from_mountinfo_contents(mountinfo, Path::new("/mnt/replacement")), None); } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_internal_restore_keeps_owner_after_cancellation() { + use crate::disk::os::prepared_publication_test_hooks as hooks; + use futures::FutureExt; + + let dir = tempfile::tempdir().expect("fixture directory"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 fixture path")).expect("endpoint"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk")); + let bucket = "single-delete-internal-restore"; + let object = format!("object-{}", Uuid::new_v4()); + ensure_test_volume(&disk, bucket).await; + let version = Uuid::new_v4(); + let data_dir = Uuid::new_v4(); + let rollback_dir = Uuid::new_v4(); + let fi = test_file_info(&object, version, Some(data_dir), None); + let original = test_meta(fi.clone()); + let object_dir = disk.io_get_object_path(bucket, &object).expect("object IO path"); + let part = object_dir.join(data_dir.to_string()).join("part.1"); + let metadata = object_dir.join(STORAGE_FORMAT_FILE); + let backup = object_dir.join(rollback_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP); + fs::create_dir_all(part.parent().expect("data parent")) + .await + .expect("data directory"); + fs::write(&part, b"x").await.expect("real shard"); + fs::write(&metadata, &original).await.expect("real version metadata"); + set_delete_version_fail_after_data_staged(&object); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release, release_rx) = std::sync::mpsc::channel::<()>(); + let hook = hooks::install_at(hooks::Stage::Rename, &metadata, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let before = ctx.namespace_commit_generation(); + let owner = ctx.begin_namespace_commit(); + let deleting_disk = Arc::clone(&disk); + let deleting_object = object.clone(); + let mut delete = tokio::spawn(async move { + deleting_disk + .delete_version_inner( + bucket, + &deleting_object, + fi, + DeleteVersionMutation { + force_del_marker: false, + opts: DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + namespace_owner: Some(owner), + }, + ) + .await + }); + let mut joined = false; + let mut entered = false; + let mut counts = None; + let observations = std::panic::AssertUnwindSafe(async { + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + result = entered_rx => { + result.expect("actual internal restore entry"); + entered = true; + } + result = &mut delete => { + joined = true; + panic!("delete returned before internal physical restore: {result:?}"); + } + } + }) + .await + .expect("internal restore must reach the physical rename"); + assert_eq!(std::fs::read(&backup).expect("real undo backup"), original); + assert_eq!(std::fs::read(&part).expect("reserved shard"), b"x"); + delete.abort(); + let result = tokio::time::timeout(Duration::from_secs(5), &mut delete).await; + joined = result.is_ok(); + assert!( + result + .expect("cancelled caller joins") + .expect_err("cancelled caller") + .is_cancelled() + ); + counts = Some((ctx.namespace_commits_pending(), ctx.namespace_commit_generation())); + assert!(hooks::drain_namespace_key(&metadata).now_or_never().is_none()); + }) + .catch_unwind() + .await; + + drop(release); + drop(hook); + let coordinator_drained = joined || tokio::time::timeout(Duration::from_secs(10), &mut delete).await.is_ok(); + if !coordinator_drained { + delete.abort(); + let _ = tokio::time::timeout(Duration::from_secs(5), &mut delete).await; + } + let physical_drained = tokio::time::timeout(Duration::from_secs(5), hooks::drain_namespace_key(&metadata)) + .await + .is_ok(); + let owner_drained = tokio::time::timeout(Duration::from_secs(5), async { + while ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await + .is_ok(); + if !entered || !coordinator_drained || !physical_drained || !owner_drained { + eprintln!("internal restore cleanup incomplete; retained root: {:?}", dir.keep()); + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + panic!("internal restore cleanup must drain before removing its root"); + } + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + assert_eq!(std::fs::read(&metadata).expect("late restored metadata"), original); + assert!(!backup.exists(), "the actual backup rename must have completed"); + assert_eq!(std::fs::read(&part).expect("old shard survives"), b"x"); + disk.read_version("", bucket, &object, &version.to_string(), &ReadOptions::default()) + .await + .expect("restored version"); + let (pending, generation) = counts.expect("observations completed"); + assert!(pending, "internal error recovery lost the physical namespace owner"); + assert_eq!(generation, before + 1); + assert_eq!(ctx.namespace_commit_generation(), before + 2); + assert!(!ctx.namespace_commits_pending()); + } } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 5e6ce9786..1b8d23e58 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -732,6 +732,46 @@ impl Disk { } } + pub(crate) async fn delete_version_with_namespace_owner( + &self, + volume: &str, + path: &str, + fi: FileInfo, + force_del_marker: bool, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + match self { + Self::Local(disk) => { + disk.delete_version_with_namespace_owner(volume, path, fi, force_del_marker, opts, namespace_owner) + .await + } + Self::Remote(disk) => { + let result = disk.delete_version(volume, path, fi, force_del_marker, opts).await; + // This is sender lifetime only, not proof of a remote physical drain. + drop(namespace_owner); + result + } + } + } + + pub(crate) async fn delete_with_namespace_owner( + &self, + volume: &str, + path: &str, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + match self { + Self::Local(disk) => disk.delete_with_namespace_owner(volume, path, opts, namespace_owner).await, + Self::Remote(disk) => { + let result = disk.delete(volume, path, opts).await; + drop(namespace_owner); + result + } + } + } + /// Keep local undo publication owned independently of the wrapper deadline. /// Remote undo retains its existing RPC contract; this is not a remote drain proof. pub(crate) async fn undo_write_with_namespace_owner( diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 9940d36ba..ff9d3d5d6 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -288,6 +288,46 @@ pub(crate) mod prepared_publication_test_hooks { hook(); } } + + #[cfg(test)] + type RenameDestinationHook = Box; + #[cfg(test)] + static RENAME_DESTINATIONS: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + + #[cfg(test)] + pub(crate) struct RenameDestinationGuard(PathBuf); + + #[cfg(test)] + impl Drop for RenameDestinationGuard { + fn drop(&mut self) { + RENAME_DESTINATIONS.lock().remove(&self.0); + } + } + + #[cfg(test)] + pub(crate) fn observe_rename_destination(source: &Path, hook: impl FnOnce(&Path) + Send + 'static) -> RenameDestinationGuard { + assert!( + RENAME_DESTINATIONS + .lock() + .insert(source.to_path_buf(), Box::new(hook)) + .is_none() + ); + RenameDestinationGuard(source.to_path_buf()) + } + + #[cfg(test)] + pub(crate) async fn drain_namespace_key(path: &Path) { + drop(super::acquire_namespace_mutation_lease(path).await); + } + + #[cfg(test)] + pub(super) fn run_rename_destination(source: &Path, destination: &Path) { + let hook = RENAME_DESTINATIONS.lock().remove(source); + if let Some(hook) = hook { + hook(destination); + } + } } #[cfg(all(test, windows))] @@ -1361,7 +1401,7 @@ async fn acquire_namespace_mutation_lease(path: &Path) -> Arc>, ) -> Arc { @@ -1936,6 +1976,42 @@ pub(crate) async fn remove_dir_with_owner( run_blocking_namespace_operation(lease, move || std::fs::remove_dir(path)).await } +/// Preserve raw rename semantics while retaining a counted owner in the syscall. +/// Unlike reliable rename, this never creates parents or retries a missing source. +pub(in crate::disk) async fn rename_with_namespace_owner( + src: &Path, + dst: &Path, + namespace_owner: Option>, +) -> io::Result<()> { + if namespace_owner.is_none() { + return tokio::fs::rename(src, dst).await; + } + let src = src.to_path_buf(); + let dst = dst.to_path_buf(); + let lease = acquire_namespace_mutation_lease_with_owner(&dst, namespace_owner).await; + run_blocking_namespace_operation(lease, move || { + #[cfg(all(test, not(windows)))] + { + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src); + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst); + } + std::fs::rename(src, dst) + }) + .await +} + +pub(in crate::disk) async fn create_dir_all_with_namespace_owner( + path: &Path, + namespace_owner: Option>, +) -> io::Result<()> { + if namespace_owner.is_none() { + return tokio::fs::create_dir_all(path).await; + } + let path = path.to_path_buf(); + let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await; + run_blocking_namespace_operation(lease, move || std::fs::create_dir_all(path)).await +} + #[tracing::instrument(name = "rename_all", level = "debug", skip_all)] pub(crate) async fn rename_all_with_owner( src_file_path: impl AsRef, @@ -2272,6 +2348,8 @@ async fn reliable_rename_inner_with_lease( move || { let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?; #[cfg(all(test, not(windows)))] + prepared_publication_test_hooks::run_rename_destination(&src_file_path, &dst_file_path); + #[cfg(all(test, not(windows)))] { prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src_file_path); prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 4fcb450bd..fed1d0dbe 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -7497,6 +7497,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let transported = delete_file_info_with_replication_transport_metadata(fi); let fi = &transported; let disks = self.disk_inventory().await; + let namespace_owner = (!is_meta_bucketname(bucket)).then(|| self.ctx.begin_namespace_commit()); let write_quorum = disks.len() / 2 + 1; let rollback_dir = Uuid::new_v4(); @@ -7504,10 +7505,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let mut errs = Vec::with_capacity(disks.len()); for disk in disks.iter() { + let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc); futures.push(async move { if let Some(disk) = disk { match disk - .delete_version( + .delete_version_with_namespace_owner( bucket, object, fi.clone(), @@ -7516,6 +7518,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { old_data_dir: Some(rollback_dir), ..Default::default() }, + disk_namespace_owner, ) .await { @@ -7563,10 +7566,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let bucket = bucket.to_string(); let object = object.to_string(); let fi = fi.clone(); + let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc); rollback_futures.push(async move { if should_rollback { if let Err(err) = disk - .delete_version( + .delete_version_with_namespace_owner( &bucket, &object, fi, @@ -7577,6 +7581,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { old_data_dir: Some(rollback_dir), ..Default::default() }, + disk_namespace_owner, ) .await { @@ -7591,7 +7596,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } else { let rollback_path = format!("{object}/{rollback_dir}"); if let Err(err) = disk - .delete( + .delete_with_namespace_owner( &bucket, &rollback_path, DeleteOptions { @@ -7599,6 +7604,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { immediate: true, ..Default::default() }, + disk_namespace_owner, ) .await && err != DiskError::FileNotFound @@ -7617,6 +7623,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } join_all(rollback_futures).await; + drop(namespace_owner); quorum_result } @@ -21044,3 +21051,417 @@ mod body_cache_hook_e2e_tests { ); } } + +#[cfg(test)] +mod single_delete_namespace_owner_tests { + use super::hermetic_set_disks_support::hermetic_set_disks_isolated; + use super::*; + use crate::disk::ReadOptions; + #[cfg(not(windows))] + use crate::disk::STORAGE_FORMAT_FILE; + use crate::object_api::WriteCompletion; + #[cfg(not(windows))] + use tokio::io::AsyncReadExt; + + async fn seed_version(set: &Arc, bucket: &str, object: &str, version: Uuid, body: &[u8]) { + set.put_object( + bucket, + object, + &mut PutObjReader::from_vec(body.to_vec()), + &ObjectOptions { + versioned: true, + version_id: Some(version.to_string()), + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("seed a complete real object version"); + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_advances_namespace_generation_through_cleanup() { + let (dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "single-delete-namespace"; + let object = "last-version"; + for disk in &disks { + disk.make_volume(bucket).await.expect("fixture bucket"); + } + let version = Uuid::new_v4(); + seed_version(&set, bucket, object, version, &vec![0x41; 256 * 1024]).await; + let before = set.ctx.namespace_commit_generation(); + assert!(!set.ctx.namespace_commits_pending()); + let request = FileInfo { + name: object.to_string(), + version_id: Some(version), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + let result = + tokio::time::timeout(Duration::from_secs(10), set.delete_object_version(bucket, object, &request, false)).await; + if !matches!(result, Ok(Ok(()))) { + let retained = dirs.into_iter().map(tempfile::TempDir::keep).collect::>(); + panic!("single delete and cleanup did not finish: {result:?}; retained roots: {retained:?}"); + } + for (disk, dir) in disks.iter().zip(&dirs) { + let result = disk + .read_version("", bucket, object, &version.to_string(), &ReadOptions::default()) + .await; + assert!(matches!(result, Err(DiskError::FileNotFound | DiskError::FileVersionNotFound))); + assert!( + !dir.path().join(bucket).join(object).exists(), + "immediate cleanup must remove the rollback object tree" + ); + } + assert!(!set.ctx.namespace_commits_pending()); + assert_eq!( + set.ctx.namespace_commit_generation(), + before + 2, + "single delete must count one complete root lifetime" + ); + } + + #[cfg(not(windows))] + async fn assert_single_delete_physical_owner(case: &'static str) { + use crate::disk::os::prepared_publication_test_hooks as hooks; + use futures::FutureExt; + + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async { + let (dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "single-delete-physical-owner"; + let first = Uuid::new_v4(); + let second = Uuid::new_v4(); + let first_body = vec![0x51; 256 * 1024]; + let second_body = vec![0x62; 4096]; + let missing = case == "missing-marker"; + let rollback = case == "rollback"; + let last = case == "last-version"; + let cleanup = case == "immediate-cleanup"; + for disk in &disks { + disk.make_volume(bucket).await.expect("fixture bucket"); + } + if !missing { + seed_version(&set, bucket, case, first, &first_body).await; + if !last && !cleanup { + seed_version(&set, bucket, case, second, &second_body).await; + } + } + let request = FileInfo { + name: case.to_string(), + version_id: Some(first), + deleted: missing, + mark_deleted: missing, + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + let before = set.ctx.namespace_commit_generation(); + assert!(!set.ctx.namespace_commits_pending()); + let mut metadata_paths = Vec::new(); + let mut originals = Vec::new(); + let mut cleanup_sources = Vec::new(); + let mut cleanup_parts = Vec::new(); + for disk in &disks { + let crate::disk::Disk::Local(local) = disk.as_ref() else { + panic!("local fixture required"); + }; + let path = local + .get_disk() + .get_object_path_for_io(bucket, case) + .expect("leased IO path") + .join(STORAGE_FORMAT_FILE); + originals.push(if missing { + None + } else { + Some(std::fs::read(&path).expect("seeded raw metadata")) + }); + if cleanup { + let fi = disk.read_version("", bucket, case, &first.to_string(), &ReadOptions::default()) + .await.expect("real non-inline data directory"); + assert!(!fi.inline_data(), "cleanup fixture must have real shard files"); + let data = path.parent().expect("object parent").join(fi.data_dir.expect("data directory").to_string()); + cleanup_parts.push(std::fs::read(data.join("part.1")).expect("real pre-delete shard")); + cleanup_sources.push(data); + } + metadata_paths.push(path); + } + if rollback { + // Two disks apply the real deletion and then error. Undo must + // restore all four disks, including these post-apply failures. + for disk in disks.iter().take(2) { + crate::disk::local::set_delete_version_fail_after_commit(disk.path().as_path(), case); + } + } + let (entered_tx, mut entered_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut guards = Vec::new(); + let mut destination_guards = Vec::new(); + let later_guards = Arc::new(std::sync::Mutex::new(Vec::new())); + let mut releases = Vec::new(); + for (index, path) in metadata_paths.iter().enumerate() { + let tx = entered_tx.clone(); + let (release, rx) = std::sync::mpsc::channel::<()>(); + if last || cleanup { + let source = if cleanup { &cleanup_sources[index] } else { path }; + destination_guards.push(hooks::observe_rename_destination(source, move |destination| { + let _ = tx.send((index, destination.to_path_buf())); + let _ = rx.recv(); + })); + } else { + let hook_path = path.clone(); + let path = path.clone(); + let pause_path = path.clone(); + let later_guards = Arc::clone(&later_guards); + guards.push(hooks::install_at(hooks::Stage::Rename, &hook_path, move || { + let pause = move || { + let _ = tx.send((index, pause_path)); + let _ = rx.recv(); + }; + if rollback { + // This first callback precedes forward metadata publication. + // Arm only the subsequent real backup-restore rename. + let next = hooks::install_at(hooks::Stage::Rename, &path, pause); + later_guards.lock().expect("fixture hook guards").push(next); + } else { + pause(); + } + })); + } + releases.push(release); + } + drop(entered_tx); + let deleting_set = Arc::clone(&set); + let mut delete = + tokio::spawn(async move { deleting_set.delete_object_version(bucket, case, &request, missing).await }); + let mut joined = false; + let mut counts = None; + let mut physical_keys = std::collections::BTreeMap::new(); + let observations = std::panic::AssertUnwindSafe(async { + tokio::time::timeout(Duration::from_secs(10), async { + while physical_keys.len() < 4 { + tokio::select! { + entry = entered_rx.recv() => { + let (index, key) = entry.expect("actual physical delete entry"); + assert!(physical_keys.insert(index, key).is_none()); + } + result = &mut delete => { + joined = true; + panic!("delete returned before physical entry: {result:?}"); + } + } + } + }) + .await + .expect("all four physical mutations must enter"); + let pending_at_entry = set.ctx.namespace_commits_pending(); + let generation_at_entry = set.ctx.namespace_commit_generation(); + for (path, original) in metadata_paths.iter().zip(&originals) { + if missing { + assert!(!path.exists(), "missing marker must still be unpublished at entry"); + } else if cleanup { + assert!(!path.exists(), "cleanup must follow the actual last-version deletion"); + } else { + let bytes = std::fs::read(path).expect("paused metadata is readable"); + let metadata = rustfs_filemeta::FileMeta::load(&bytes).expect("real metadata must parse"); + if !last && !cleanup { + assert!(metadata.find_version(Some(second)).is_ok()); + } + assert_eq!( + metadata.find_version(Some(first)).is_err(), + rollback, + "undo entry must follow actual deletion" + ); + if !rollback { + assert_eq!(Some(&bytes), original.as_ref()); + } + } + } + if rollback { + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let result = tokio::time::timeout(Duration::from_secs(5), &mut delete).await; + joined = result.is_ok(); + let result = result + .expect("ordinary undo deadlines must return") + .expect("delete coordinator must not panic"); + assert!( + matches!(&result, Err(StorageError::InsufficientWriteQuorum(error_bucket, error_object)) if error_bucket == bucket && error_object == case), + "keep the original failed delete quorum: {result:?}" + ); + } else { + delete.abort(); + let result = tokio::time::timeout(Duration::from_secs(5), &mut delete).await; + joined = result.is_ok(); + assert!( + result + .expect("cancelled caller must join") + .expect_err("the caller must be cancelled") + .is_cancelled() + ); + } + counts = Some(( + pending_at_entry, + generation_at_entry, + set.ctx.namespace_commits_pending(), + set.ctx.namespace_commit_generation(), + )); + for path in physical_keys.values() { + assert!( + hooks::drain_namespace_key(path) + .now_or_never() + .is_none(), + "the physical metadata executor must still own its exact key" + ); + } + }) + .catch_unwind() + .await; + + drop(releases); + drop(guards); + drop(destination_guards); + let coordinator_drained = joined || tokio::time::timeout(Duration::from_secs(10), &mut delete).await.is_ok(); + if !coordinator_drained { + delete.abort(); + let _ = tokio::time::timeout(Duration::from_secs(5), &mut delete).await; + } + later_guards.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clear(); + let drains = futures::future::join_all(physical_keys.values().map(|path| { + tokio::time::timeout(Duration::from_secs(5), hooks::drain_namespace_key(path)) + })).await; + let owner_drained = tokio::time::timeout(Duration::from_secs(5), async { + while set.ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await + .is_ok(); + if physical_keys.len() != 4 || !coordinator_drained || !owner_drained || drains.iter().any(|result| result.is_err()) { + let retained = dirs.into_iter().map(tempfile::TempDir::keep).collect::>(); + eprintln!("single delete cleanup incomplete; retained roots: {retained:?}"); + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + panic!("single delete physical cleanup did not drain"); + } + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + for (index, (disk, (path, original))) in disks.iter().zip(metadata_paths.iter().zip(&originals)).enumerate() { + if cleanup { + assert!(!path.exists(), "metadata must remain deleted after cleanup cancellation"); + assert!(!cleanup_sources[index].exists(), "physical cleanup must remove the shard directory"); + assert_eq!( + std::fs::read(physical_keys[&index].join("part.1")).expect("actual trashed shard"), + cleanup_parts[index], + "trash must contain the exact original shard" + ); + continue; + } + if last { + assert!(!path.exists(), "late trash rename must remove the last metadata"); + assert_eq!( + Some(std::fs::read(&physical_keys[&index]).expect("actual trash destination")), + *original, + "last-version trash must contain the exact old metadata" + ); + continue; + } + let bytes = std::fs::read(path).expect("late metadata publication must finish"); + let metadata = rustfs_filemeta::FileMeta::load(&bytes).expect("final metadata must parse"); + if missing { + assert!( + metadata + .find_version(Some(first)) + .expect("the marker must be published") + .1 + .delete_marker + .is_some() + ); + } else { + assert!(metadata.find_version(Some(second)).is_ok()); + assert_eq!(metadata.find_version(Some(first)).is_ok(), rollback); + if rollback { + assert_eq!(Some(&bytes), original.as_ref(), "physical undo must restore exact old metadata"); + } + let fi = disk + .read_version("", bucket, case, &second.to_string(), &ReadOptions { read_data: true, ..Default::default() }) + .await + .expect("remaining version"); + if fi.inline_data() { + assert!(fi.data.as_ref().is_some_and(|data| !data.is_empty()), "remaining inline shard must survive"); + } else { + let parts = disk.check_parts(bucket, case, &fi).await.expect("remaining shard check"); + assert_eq!(parts.results, vec![crate::disk::CHECK_PART_SUCCESS; fi.parts.len()]); + } + } + } + if !missing && !last && !cleanup { + let mut actual = Vec::new(); + let read_opts = ObjectOptions { + version_id: Some(if rollback { first } else { second }.to_string()), + versioned: true, + ..Default::default() + }; + let mut reader = tokio::time::timeout( + Duration::from_secs(5), + set.get_object_reader(bucket, case, None, HeaderMap::new(), &read_opts), + ) + .await + .expect("final GET must finish") + .expect("the surviving version must be readable"); + tokio::time::timeout(Duration::from_secs(5), reader.stream.read_to_end(&mut actual)) + .await + .expect("body must drain") + .expect("read surviving body"); + assert_eq!(actual, if rollback { first_body } else { second_body }); + } + let (pending_at_entry, generation_at_entry, pending_after_return, generation_after_return) = + counts.expect("complete observations"); + assert!( + pending_at_entry && pending_after_return, + "physical single delete outlived namespace accounting: {case}" + ); + assert_eq!(generation_at_entry, before + 1); + assert_eq!(generation_after_return, generation_at_entry); + assert_eq!(set.ctx.namespace_commit_generation(), before + 2); + assert!(!set.ctx.namespace_commits_pending()); + }) + .await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_cancel_keeps_owner_until_immediate_data_cleanup() { + assert_single_delete_physical_owner("immediate-cleanup").await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_cancel_keeps_owner_until_last_version_trash() { + assert_single_delete_physical_owner("last-version").await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_cancel_keeps_owner_until_metadata_rewrite() { + assert_single_delete_physical_owner("remaining-version").await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_cancel_keeps_owner_until_missing_marker_publication() { + assert_single_delete_physical_owner("missing-marker").await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn single_delete_failed_quorum_keeps_owner_until_physical_undo() { + assert_single_delete_physical_owner("rollback").await; + } +}