gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 12:09:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): retain single-delete physical namespace ownership (#7287)

Browse Source
This commit is contained in:
Zhengchao An
2026-09-06 16:56:24 +08:00
committed by GitHub
parent 5b962b6c58
commit 30a0937a7d
5 changed files with 1041 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/disk_store.rs
+40
View File
@@ -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<Arc<dyn Send + Sync>>,
) -> 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<Arc<dyn Send + Sync>>,
) -> 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,
crates/ecstore/src/disk/local.rs
+458 -108
View File
@@ -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<Arc<dyn Send + Sync>>,
) -> 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<Arc<dyn Send + Sync>>,
}
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<Arc<dyn Send + Sync>>,
) -> 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<Arc<dyn Send + Sync>>,
) -> 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<Arc<dyn Send + Sync>>,
) -> Result<DataDirDeleteStatus> {
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<Uuid>,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> 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<bool> {
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<Arc<dyn Send + Sync>>,
) -> Result<bool> {
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<Arc<dyn Send + Sync>>,
) -> 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<bool> {
async fn finish_version_delete(
&self,
volume: &str,
object: &str,
rollback_dir: Uuid,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<bool> {
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<DataDirDeleteStatus> {
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());
}
}
crates/ecstore/src/disk/mod.rs
+40
View File
@@ -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<Arc<dyn Send + Sync>>,
) -> 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<Arc<dyn Send + Sync>>,
) -> 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(
crates/ecstore/src/disk/os.rs
+79 -1
View File
@@ -288,6 +288,46 @@ pub(crate) mod prepared_publication_test_hooks {
hook();
}
}
#[cfg(test)]
type RenameDestinationHook = Box<dyn FnOnce(&Path) + Send>;
#[cfg(test)]
static RENAME_DESTINATIONS: LazyLock<Mutex<HashMap<PathBuf, RenameDestinationHook>>> =
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<NamespaceMutationL
acquire_namespace_mutation_lease_with_owner(path, None).await
}
async fn acquire_namespace_mutation_lease_with_owner(
pub(in crate::disk) async fn acquire_namespace_mutation_lease_with_owner(
path: &Path,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Arc<NamespaceMutationLease> {
@@ -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<Arc<dyn Send + Sync>>,
) -> 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<Arc<dyn Send + Sync>>,
) -> 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<Path>,
@@ -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);
crates/ecstore/src/set_disk/ops/object.rs
+424 -3
View File
@@ -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<dyn Send + Sync>);
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<dyn Send + Sync>);
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<SetDisks>, 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::<Vec<_>>();
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::<Vec<_>>();
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;
}
}