From b0c73c1224fdd7041055116932456e996cdf26bd Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 6 Sep 2026 10:51:38 +0800 Subject: [PATCH] fix(ecstore): retain namespace owners through local disk completion (#7245) * fix(ecstore): retain namespace owners through local physical tails (cherry picked from commit a2f242463316e87604feadbdac5e4148140e72c0) * test(ecstore): expose stale fsync group cleanup * fix(ecstore): capture complete fsync worker guard (cherry picked from commit 1dc90bb836e20ea9ee45d0a629a9201e20d231c4) * fix(ecstore): preserve successor fsync group registration (cherry picked from commit c7dfaad90526052e56c57dafffa4813bdcde46ca) * test(ecstore): mark physical owner fixtures as inline * test(ecstore): wait for namespace owner release before asserting The namespace owner tests decided that ownership had ended when the Weak probe stopped upgrading or when the mutation lease could be reacquired. Both signals fire before the owner guard's Drop decrements the pending counter: Arc releases its strong count before running Drop, and the lease drops its locks before its owner field. The rio-v2 lane hit that window in undo_fresh_version_keeps_physical_namespace_owner_after_timeout. Extend every drain wait to also require namespace_commits_pending() to be false, so the assertions observe the completed release instead of racing it. --- crates/ecstore/src/disk/disk_store.rs | 68 +- crates/ecstore/src/disk/local.rs | 1316 ++++++++++++----- crates/ecstore/src/disk/local/commit.rs | 52 +- crates/ecstore/src/disk/mod.rs | 33 +- crates/ecstore/src/disk/os.rs | 571 ++++++- .../src/set_disk/core/io_primitives.rs | 379 ++++- 6 files changed, 2002 insertions(+), 417 deletions(-) diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 46487c8c9..a11594997 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -324,6 +324,30 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper { } impl LocalDiskWrapper { + pub(in crate::disk) async fn undo_write_with_namespace_owner( + &self, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + self.track_disk_health_mutation( + "delete_version", + DiskMetricMutation::Delete, + || async { + // Preserve the old DiskAPI future's boxing boundary. + Box::pin( + self.disk + .undo_write_with_namespace_owner(volume, path, fi, opts, namespace_owner), + ) + .await + }, + get_max_timeout_duration(), + ) + .await + } + pub(in crate::disk) async fn rename_data_observed( &self, src_volume: &str, @@ -333,6 +357,34 @@ impl LocalDiskWrapper { dst_path: &str, external_guard: Option>, ) -> super::RenameDataObservation { + self.rename_data_observed_with_guards( + src_volume, + src_path, + fi, + dst_volume, + dst_path, + super::RenameDataGuards { + external_guard, + ..Default::default() + }, + ) + .await + } + + pub(in crate::disk) async fn rename_data_observed_with_guards( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + guards: super::RenameDataGuards, + ) -> super::RenameDataObservation { + let super::RenameDataGuards { + external_guard, + namespace_owner, + .. + } = guards; let operation = self.clone(); let src_volume = src_volume.to_owned(); let src_path = src_path.to_owned(); @@ -357,13 +409,15 @@ impl LocalDiskWrapper { DiskMetricMutation::Write, || async { // Preserve the former DiskAPI future's single boxing boundary. - let observed = - Box::pin( - operation - .disk - .rename_data_observed(&src_volume, &src_path, &fi, &dst_volume, &dst_path), - ) - .await; + let observed = Box::pin(operation.disk.rename_data_observed( + &src_volume, + &src_path, + &fi, + &dst_volume, + &dst_path, + namespace_owner, + )) + .await; preflight_rejection = observed.preflight_rejection; observed.result }, diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 1fd76d943..c24c90724 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -204,15 +204,25 @@ async fn restore_metadata_backup( xl_path: &Path, rollback_dir: Uuid, publication_root: &os::PublicationRoot, +) -> Result<()> { + restore_metadata_backup_with_namespace_owner(object_dir, xl_path, rollback_dir, publication_root, None).await +} + +async fn restore_metadata_backup_with_namespace_owner( + object_dir: &Path, + xl_path: &Path, + rollback_dir: Uuid, + publication_root: &os::PublicationRoot, + namespace_owner: Option>, ) -> Result<()> { let rollback_path = object_dir.join(rollback_dir.to_string()); let backup_path = rollback_path.join(STORAGE_FORMAT_FILE_BACKUP); - rename_all(&backup_path, xl_path, object_dir, publication_root).await?; + os::rename_all_with_owner(&backup_path, xl_path, object_dir, publication_root, namespace_owner.clone()).await?; // A synthetic inline rollback dir held only the backup the rename above // just consumed; reclaim it so the object dir can empty out. A real data // dir still holds its parts, so the non-recursive remove is a benign // no-op there (mirrors restore_delete_rollback). - let _ = fs::remove_dir(&rollback_path).await; + let _ = os::remove_dir_with_owner(&rollback_path, namespace_owner.clone()).await; Ok(()) } @@ -222,7 +232,17 @@ async fn restore_delete_rollback( rollback_dir: Uuid, publication_root: &os::PublicationRoot, ) -> Result<()> { - remove_version_delete_markers(object_dir, rollback_dir).await?; + restore_delete_rollback_with_namespace_owner(object_dir, xl_path, rollback_dir, publication_root, None).await +} + +async fn restore_delete_rollback_with_namespace_owner( + object_dir: &Path, + xl_path: &Path, + rollback_dir: Uuid, + publication_root: &os::PublicationRoot, + namespace_owner: Option>, +) -> Result<()> { + remove_version_delete_markers(object_dir, rollback_dir, namespace_owner.clone()).await?; let rollback_path = object_dir.join(rollback_dir.to_string()); let mut staged_paths = Vec::new(); let mut remove_new_metadata = false; @@ -243,34 +263,38 @@ async fn restore_delete_rollback( let had_staged_paths = !staged_paths.is_empty(); for (src, dst) in staged_paths { - rename_all(&src, &dst, object_dir, publication_root).await?; + os::rename_all_with_owner(&src, &dst, object_dir, publication_root, namespace_owner.clone()).await?; } let backup_path = rollback_path.join(STORAGE_FORMAT_FILE_BACKUP); - match rename_all(&backup_path, xl_path, object_dir, publication_root).await { + match os::rename_all_with_owner(&backup_path, xl_path, object_dir, publication_root, namespace_owner.clone()).await { Ok(()) => { - let _ = fs::remove_dir(&rollback_path).await; + let _ = os::remove_dir_with_owner(&rollback_path, namespace_owner.clone()).await; Ok(()) } // A missing backup only means "remove the newly-created delete marker" // when the marker proves there was no old metadata to restore. - Err(DiskError::FileNotFound) if remove_new_metadata => match fs::remove_file(xl_path).await { - Ok(()) => { - let _ = fs::remove_file(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE)).await; - let _ = fs::remove_dir(&rollback_path).await; - Ok(()) + Err(DiskError::FileNotFound) if remove_new_metadata => { + match os::remove_file_with_owner(xl_path, namespace_owner.clone()).await { + Ok(()) => { + let _ = os::remove_file_with_owner(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE), namespace_owner.clone()) + .await; + let _ = os::remove_dir_with_owner(&rollback_path, namespace_owner.clone()).await; + Ok(()) + } + Err(err) if err.kind() == ErrorKind::NotFound => { + let _ = os::remove_file_with_owner(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE), namespace_owner.clone()) + .await; + let _ = os::remove_dir_with_owner(&rollback_path, namespace_owner.clone()).await; + Ok(()) + } + Err(err) => Err(to_file_error(err).into()), } - Err(err) if err.kind() == ErrorKind::NotFound => { - let _ = fs::remove_file(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE)).await; - let _ = fs::remove_dir(&rollback_path).await; - Ok(()) - } - Err(err) => Err(to_file_error(err).into()), - }, + } Err(DiskError::FileNotFound) if had_staged_paths => Err(DiskError::FileNotFound), Err(DiskError::FileNotFound) => match fs::metadata(xl_path).await { Ok(_) => { - let _ = fs::remove_dir(&rollback_path).await; + let _ = os::remove_dir_with_owner(&rollback_path, namespace_owner.clone()).await; Ok(()) } Err(err) if err.kind() == ErrorKind::NotFound => Err(DiskError::FileNotFound), @@ -280,7 +304,11 @@ async fn restore_delete_rollback( } } -async fn remove_version_delete_markers(object_dir: &Path, rollback_dir: Uuid) -> Result<()> { +async fn remove_version_delete_markers( + object_dir: &Path, + rollback_dir: Uuid, + namespace_owner: Option>, +) -> Result<()> { let reserved_name = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}"); let committed_name = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}"); let mut entries = match fs::read_dir(object_dir).await { @@ -295,7 +323,7 @@ async fn remove_version_delete_markers(object_dir: &Path, rollback_dir: Uuid) -> continue; } for marker_name in [&reserved_name, &committed_name] { - match fs::remove_file(entry.path().join(marker_name)).await { + match os::remove_file_with_owner(entry.path().join(marker_name), namespace_owner.clone()).await { Ok(()) => {} Err(err) if err.kind() == ErrorKind::NotFound => {} Err(err) => return Err(to_file_error(err).into()), @@ -305,6 +333,12 @@ async fn remove_version_delete_markers(object_dir: &Path, rollback_dir: Uuid) -> Ok(()) } +struct DeleteVersionMutation { + force_del_marker: bool, + opts: DeleteOptions, + namespace_owner: Option>, +} + struct DeleteRollbackFailure { stage: &'static str, error: DiskError, @@ -5620,7 +5654,377 @@ impl LocalDisk { // }) // } + async fn delete_version_inner(&self, volume: &str, path: &str, fi: FileInfo, mutation: DeleteVersionMutation) -> Result<()> { + let DeleteVersionMutation { + force_del_marker, + opts, + namespace_owner, + } = mutation; + if path.starts_with(SLASH_SEPARATOR) { + return self + .delete_with_namespace_owner( + volume, + path, + DeleteOptions { + recursive: false, + immediate: false, + ..Default::default() + }, + namespace_owner, + ) + .await; + } + + let volume_dir = self.io_get_bucket_path(volume)?; + + let file_path = self.io_get_object_path(volume, path)?; + + check_path_length(file_path.to_string_lossy().as_ref())?; + + let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]); + if opts.old_data_dir.is_some() && opts.undo_write { + return self.undo_write(file_path.as_path(), &fi, &opts, namespace_owner).await; + } + + let rollback_dir = opts.old_data_dir; + let buf = match self.read_all_data(volume, &volume_dir, &xl_path).await { + Ok(res) => res, + Err(err) => { + if err != DiskError::FileNotFound { + return Err(err); + } + + if fi.deleted && force_del_marker { + return self + .write_missing_delete_marker(volume, path, fi, file_path.as_path(), &xl_path, rollback_dir) + .await; + } + + return if fi.version_id.is_some() { + Err(DiskError::FileVersionNotFound) + } else { + Err(DiskError::FileNotFound) + }; + } + }; + + let mut meta = FileMeta::load(&buf)?; + 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?; + } + + if let Some(uuid) = old_dir { + let vid = fi.version_id.unwrap_or_default(); + if let Err(err) = meta.data.remove(vec![vid, uuid]) { + let err: DiskError = err.into(); + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_metadata_update", + error: err, + }, + &self.publication_root, + ) + .await); + } + + let old_path = path_join(&[file_path.as_path(), Path::new(uuid.to_string().as_str())]); + if let Err(err) = check_path_length(old_path.to_string_lossy().as_ref()) { + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_data_path", + error: err, + }, + &self.publication_root, + ) + .await); + } + + 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 { + let err: DiskError = to_file_error(err).into(); + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_rollback_dir", + error: err, + }, + &self.publication_root, + ) + .await); + } + reserved_version_delete = match self.reserve_version_delete(volume, path, uuid, rollback_dir).await { + Ok(reserved) => reserved, + Err(err) => { + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_reserve_data", + error: err, + }, + &self.publication_root, + ) + .await); + } + }; + 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 + { + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_stage_data", + error: err, + }, + &self.publication_root, + ) + .await); + } + if should_fail_after_delete_data_staged(path) { + if reserved_version_delete { + return Err(self + .abort_reserved_version_delete( + file_path.as_path(), + rollback_dir, + volume, + path, + "delete_version_test_after_stage", + DiskError::Unexpected, + ) + .await); + } + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_test_after_stage", + error: DiskError::Unexpected, + }, + &self.publication_root, + ) + .await); + } + } else if let Err(err) = self + .move_to_trash_with_namespace_owner(&old_path, true, false, namespace_owner.clone()) + .await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + return Err(err); + } + + // The version's data dir was staged for rollback or trashed, so its + // `part.N` inodes no longer exist for readers. A cached io_uring + // descriptor would keep serving them, so drop every cached fd under + // this data dir (rustfs/backlog#1175). If a later rollback restores + // the dir, the next read simply re-opens it. + self.io_backend.invalidate_cached_fds_under(volume, &format!("{path}/{uuid}")); + } + + let commit_result = if !meta.versions.is_empty() { + let buf = match meta.marshal_msg() { + Ok(buf) => buf, + Err(err) => { + let err: DiskError = err.into(); + 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_metadata_encode", + err, + ) + .await); + } + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_metadata_encode", + error: err, + }, + &self.publication_root, + ) + .await); + } + }; + self.write_all_meta_with_namespace_owner( + volume, + format!("{path}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE}").as_str(), + &buf, + true, + namespace_owner.clone(), + ) + .await + } else { + self.delete_file_with_namespace_owner(&volume_dir, &xl_path, true, false, namespace_owner.clone()) + .await + }; + + 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) + .await); + } + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + DeleteRollbackFailure { + stage: "delete_version_commit", + error: err, + }, + &self.publication_root, + ) + .await); + } + + if reserved_version_delete + && let Some(rollback_dir) = rollback_dir + && let Err(err) = self.commit_reserved_version_delete(volume, path, rollback_dir).await + { + return Err(self + .abort_reserved_version_delete( + file_path.as_path(), + rollback_dir, + volume, + path, + "delete_version_commit_intent", + err, + ) + .await); + } + + if should_fail_after_delete_commit(self.root.as_path(), path) { + return Err(DiskError::Unexpected); + } + + Ok(()) + } + + #[tracing::instrument(name = "delete_version", level = "trace", skip_all)] + pub(in crate::disk) async fn undo_write_with_namespace_owner( + &self, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + // This entry is reserved for rollback, not general version deletion. + if !opts.undo_write { + return Err(DiskError::FileCorrupt); + } + self.delete_version_inner( + volume, + path, + fi, + DeleteVersionMutation { + force_del_marker: false, + opts, + namespace_owner, + }, + ) + .await + } + + async fn undo_write( + &self, + file_path: &Path, + fi: &FileInfo, + opts: &DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + let old_data_dir = opts.old_data_dir.ok_or(DiskError::FileCorrupt)?; + let xl_path = path_join(&[file_path, Path::new(STORAGE_FORMAT_FILE)]); + if opts.undo_delete { + restore_delete_rollback_with_namespace_owner( + file_path, + &xl_path, + old_data_dir, + &self.publication_root, + namespace_owner.clone(), + ) + .await?; + } else { + restore_metadata_backup_with_namespace_owner( + file_path, + &xl_path, + old_data_dir, + &self.publication_root, + namespace_owner.clone(), + ) + .await?; + } + + if !opts.undo_delete + && let Some(new_data_dir) = fi.data_dir + { + let new_data_path = path_join(&[file_path, Path::new(new_data_dir.to_string().as_str())]); + check_path_length(new_data_path.to_string_lossy().as_ref())?; + if let Err(err) = self + .move_to_trash_with_namespace_owner(&new_data_path, true, false, namespace_owner) + .await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + return Err(err); + } + } + + Ok(()) + } + async fn move_to_trash(&self, delete_path: &PathBuf, recursive: bool, immediate_purge: bool) -> Result<()> { + self.move_to_trash_with_namespace_owner(delete_path, recursive, immediate_purge, None) + .await + } + + async fn move_to_trash_with_namespace_owner( + &self, + delete_path: &PathBuf, + recursive: bool, + immediate_purge: bool, + namespace_owner: Option>, + ) -> Result<()> { // if recursive { // remove_all_std(delete_path).map_err(to_volume_error)?; // } else { @@ -5639,11 +6043,12 @@ impl LocalDisk { // } let err = if recursive { - rename_all_ignore_missing_source( + os::rename_all_ignore_missing_source_with_owner( delete_path, trash_path, self.io_get_bucket_path(RUSTFS_META_TMP_DELETED_BUCKET)?, &self.publication_root, + namespace_owner.clone(), ) .await .err() @@ -5661,11 +6066,12 @@ impl LocalDisk { if immediate_purge || delete_path.to_string_lossy().ends_with(SLASH_SEPARATOR) { let trash_path2 = self.io_get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?; - let _ = rename_all_ignore_missing_source( + let _ = os::rename_all_ignore_missing_source_with_owner( encode_dir_object(delete_path.to_string_lossy().as_ref()), trash_path2, self.io_get_bucket_path(RUSTFS_META_TMP_DELETED_BUCKET)?, &self.publication_root, + namespace_owner.clone(), ) .await; } @@ -5691,7 +6097,44 @@ impl LocalDisk { Ok(()) } + #[tracing::instrument(name = "delete", level = "trace", skip_all)] + async fn delete_with_namespace_owner( + &self, + volume: &str, + path: &str, + opt: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + crate::hp_guard!("LocalDisk::delete"); + let handled_version_delete = if opt.recursive + && opt.immediate + && 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? + } else { + false + }; + match self + .delete_unleased_with_namespace_owner(volume, path, &opt, namespace_owner) + .await + { + Err(DiskError::FileNotFound) if handled_version_delete => Ok(()), + result => result, + } + } + async fn delete_unleased(&self, volume: &str, path: &str, opt: &DeleteOptions) -> Result<()> { + self.delete_unleased_with_namespace_owner(volume, path, opt, None).await + } + + async fn delete_unleased_with_namespace_owner( + &self, + volume: &str, + path: &str, + opt: &DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { let volume_dir = self.io_get_bucket_path(volume)?; if !skip_access_checks(volume) && let Err(e) = cached_access(&volume_dir).await @@ -5701,21 +6144,33 @@ impl LocalDisk { let file_path = self.io_get_object_path(volume, path)?; check_path_length(file_path.to_string_lossy().as_ref())?; - self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate) + self.delete_file_with_namespace_owner(&volume_dir, &file_path, opt.recursive, opt.immediate, namespace_owner) .await?; // A deleted shard must not remain readable through the io_uring fd cache. self.io_backend.invalidate_cached_fds_under(volume, path); Ok(()) } - #[tracing::instrument(level = "trace", skip_all)] - #[async_recursion::async_recursion] async fn delete_file( &self, base_path: &PathBuf, delete_path: &PathBuf, recursive: bool, immediate_purge: bool, + ) -> Result<()> { + self.delete_file_with_namespace_owner(base_path, delete_path, recursive, immediate_purge, None) + .await + } + + #[tracing::instrument(name = "delete_file", level = "trace", skip_all)] + #[async_recursion::async_recursion] + async fn delete_file_with_namespace_owner( + &self, + base_path: &PathBuf, + delete_path: &PathBuf, + recursive: bool, + immediate_purge: bool, + namespace_owner: Option>, ) -> Result<()> { // debug!("delete_file {:?}\n base_path:{:?}", &delete_path, &base_path); @@ -5730,10 +6185,11 @@ impl LocalDisk { } if recursive { - self.move_to_trash(delete_path, recursive, immediate_purge).await?; + self.move_to_trash_with_namespace_owner(delete_path, recursive, immediate_purge, namespace_owner.clone()) + .await?; } else if delete_path.is_dir() { // debug!("delete_file remove_dir {:?}", &delete_path); - if let Err(err) = fs::remove_dir(&delete_path).await { + if let Err(err) = os::remove_dir_with_owner(delete_path, namespace_owner.clone()).await { // debug!("remove_dir err {:?} when {:?}", &err, &delete_path); // A missing or still-populated directory is benign here; see // is_benign_object_rmdir_error (handles the illumos/Solaris EEXIST @@ -5755,7 +6211,7 @@ impl LocalDisk { } } // debug!("delete_file remove_dir done {:?}", &delete_path); - } else if let Err(err) = fs::remove_file(&delete_path).await { + } else if let Err(err) = os::remove_file_with_owner(delete_path, namespace_owner.clone()).await { // debug!("remove_file err {:?} when {:?}", &err, &delete_path); match err.kind() { ErrorKind::NotFound => (), @@ -5778,7 +6234,14 @@ impl LocalDisk { } if let Some(dir_path) = delete_path.parent() { - Box::pin(self.delete_file(base_path, &PathBuf::from(dir_path), false, false)).await?; + Box::pin(self.delete_file_with_namespace_owner( + base_path, + &PathBuf::from(dir_path), + false, + false, + namespace_owner.clone(), + )) + .await?; } // debug!("delete_file done {:?}", &delete_path); @@ -6365,6 +6828,17 @@ impl LocalDisk { } async fn write_all_meta(&self, volume: &str, path: &str, buf: &[u8], sync: bool) -> Result<()> { + self.write_all_meta_with_namespace_owner(volume, path, buf, sync, None).await + } + + async fn write_all_meta_with_namespace_owner( + &self, + volume: &str, + path: &str, + buf: &[u8], + sync: bool, + namespace_owner: Option>, + ) -> Result<()> { let volume_dir = self.io_get_bucket_path(volume)?; let file_path = self.io_get_object_path(volume, path)?; check_path_length(file_path.to_string_lossy().as_ref())?; @@ -6397,13 +6871,15 @@ impl LocalDisk { return Err(DiskError::Unexpected); } - rename_all(tmp_file_path, &file_path, volume_dir, &self.publication_root).await?; + os::rename_all_with_owner(tmp_file_path, &file_path, volume_dir, &self.publication_root, namespace_owner.clone()).await?; if sync && durability.syncs_commit_metadata() && let Some(parent) = file_path.parent() { - os::fsync_dir(parent).await.map_err(to_file_error)?; + os::fsync_dir_with_owner(parent, namespace_owner) + .await + .map_err(to_file_error)?; } Ok(()) @@ -8015,22 +8491,8 @@ impl DiskAPI for LocalDisk { LocalDisk::has_replacement_mount_lease(self) } - #[tracing::instrument(level = "trace", skip_all)] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { - crate::hp_guard!("LocalDisk::delete"); - let handled_version_delete = if opt.recursive - && opt.immediate - && 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? - } else { - false - }; - match self.delete_unleased(volume, path, &opt).await { - Err(DiskError::FileNotFound) if handled_version_delete => Ok(()), - result => result, - } + self.delete_with_namespace_owner(volume, path, opt, None).await } #[tracing::instrument(level = "trace", skip_all)] @@ -8888,7 +9350,7 @@ impl DiskAPI for LocalDisk { dst_volume: &str, dst_path: &str, ) -> Result { - self.rename_data_inner(src_volume, src_path, fi, dst_volume, dst_path, &mut None) + self.rename_data_inner(src_volume, src_path, fi, dst_volume, dst_path, &mut Default::default()) .await } @@ -9405,295 +9867,17 @@ impl DiskAPI for LocalDisk { force_del_marker: bool, opts: DeleteOptions, ) -> Result<()> { - if path.starts_with(SLASH_SEPARATOR) { - return self - .delete( - volume, - path, - DeleteOptions { - recursive: false, - immediate: false, - ..Default::default() - }, - ) - .await; - } - - let volume_dir = self.io_get_bucket_path(volume)?; - - let file_path = self.io_get_object_path(volume, path)?; - - check_path_length(file_path.to_string_lossy().as_ref())?; - - let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]); - if let Some(old_data_dir) = opts.old_data_dir - && opts.undo_write - { - if opts.undo_delete { - restore_delete_rollback(file_path.as_path(), &xl_path, old_data_dir, &self.publication_root).await?; - } else { - restore_metadata_backup(file_path.as_path(), &xl_path, old_data_dir, &self.publication_root).await?; - } - - if !opts.undo_delete - && let Some(new_data_dir) = fi.data_dir - { - let new_data_path = path_join(&[file_path.as_path(), Path::new(new_data_dir.to_string().as_str())]); - check_path_length(new_data_path.to_string_lossy().as_ref())?; - if let Err(err) = self.move_to_trash(&new_data_path, true, false).await - && err != DiskError::FileNotFound - && err != DiskError::VolumeNotFound - { - return Err(err); - } - } - - return Ok(()); - } - - let rollback_dir = opts.old_data_dir; - let buf = match self.read_all_data(volume, &volume_dir, &xl_path).await { - Ok(res) => res, - Err(err) => { - if err != DiskError::FileNotFound { - return Err(err); - } - - if fi.deleted && force_del_marker { - return self - .write_missing_delete_marker(volume, path, fi, file_path.as_path(), &xl_path, rollback_dir) - .await; - } - - return if fi.version_id.is_some() { - Err(DiskError::FileVersionNotFound) - } else { - Err(DiskError::FileNotFound) - }; - } - }; - - let mut meta = FileMeta::load(&buf)?; - 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?; - } - - if let Some(uuid) = old_dir { - let vid = fi.version_id.unwrap_or_default(); - if let Err(err) = meta.data.remove(vec![vid, uuid]) { - let err: DiskError = err.into(); - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - rollback_dir, - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_metadata_update", - error: err, - }, - &self.publication_root, - ) - .await); - } - - let old_path = path_join(&[file_path.as_path(), Path::new(uuid.to_string().as_str())]); - if let Err(err) = check_path_length(old_path.to_string_lossy().as_ref()) { - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - rollback_dir, - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_data_path", - error: err, - }, - &self.publication_root, - ) - .await); - } - - 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 { - let err: DiskError = to_file_error(err).into(); - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - Some(rollback_dir), - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_rollback_dir", - error: err, - }, - &self.publication_root, - ) - .await); - } - reserved_version_delete = match self.reserve_version_delete(volume, path, uuid, rollback_dir).await { - Ok(reserved) => reserved, - Err(err) => { - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - Some(rollback_dir), - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_reserve_data", - error: err, - }, - &self.publication_root, - ) - .await); - } - }; - 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 - { - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - Some(rollback_dir), - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_stage_data", - error: err, - }, - &self.publication_root, - ) - .await); - } - if should_fail_after_delete_data_staged(path) { - if reserved_version_delete { - return Err(self - .abort_reserved_version_delete( - file_path.as_path(), - rollback_dir, - volume, - path, - "delete_version_test_after_stage", - DiskError::Unexpected, - ) - .await); - } - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - Some(rollback_dir), - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_test_after_stage", - error: DiskError::Unexpected, - }, - &self.publication_root, - ) - .await); - } - } else if let Err(err) = self.move_to_trash(&old_path, true, false).await - && err != DiskError::FileNotFound - && err != DiskError::VolumeNotFound - { - return Err(err); - } - - // The version's data dir was staged for rollback or trashed, so its - // `part.N` inodes no longer exist for readers. A cached io_uring - // descriptor would keep serving them, so drop every cached fd under - // this data dir (rustfs/backlog#1175). If a later rollback restores - // the dir, the next read simply re-opens it. - self.io_backend.invalidate_cached_fds_under(volume, &format!("{path}/{uuid}")); - } - - let commit_result = if !meta.versions.is_empty() { - let buf = match meta.marshal_msg() { - Ok(buf) => buf, - Err(err) => { - let err: DiskError = err.into(); - 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_metadata_encode", - err, - ) - .await); - } - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - rollback_dir, - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_metadata_encode", - error: err, - }, - &self.publication_root, - ) - .await); - } - }; - self.write_all_meta(volume, format!("{path}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE}").as_str(), &buf, true) - .await - } else { - self.delete_file(&volume_dir, &xl_path, true, false).await - }; - - 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) - .await); - } - return Err(restore_delete_rollback_after_error( - file_path.as_path(), - &xl_path, - rollback_dir, - volume, - path, - DeleteRollbackFailure { - stage: "delete_version_commit", - error: err, - }, - &self.publication_root, - ) - .await); - } - - if reserved_version_delete - && let Some(rollback_dir) = rollback_dir - && let Err(err) = self.commit_reserved_version_delete(volume, path, rollback_dir).await - { - return Err(self - .abort_reserved_version_delete( - file_path.as_path(), - rollback_dir, - volume, - path, - "delete_version_commit_intent", - err, - ) - .await); - } - - if should_fail_after_delete_commit(self.root.as_path(), path) { - return Err(DiskError::Unexpected); - } - - Ok(()) + self.delete_version_inner( + volume, + path, + fi, + DeleteVersionMutation { + force_del_marker, + opts, + namespace_owner: None, + }, + ) + .await } #[tracing::instrument(level = "trace", skip_all)] async fn delete_versions(&self, volume: &str, versions: Vec, opts: DeleteOptions) -> Vec> { @@ -12984,6 +13168,464 @@ mod test { ); } + #[cfg(not(windows))] + async fn assert_undo_physical_namespace_owner(case: &str, cancel_waiter: bool) { + use crate::disk::{disk_store::LocalDiskWrapper, os::prepared_publication_test_hooks as hooks}; + + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async { + 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 = "physical-undo"; + let object = format!("object-{case}"); + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_DELETED_BUCKET).await; + let object_dir = disk.io_get_object_path(bucket, &object).expect("object IO path"); + let xl_path = object_dir.join(STORAGE_FORMAT_FILE); + let old_version = Uuid::new_v4(); + let mut old = test_file_info(&object, old_version, None, Some(Bytes::from_static(b"old-payload"))); + old.set_inline_data(); + let old_meta = test_meta(old.clone()); + let new_version = if case == "backup" { old_version } else { Uuid::new_v4() }; + let mut new = test_file_info(&object, new_version, None, Some(Bytes::from_static(b"new-payload"))); + new.set_inline_data(); + let rollback_dir = Uuid::new_v4(); + let mut opts = DeleteOptions { + undo_write: true, + ..Default::default() + }; + fs::create_dir_all(&object_dir).await.expect("object directory"); + let stage = match case { + "backup" => { + fs::write(&xl_path, test_meta(new.clone())).await.expect("current metadata"); + write_metadata_rollback_backup(&object_dir, rollback_dir, &old_meta) + .await + .expect("rollback backup"); + opts.old_data_dir = Some(rollback_dir); + hooks::Stage::Rename + } + "marker" => { + new = FileInfo { + name: object.clone(), + version_id: Some(new_version), + deleted: true, + mark_deleted: true, + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + disk.delete_version( + bucket, + &object, + new.clone(), + true, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("create the real no-backup delete-marker rollback intent"); + assert!( + object_dir + .join(rollback_dir.to_string()) + .join(DELETE_MARKER_ROLLBACK_FILE) + .exists() + ); + opts.old_data_dir = Some(rollback_dir); + opts.undo_delete = true; + hooks::Stage::Remove + } + "fresh" => { + fs::write(&xl_path, test_meta(new.clone())) + .await + .expect("new version metadata"); + hooks::Stage::Rename + } + "remaining" => { + let mut meta = FileMeta::load(&old_meta).expect("old metadata parses"); + meta.add_version(new.clone()).expect("add distinct new version"); + fs::write(&xl_path, meta.marshal_msg().expect("encode both versions")) + .await + .expect("versioned metadata"); + hooks::Stage::Rename + } + _ => panic!("unknown undo fixture"), + }; + let before = fs::read(&xl_path).await.expect("metadata exists before undo"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install_at(stage, &xl_path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let owner_probe = Arc::downgrade(&owner); + let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false); + let mut undo = Box::pin(wrapper.undo_write_with_namespace_owner(bucket, &object, new, opts, Some(owner))); + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + entered = entered_rx => entered.expect("physical undo must signal entry"), + _ = undo.as_mut() => panic!("undo returned before its physical publication"), + } + }) + .await + .expect("undo must enter its real filesystem executor"); + assert_eq!(std::fs::read(&xl_path).expect("pre-publication metadata"), before); + assert!(ctx.namespace_commits_pending()); + if cancel_waiter { + drop(undo); + } else { + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let error = tokio::time::timeout(Duration::from_secs(5), undo) + .await + .expect("ordinary undo timeout must not wait for its syscall") + .expect_err("the blocked undo must time out"); + assert_eq!(error, DiskError::Timeout); + } + let pending_while_blocked = ctx.namespace_commits_pending(); + let owner_alive_while_blocked = owner_probe.upgrade().is_some(); + let generation_before_publication = ctx.namespace_commit_generation(); + assert_eq!(std::fs::read(&xl_path).expect("still blocked metadata"), before); + drop(release_tx); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let published = match case { + "backup" => std::fs::read(&xl_path).is_ok_and(|data| data == old_meta), + "marker" | "fresh" => !xl_path.exists(), + "remaining" => std::fs::read(&xl_path) + .ok() + .and_then(|data| FileMeta::load(&data).ok()) + .is_some_and(|meta| { + meta.find_version(Some(old_version)).is_ok() && meta.find_version(Some(new_version)).is_err() + }), + _ => unreachable!(), + }; + if published && owner_probe.upgrade().is_none() && !ctx.namespace_commits_pending() { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("released physical undo must publish and release ownership"); + if matches!(case, "backup" | "remaining") { + let restored = disk + .read_version( + "", + bucket, + &object, + &old_version.to_string(), + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("the old version remains readable after physical undo"); + assert_eq!(restored.data.as_deref(), Some(b"old-payload".as_slice())); + } + assert!( + pending_while_blocked && owner_alive_while_blocked, + "undo {case} lost namespace ownership while physical publication was pending" + ); + assert!(!ctx.namespace_commits_pending()); + assert!(ctx.namespace_commit_generation() > generation_before_publication); + }) + .await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn undo_backup_restore_keeps_physical_namespace_owner_after_cancellation() { + assert_undo_physical_namespace_owner("backup", true).await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn undo_delete_marker_removal_keeps_physical_namespace_owner_after_timeout() { + assert_undo_physical_namespace_owner("marker", false).await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn undo_fresh_version_keeps_physical_namespace_owner_after_timeout() { + assert_undo_physical_namespace_owner("fresh", false).await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn undo_version_rewrite_keeps_physical_namespace_owner_after_cancellation() { + assert_undo_physical_namespace_owner("remaining", true).await; + } + + #[cfg(not(windows))] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[serial_test::serial(capacity_dirty_scope)] + async fn inline_rollback_keeps_namespace_owner_while_cancellation_is_requested() { + use crate::disk::{disk_store::LocalDiskWrapper, os::prepared_publication_test_hooks as hooks}; + 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 = "physical-internal-rollback"; + let object = "physical-owner-inline-internal-rollback-object"; + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + let object_dir = disk.io_get_object_path(bucket, object).expect("object IO path"); + let xl_path = object_dir.join(STORAGE_FORMAT_FILE); + fs::create_dir_all(&object_dir).await.expect("object directory"); + let version_id = Uuid::new_v4(); + let mut old = test_file_info(object, version_id, None, Some(Bytes::from_static(b"old-payload"))); + old.set_inline_data(); + let old_meta = test_meta(old); + fs::write(&xl_path, &old_meta).await.expect("old metadata"); + set_rename_data_fail_after_metadata_commit(object); + let mut new = test_file_info(object, version_id, None, Some(Bytes::from_static(b"new-payload"))); + new.set_inline_data(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install_at(hooks::Stage::Rollback, &xl_path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let owner_probe = Arc::downgrade(&owner); + let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false); + let mut rename = tokio::spawn(async move { + wrapper + .rename_data_observed_with_guards( + RUSTFS_META_TMP_BUCKET, + "source", + &new, + bucket, + object, + crate::disk::RenameDataGuards { + namespace_owner: Some(owner), + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + entered = entered_rx => entered.expect("internal rollback must signal entry"), + _ = &mut rename => panic!("rename returned before the physical internal rollback"), + } + }) + .await + .expect("post-commit fault must reach the physical rollback"); + let published = disk + .read_version( + "", + bucket, + object, + &version_id.to_string(), + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("new metadata was really published before rollback"); + assert_eq!(published.data.as_deref(), Some(b"new-payload".as_slice())); + rename.abort(); + let pending = ctx.namespace_commits_pending(); + let alive = owner_probe.upgrade().is_some(); + let generation = ctx.namespace_commit_generation(); + drop(release_tx); + tokio::time::timeout(Duration::from_secs(5), async { + while !std::fs::read(&xl_path).is_ok_and(|data| data == old_meta) + || owner_probe.upgrade().is_some() + || ctx.namespace_commits_pending() + { + tokio::task::yield_now().await; + } + }) + .await + .expect("physical rollback must restore the old bytes"); + let _cancelled = rename.await; + assert!(pending && alive, "cancelled internal rollback must retain its namespace owner"); + assert!(!ctx.namespace_commits_pending()); + assert!(ctx.namespace_commit_generation() > generation); + } + + #[cfg(not(windows))] + async fn assert_physical_namespace_owner_and_quota_claim(pause_at: &str) { + use crate::disk::{disk_store::LocalDiskWrapper, os::prepared_publication_test_hooks as hooks}; + let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(pause_at == "fsync"); + let _durability = durability_mode_override::set(DurabilityMode::Strict); + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async { + 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 = "physical-quota-coexistence"; + let object = "object"; + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + let object_dir = disk.io_get_object_path(bucket, object).expect("object IO path"); + let xl_path = object_dir.join(STORAGE_FORMAT_FILE); + let fence_path = quota_mutation_fence_path(bucket, object); + let token = disk + .acquire_snapshot_lease(RUSTFS_META_BUCKET, &fence_path) + .await + .expect("quota fence"); + let fence = disk + .snapshot_leases + .lock() + .await + .entries + .get(&SnapshotLeaseKey { + volume: RUSTFS_META_BUCKET.to_string(), + path: fence_path.clone(), + }) + .and_then(|entry| entry.mutation_fence.clone()) + .expect("real quota fence state"); + let version_id = Uuid::new_v4(); + let staged_backup = disk + .io_get_object_path(RUSTFS_META_TMP_BUCKET, "source/xl.meta.bkp") + .expect("staged backup IO path"); + if pause_at != "prepared" { + let mut old = test_file_info(object, version_id, None, Some(Bytes::from_static(b"old-payload"))); + old.set_inline_data(); + fs::create_dir_all(&object_dir).await.expect("existing object directory"); + fs::write(&xl_path, test_meta(old)) + .await + .expect("old metadata requiring a rollback backup"); + } + let mut new = test_file_info(object, version_id, None, Some(Bytes::from_static(b"new-payload"))); + new.set_inline_data(); + rustfs_utils::http::metadata_compat::insert_str( + &mut new.metadata, + super::super::QUOTA_MUTATION_FENCE_METADATA_SUFFIX, + token.as_uuid().to_string(), + ); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let (stage, pause_path) = match pause_at { + "backup" => (hooks::Stage::Rename, staged_backup.clone()), + // Group fsync keys use canonical paths, including on Linux mount-FD IO paths. + "fsync" => (hooks::Stage::DirFsync, object_dir.canonicalize().expect("canonical group fsync path")), + "prepared" => (hooks::Stage::PreparedRename, xl_path.clone()), + _ => panic!("unknown physical quota pause"), + }; + let _hook = hooks::install_at(stage, &pause_path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let owner_probe = Arc::downgrade(&owner); + let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false); + let mut rename = Box::pin(wrapper.rename_data_observed_with_guards( + RUSTFS_META_TMP_BUCKET, + "source", + &new, + bucket, + object, + crate::disk::RenameDataGuards { + namespace_owner: Some(owner), + ..Default::default() + }, + )); + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + entered = entered_rx => entered.expect("physical publication entry"), + _ = rename.as_mut() => panic!("rename returned before publication"), + } + }) + .await + .expect("publication must enter with a real quota claim"); + assert_eq!(fence.running.load(Ordering::Acquire), 1); + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let observed = tokio::time::timeout(Duration::from_secs(5), rename) + .await + .expect("namespace owner must not select the external-guard unlimited deadline"); + assert!(!observed.rejected_before_publication()); + assert_eq!(observed.result.expect_err("ordinary timeout"), DiskError::Timeout); + assert_eq!( + fence.running.load(Ordering::Acquire), + 1, + "namespace owner must not replace the quota claim" + ); + assert!(ctx.namespace_commits_pending()); + assert!(owner_probe.upgrade().is_some()); + let generation = ctx.namespace_commit_generation(); + let mut revoke = + Box::pin(disk.release_snapshot_lease(RUSTFS_META_BUCKET, &fence_path, SnapshotLeaseToken::revoke_all())); + assert!(futures::poll!(tokio::task::unconstrained(revoke.as_mut())).is_pending()); + assert!(fence.revoked.load(Ordering::Acquire), "revoke must actually enter its claim-drain wait"); + drop(release_tx); + tokio::time::timeout(Duration::from_secs(5), revoke) + .await + .expect("quota claim drains after syscall") + .expect("revoke"); + tokio::time::timeout(Duration::from_secs(5), async { + while owner_probe.upgrade().is_some() || ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await + .expect("namespace owner drains alongside quota claim"); + assert_eq!(fence.running.load(Ordering::Acquire), 0); + assert!(!ctx.namespace_commits_pending()); + assert!(ctx.namespace_commit_generation() > generation); + let stored = disk + .read_version( + "", + bucket, + object, + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("late publication remains readable"); + let expected = if pause_at == "backup" { + b"old-payload".as_slice() + } else { + b"new-payload".as_slice() + }; + assert_eq!(stored.data.as_deref(), Some(expected)); + if pause_at == "backup" { + assert!(!staged_backup.exists(), "the detached sibling lease must really publish the backup"); + } + }) + .await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope, dst_dir_fsync_group_commit)] + async fn physical_namespace_owner_and_quota_claim_both_survive_rename_timeout() { + assert_physical_namespace_owner_and_quota_claim("prepared").await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope, dst_dir_fsync_group_commit)] + async fn rollback_backup_sibling_lease_keeps_namespace_owner_and_quota_after_timeout() { + assert_physical_namespace_owner_and_quota_claim("backup").await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope, dst_dir_fsync_group_commit)] + async fn grouped_fsync_keeps_physical_namespace_owner_and_quota_after_timeout() { + assert_physical_namespace_owner_and_quota_claim("fsync").await; + } + #[tokio::test] async fn observed_rename_timeout_has_no_preflight_proof_and_retains_namespace_lease() { use crate::disk::disk_store::LocalDiskWrapper; diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs index 94c57d313..ccb950c27 100644 --- a/crates/ecstore/src/disk/local/commit.rs +++ b/crates/ecstore/src/disk/local/commit.rs @@ -34,7 +34,7 @@ use crate::disk::{ error::{DiskError, Result}, error_conv::{to_access_error, to_file_error}, os, - os::{check_path_length, rename_all}, + os::check_path_length, }; use bytes::Bytes; use rustfs_filemeta::{FileInfo, FileMeta}; @@ -74,6 +74,8 @@ fn rollback_inline_metadata_commit_std( rollback_data_dir: Option, local_rollback_path: Option<&Path>, ) -> std::io::Result<()> { + #[cfg(all(test, not(windows)))] + os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Rollback, dst_file_path); if let Some(backup_path) = local_rollback_path { // The commit immediately before this rollback renamed the staged // xl.meta from the same directory as `backup_path` onto @@ -233,6 +235,12 @@ async fn restore_published_data_source( #[derive(Debug)] pub(in crate::disk) struct LocalRenamePreflightRejection(()); +#[derive(Default)] +pub(super) struct RenameDataState { + namespace_owner: Option>, + preflight_rejection: Option, +} + impl LocalDisk { #[tracing::instrument(name = "rename_data", target = "rustfs_ecstore::disk::local", level = "trace", skip_all)] pub(super) async fn rename_data_inner( @@ -242,7 +250,7 @@ impl LocalDisk { fi: FileInfo, dst_volume: &str, dst_path: &str, - preflight_rejection: &mut Option, + state: &mut RenameDataState, ) -> Result { crate::hp_guard!("LocalDisk::rename_data"); let mut fi = fi; @@ -271,7 +279,13 @@ impl LocalDisk { Some(token) => Some(self.claim_quota_mutation_fence(dst_volume, dst_path, token).await?), None => None, }; - let mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, dst_volume, &destination_object_path).await; + let mutation_lease = os::acquire_rename_data_mutation_lease_with_owner( + &self.root, + dst_volume, + &destination_object_path, + state.namespace_owner.take(), + ) + .await; if let Some(claim) = quota_fence_claim { mutation_lease.attach_external_guard(claim); } @@ -304,7 +318,7 @@ impl LocalDisk { error = %e, "Disk local access check failed" ); - *preflight_rejection = Some(LocalRenamePreflightRejection(())); + state.preflight_rejection = Some(LocalRenamePreflightRejection(())); return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } @@ -322,7 +336,7 @@ impl LocalDisk { error = %e, "Disk local access check failed" ); - *preflight_rejection = Some(LocalRenamePreflightRejection(())); + state.preflight_rejection = Some(LocalRenamePreflightRejection(())); return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } @@ -530,7 +544,9 @@ impl LocalDisk { // rename below. if fi_healing && let Some((_, dst_data_path)) = has_data_dir_path.as_ref() - && let Err(err) = self.move_to_trash(dst_data_path, true, false).await + && let Err(err) = self + .move_to_trash_with_namespace_owner(dst_data_path, true, false, Some(mutation_lease.clone())) + .await { warn!( target: "rustfs_ecstore::disk::local", @@ -757,7 +773,7 @@ impl LocalDisk { && let Some(parent) = dst_file_path.parent() { let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = os::fsync_dst_dir_group_commit(parent).await { + if let Err(err) = os::fsync_dst_dir_group_commit(parent, Some(mutation_lease.clone())).await { rustfs_io_metrics::record_put_object_stage_duration_from( rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, fsync_started, @@ -795,7 +811,7 @@ impl LocalDisk { break; } let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = os::fsync_dir(dir).await { + if let Err(err) = os::fsync_dir_with_owner(dir, Some(mutation_lease.clone())).await { rustfs_io_metrics::record_put_object_stage_duration_from( rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, fsync_started, @@ -1026,7 +1042,15 @@ impl LocalDisk { // rename_all acquires the backup path's namespace lease. Do not // hold a disk admission while acquiring another namespace lock. drop(file_sync_admission.take()); - if let Err(err) = rename_all(staged_backup, &backup_path, &dst_volume_dir, &self.publication_root).await { + if let Err(err) = os::rename_all_with_owner( + staged_backup, + &backup_path, + &dst_volume_dir, + &self.publication_root, + Some(mutation_lease.clone()), + ) + .await + { let _ = remove_file_if_exists(staged_backup); return Err(err); } @@ -1222,14 +1246,18 @@ impl LocalDisk { fi: &FileInfo, dst_volume: &str, dst_path: &str, + namespace_owner: Option>, ) -> super::super::RenameDataObservation { - let mut preflight_rejection = None; + let mut state = RenameDataState { + namespace_owner, + ..Default::default() + }; let result = self - .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut preflight_rejection) + .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut state) .await; super::super::RenameDataObservation { result, - preflight_rejection, + preflight_rejection: state.preflight_rejection, } } } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index aac24b781..5e6ce9786 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -75,6 +75,14 @@ use time::OffsetDateTime; use tokio::io::{AsyncRead, AsyncWrite}; use uuid::Uuid; +/// Independent admission and physical ownership for one disk rename. +#[derive(Default)] +pub(crate) struct RenameDataGuards { + pub(crate) scanner_publication_lease_token: Option, + pub(crate) external_guard: Option>, + pub(crate) namespace_owner: Option>, +} + /// Local preflight evidence stays outside DiskAPI and the RPC response format. pub(crate) struct RenameDataObservation { pub(crate) result: Result, @@ -724,6 +732,25 @@ impl Disk { } } + /// 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( + &self, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, + namespace_owner: Option>, + ) -> Result<()> { + match self { + Self::Local(disk) => { + disk.undo_write_with_namespace_owner(volume, path, fi, opts, namespace_owner) + .await + } + Self::Remote(disk) => disk.delete_version(volume, path, fi, false, opts).await, + } + } + pub(crate) async fn rename_data_borrowed( &self, src_volume: &str, @@ -743,12 +770,12 @@ impl Disk { fi: &FileInfo, dst_volume: &str, dst_path: &str, - scanner_publication_lease_token: Option, + guards: RenameDataGuards, ) -> RenameDataObservation { match self { Disk::Local(local_disk) => { local_disk - .rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None) + .rename_data_observed_with_guards(src_volume, src_path, fi, dst_volume, dst_path, guards) .await } Disk::Remote(remote_disk) => RenameDataObservation::unknown( @@ -759,7 +786,7 @@ impl Disk { fi, dst_volume, dst_path, - scanner_publication_lease_token, + guards.scanner_publication_lease_token, ) .await, ), diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index a0f5e0580..9940d36ba 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -246,6 +246,50 @@ pub(crate) mod fsync_dir_recorder { } } +/// Pause a real namespace mutation inside its physical executor. +#[cfg(all(test, not(windows)))] +pub(crate) mod prepared_publication_test_hooks { + use super::*; + + #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] + pub(crate) enum Stage { + PreparedRename, + Rename, + Remove, + Rollback, + DirFsync, + } + + type Hook = Box; + type Key = (Stage, PathBuf); + static BEFORE_PUBLICATION: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); + + pub(crate) struct Guard(Key); + + impl Drop for Guard { + fn drop(&mut self) { + BEFORE_PUBLICATION.lock().remove(&self.0); + } + } + + pub(crate) fn install(path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard { + install_at(Stage::PreparedRename, path, hook) + } + + pub(crate) fn install_at(stage: Stage, path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard { + let key = (stage, path.to_path_buf()); + assert!(BEFORE_PUBLICATION.lock().insert(key.clone(), Box::new(hook)).is_none()); + Guard(key) + } + + pub(crate) fn run(stage: Stage, path: &Path) { + let hook = BEFORE_PUBLICATION.lock().remove(&(stage, path.to_path_buf())); + if let Some(hook) = hook { + hook(); + } + } +} + #[cfg(all(test, windows))] pub(crate) mod windows_rename_test_hooks { use super::*; @@ -580,6 +624,7 @@ impl OpenedDstDirFsyncGroup { } struct DstDirFsyncWaiter { + namespace_owner: Option>, result_tx: oneshot::Sender, } @@ -638,6 +683,7 @@ impl DstDirFsyncGroupCommit { fn enqueue_opened( &self, opened: OpenedDstDirFsyncGroup, + namespace_owner: Option>, ) -> io::Result<(oneshot::Receiver, Option>)> { let (result_tx, result_rx) = oneshot::channel(); let mut registry = self.inner.lock(); @@ -668,7 +714,10 @@ impl DstDirFsyncGroupCommit { group }; let mut group_state = group.inner.lock(); - group_state.pending.push_back(DstDirFsyncWaiter { result_tx }); + group_state.pending.push_back(DstDirFsyncWaiter { + result_tx, + namespace_owner, + }); let start_worker = !group_state.worker_running; if start_worker { group_state.worker_running = true; @@ -690,7 +739,13 @@ impl DstDirFsyncGroupCommit { fn remove_idle_group(&self, group: &Arc) { let mut registry = self.inner.lock(); let group_state = group.inner.lock(); - if !group_state.worker_running && group_state.pending.is_empty() { + if !group_state.worker_running + && group_state.pending.is_empty() + && registry + .groups + .get(&group.key) + .is_some_and(|registered| Arc::ptr_eq(registered, group)) + { registry.groups.remove(&group.key); } } @@ -713,16 +768,20 @@ impl DstDirFsyncGroupCommit { &self, dir: &Path, ) -> io::Result<(oneshot::Receiver, Option>)> { - self.enqueue_opened(OpenedDstDirFsyncGroup::open(dir)?) + self.enqueue_opened(OpenedDstDirFsyncGroup::open(dir)?, None) } } #[cfg(unix)] -async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> { +async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup, namespace_owners: Vec>) -> io::Result<()> { #[cfg(test)] let dir = group.dir.clone(); let dir_file = group.dir_file.clone(); fsync_spawn_blocking(move || { + // The batch worker may be cancelled while this syscall is still running. + let _namespace_owners = namespace_owners; + #[cfg(all(test, not(windows)))] + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::DirFsync, &dir); #[cfg(test)] { if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) { @@ -737,66 +796,118 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> { } #[cfg(not(unix))] -async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> { +async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup, namespace_owners: Vec>) -> io::Result<()> { + let _namespace_owners = namespace_owners; fsync_dir(&group.dir).await } -async fn run_dst_dir_fsync_group_worker(group: Arc) { - loop { - #[cfg(test)] - fsync_dir_recorder::run_before_group_batch(&group.dir); - tokio::task::yield_now().await; - let batch: Vec = { - let mut group_state = group.inner.lock(); - group_state.pending.drain(..).collect() - }; - if batch.is_empty() { - let mut group_state = group.inner.lock(); +struct DstDirFsyncWorkerGuard { + group: Arc, + in_flight: usize, + armed: bool, +} + +impl Drop for DstDirFsyncWorkerGuard { + fn drop(&mut self) { + if !self.armed { + return; + } + // Cancellation must release queued owners, but the physical batch keeps + // its own owners until its blocking syscall returns. + let pending = { + let mut registry = DST_DIR_FSYNC_GROUP_COMMIT.inner.lock(); + let mut group_state = self.group.inner.lock(); + let pending = std::mem::take(&mut group_state.pending); group_state.worker_running = false; - drop(group_state); - DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group); - return; - } - - #[cfg(test)] - fsync_dir_recorder::record_grouped(&group.dir, batch.len()); - let result = fsync_open_dst_dir_group(&group) - .await - .map_err(SharedDstDirFsyncError::from_error); - let batch_len = batch.len(); - DST_DIR_FSYNC_GROUP_COMMIT.complete_batch(batch_len); - - let should_stop = { - let mut group_state = group.inner.lock(); - if group_state.pending.is_empty() { - group_state.worker_running = false; - true - } else { - false + if registry + .groups + .get(&self.group.key) + .is_some_and(|group| Arc::ptr_eq(group, &self.group)) + { + registry.total_waiters = registry.total_waiters.saturating_sub(pending.len() + self.in_flight); + registry.groups.remove(&self.group.key); } + pending }; - if should_stop { - DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group); - } - for waiter in batch { - let _ = waiter.result_tx.send(result.clone()); - } - if should_stop { - return; + // Lease and channel destructors must run outside the registry locks. + drop(pending); + } +} + +fn run_dst_dir_fsync_group_worker(group: Arc) -> impl std::future::Future { + // Capture before spawning: shutdown may drop the future without polling it. + let worker_guard = DstDirFsyncWorkerGuard { + group: group.clone(), + in_flight: 0, + armed: true, + }; + async move { + let mut worker_guard = worker_guard; + loop { + #[cfg(test)] + fsync_dir_recorder::run_before_group_batch(&group.dir); + tokio::task::yield_now().await; + let mut batch: Vec = { + let mut group_state = group.inner.lock(); + group_state.pending.drain(..).collect() + }; + if batch.is_empty() { + let mut group_state = group.inner.lock(); + worker_guard.armed = false; + group_state.worker_running = false; + drop(group_state); + DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group); + return; + } + worker_guard.in_flight = batch.len(); + + #[cfg(test)] + fsync_dir_recorder::record_grouped(&group.dir, batch.len()); + let namespace_owners = batch.iter_mut().filter_map(|waiter| waiter.namespace_owner.take()).collect(); + let result = fsync_open_dst_dir_group(&group, namespace_owners) + .await + .map_err(SharedDstDirFsyncError::from_error); + let batch_len = batch.len(); + DST_DIR_FSYNC_GROUP_COMMIT.complete_batch(batch_len); + worker_guard.in_flight = 0; + + let should_stop = { + let mut group_state = group.inner.lock(); + if group_state.pending.is_empty() { + worker_guard.armed = false; + group_state.worker_running = false; + true + } else { + false + } + }; + if should_stop { + DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group); + } + for waiter in batch { + let _ = waiter.result_tx.send(result.clone()); + } + if should_stop { + return; + } } } } -async fn fsync_dst_dir_group_commit_with_enabled(dir: impl AsRef, enabled: bool) -> io::Result<()> { +async fn fsync_dst_dir_group_commit_with_enabled( + dir: impl AsRef, + enabled: bool, + namespace_owner: Option>, +) -> io::Result<()> { if !enabled { - return fsync_dir(dir).await; + return fsync_dir_with_owner(dir.as_ref(), namespace_owner).await; } let dir = dir.as_ref().to_path_buf(); let opened = tokio::task::spawn_blocking(move || OpenedDstDirFsyncGroup::open(&dir)) .await .map_err(|err| io::Error::other(format!("blocking dst dir group open failed: {err}")))??; - let (result_rx, worker) = DST_DIR_FSYNC_GROUP_COMMIT.enqueue_opened(opened)?; + let (result_rx, worker) = DST_DIR_FSYNC_GROUP_COMMIT.enqueue_opened(opened, namespace_owner)?; if let Some(group) = worker { tokio::spawn(run_dst_dir_fsync_group_worker(group)); } @@ -808,8 +919,11 @@ async fn fsync_dst_dir_group_commit_with_enabled(dir: impl AsRef, enabled: } } -pub(crate) async fn fsync_dst_dir_group_commit(dir: impl AsRef) -> io::Result<()> { - fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled()).await +pub(crate) async fn fsync_dst_dir_group_commit( + dir: impl AsRef, + namespace_owner: Option>, +) -> io::Result<()> { + fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled(), namespace_owner).await } pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit( @@ -818,7 +932,7 @@ pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit( admission: &FileSyncAdmission, ) -> io::Result<()> { if dst_dir_fsync_group_commit_enabled() { - fsync_dst_dir_group_commit_with_enabled(dir, true).await + fsync_dst_dir_group_commit_with_enabled(dir, true, Some(lease)).await } else { fsync_dir_with_namespace_file_sync_limit(dir, lease, admission).await } @@ -826,7 +940,7 @@ pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit( #[cfg(test)] pub(crate) async fn fsync_dst_dir_group_commit_for_test(dir: impl AsRef, enabled: bool) -> io::Result<()> { - fsync_dst_dir_group_commit_with_enabled(dir, enabled).await + fsync_dst_dir_group_commit_with_enabled(dir, enabled, None).await } #[cfg(test)] @@ -1233,6 +1347,8 @@ pub(crate) struct NamespaceMutationLease { _namespace_guard: OwnedMutexGuard<()>, _volume_guard: Option>, external_guard: Mutex>>, + // Independent of the quota claim; both survive cancellation of the waiter. + _namespace_owner: Option>, } impl NamespaceMutationLease { @@ -1242,10 +1358,18 @@ impl NamespaceMutationLease { } async fn acquire_namespace_mutation_lease(path: &Path) -> Arc { + acquire_namespace_mutation_lease_with_owner(path, None).await +} + +async fn acquire_namespace_mutation_lease_with_owner( + path: &Path, + namespace_owner: Option>, +) -> Arc { Arc::new(NamespaceMutationLease { _namespace_guard: disk_namespace_mutation_lock(path).lock_owned().await, _volume_guard: None, external_guard: Mutex::new(None), + _namespace_owner: namespace_owner, }) } @@ -1255,6 +1379,15 @@ pub(crate) async fn acquire_rename_data_mutation_lease( root: &Path, volume: &str, destination_object: &Path, +) -> Arc { + acquire_rename_data_mutation_lease_with_owner(root, volume, destination_object, None).await +} + +pub(crate) async fn acquire_rename_data_mutation_lease_with_owner( + root: &Path, + volume: &str, + destination_object: &Path, + namespace_owner: Option>, ) -> Arc { let namespace_guard = disk_namespace_mutation_lock(destination_object).lock_owned().await; let volume_guard = disk_volume_mutation_lock(root, volume).read_owned().await; @@ -1262,6 +1395,7 @@ pub(crate) async fn acquire_rename_data_mutation_lease( _namespace_guard: namespace_guard, _volume_guard: Some(volume_guard), external_guard: Mutex::new(None), + _namespace_owner: namespace_owner, }) } @@ -1751,6 +1885,69 @@ pub async fn rename_all( Ok(()) } +pub(crate) async fn fsync_dir_with_owner(path: &Path, namespace_owner: Option>) -> io::Result<()> { + #[cfg(unix)] + { + if namespace_owner.is_none() { + return fsync_dir(path).await; + } + let path = path.to_path_buf(); + fsync_spawn_blocking(move || { + let _namespace_owner = namespace_owner; + fsync_dir_std(path) + }) + .await? + } + #[cfg(not(unix))] + { + let _ = namespace_owner; + fsync_dir(path).await + } +} + +/// Retain namespace ownership in the actual filesystem executor after timeout. +pub(crate) async fn remove_file_with_owner( + path: impl AsRef, + namespace_owner: Option>, +) -> io::Result<()> { + if namespace_owner.is_none() { + return tokio::fs::remove_file(path).await; + } + let path = path.as_ref().to_path_buf(); + let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await; + run_blocking_namespace_operation(lease, move || { + #[cfg(all(test, not(windows)))] + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Remove, &path); + std::fs::remove_file(path) + }) + .await +} + +/// Retain namespace ownership in the actual filesystem executor after timeout. +pub(crate) async fn remove_dir_with_owner( + path: impl AsRef, + namespace_owner: Option>, +) -> io::Result<()> { + if namespace_owner.is_none() { + return tokio::fs::remove_dir(path).await; + } + let path = path.as_ref().to_path_buf(); + let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await; + run_blocking_namespace_operation(lease, move || std::fs::remove_dir(path)).await +} + +#[tracing::instrument(name = "rename_all", level = "debug", skip_all)] +pub(crate) async fn rename_all_with_owner( + src_file_path: impl AsRef, + dst_file_path: impl AsRef, + base_dir: impl AsRef, + publication_root: &PublicationRoot, + namespace_owner: Option>, +) -> Result<()> { + let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await; + rename_all_with_lease(src_file_path, dst_file_path, base_dir, publication_root, lease).await +} + pub(crate) async fn rename_all_with_lease( src_file_path: impl AsRef, dst_file_path: impl AsRef, @@ -1943,6 +2140,8 @@ pub(crate) async fn rename_all_with_prepared_source( move || { validate_prepared_rename_source(&prepared_source, &src_file_path)?; let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?; + #[cfg(test)] + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path); rename_prepared(&src_file_path, &dst_file_path, &preparation) } }; @@ -1981,6 +2180,32 @@ pub async fn rename_all_ignore_missing_source( } } +#[tracing::instrument(name = "rename_all_ignore_missing_source", level = "debug", skip_all)] +pub(crate) async fn rename_all_ignore_missing_source_with_owner( + src_file_path: impl AsRef, + dst_file_path: impl AsRef, + base_dir: impl AsRef, + publication_root: &PublicationRoot, + namespace_owner: Option>, +) -> Result<()> { + let src_file_path = src_file_path.as_ref(); + let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await; + match reliable_rename_inner_with_lease( + src_file_path.to_path_buf(), + dst_file_path.as_ref().to_path_buf(), + base_dir.as_ref().to_path_buf(), + publication_root.clone(), + false, + lease, + ) + .await + { + Ok(()) => Ok(()), + Err(err) if err.kind() == io::ErrorKind::NotFound && rename_source_is_missing(src_file_path, publication_root) => Ok(()), + Err(err) => Err(to_file_error(err).into()), + } +} + #[cfg(windows)] pub(crate) fn rename_source_is_missing(src_file_path: &Path, publication_root: &PublicationRoot) -> bool { let Some(source_parent) = src_file_path.parent() else { @@ -2046,6 +2271,11 @@ async fn reliable_rename_inner_with_lease( let base_dir = base_dir.clone(); 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(prepared_publication_test_hooks::Stage::Rename, &src_file_path); + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path); + } rename_prepared(&src_file_path, &dst_file_path, &preparation) } }; @@ -6140,6 +6370,245 @@ mod tests { wait_for_dst_dir_fsync_group_commit_idle().await; } + #[cfg(unix)] + #[tokio::test] + #[serial_test::serial(dst_dir_fsync_group_commit)] + async fn grouped_fsync_physical_batch_keeps_all_owners_after_worker_cancellation() { + let temp_dir = tempdir().expect("fixture directory"); + let dir = temp_dir.path().canonicalize().expect("canonical fsync path"); + let first_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let second_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let first_owner = first_ctx.begin_namespace_commit(); + let second_owner = second_ctx.begin_namespace_commit(); + let first_probe = Arc::downgrade(&first_owner); + let second_probe = Arc::downgrade(&second_owner); + let (first_rx, group) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_opened( + OpenedDstDirFsyncGroup::open(&dir).expect("open first waiter directory"), + Some(first_owner), + ) + .expect("queue first real waiter"); + let group = group.expect("first waiter starts the group"); + let (second_rx, second_worker) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_opened( + OpenedDstDirFsyncGroup::open(&dir).expect("open second waiter directory"), + Some(second_owner), + ) + .expect("queue second real waiter"); + assert!(second_worker.is_none(), "same directory must join the same batch"); + assert_eq!(group.inner.lock().pending.len(), 2); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = + prepared_publication_test_hooks::install_at(prepared_publication_test_hooks::Stage::DirFsync, &dir, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let worker = tokio::spawn(run_dst_dir_fsync_group_worker(group.clone())); + tokio::time::timeout(Duration::from_secs(5), entered_rx) + .await + .expect("batch must reach its physical fsync") + .expect("physical fsync entry"); + assert_eq!(fsync_dir_recorder::grouped_batch_sizes(&dir), vec![2]); + assert!( + group.inner.lock().pending.is_empty(), + "both waiters were transferred into the physical batch" + ); + let queued_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let queued_owner = queued_ctx.begin_namespace_commit(); + let queued_probe = Arc::downgrade(&queued_owner); + let queued_generation = queued_ctx.namespace_commit_generation(); + let (queued_rx, queued_worker) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_opened( + OpenedDstDirFsyncGroup::open(&dir).expect("open queued waiter directory"), + Some(queued_owner), + ) + .expect("queue a waiter after the physical batch was frozen"); + assert!(queued_worker.is_none()); + assert_eq!(group.inner.lock().pending.len(), 1); + drop((first_rx, second_rx)); + worker.abort(); + assert!(worker.await.expect_err("cancel the async batch owner").is_cancelled()); + assert!(queued_rx.await.is_err(), "an undispatched waiter must observe worker cancellation"); + assert!(queued_probe.upgrade().is_none()); + assert!(!queued_ctx.namespace_commits_pending()); + assert!(queued_ctx.namespace_commit_generation() > queued_generation); + assert!(group.inner.lock().pending.is_empty()); + assert!(!group.inner.lock().worker_running); + assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0)); + let first_pending = first_ctx.namespace_commits_pending() && first_probe.upgrade().is_some(); + let second_pending = second_ctx.namespace_commits_pending() && second_probe.upgrade().is_some(); + let generations = (first_ctx.namespace_commit_generation(), second_ctx.namespace_commit_generation()); + drop(release_tx); + tokio::time::timeout(Duration::from_secs(5), async { + while Arc::strong_count(&group.dir_file) != 1 + || first_probe.upgrade().is_some() + || second_probe.upgrade().is_some() + || first_ctx.namespace_commits_pending() + || second_ctx.namespace_commits_pending() + { + tokio::task::yield_now().await; + } + }) + .await + .expect("physical fsync must release every batch owner"); + assert!(fsync_dir_recorder::was_fsynced(&dir), "the detached syscall must really execute"); + assert!( + first_pending && second_pending, + "one physical batch must preserve both independent namespace owners" + ); + assert!(!first_ctx.namespace_commits_pending()); + assert!(!second_ctx.namespace_commits_pending()); + assert!(first_ctx.namespace_commit_generation() > generations.0); + assert!(second_ctx.namespace_commit_generation() > generations.1); + } + + #[cfg(unix)] + #[tokio::test] + #[serial_test::serial(dst_dir_fsync_group_commit)] + async fn grouped_fsync_unpolled_worker_releases_queued_owner() { + let temp_dir = tempdir().expect("fixture directory"); + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let probe = Arc::downgrade(&owner); + let generation = ctx.namespace_commit_generation(); + let (rx, group) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_opened( + OpenedDstDirFsyncGroup::open(temp_dir.path()).expect("open queued waiter directory"), + Some(owner), + ) + .expect("queue a real waiter"); + let group = group.expect("first waiter starts the group"); + let worker = run_dst_dir_fsync_group_worker(group.clone()); + assert!(ctx.namespace_commits_pending()); + drop(worker); + assert!(rx.await.is_err(), "shutdown before first poll must release the waiter"); + assert!(probe.upgrade().is_none()); + assert!(!ctx.namespace_commits_pending()); + assert!(ctx.namespace_commit_generation() > generation); + assert!(group.inner.lock().pending.is_empty()); + assert!(!group.inner.lock().worker_running); + assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0)); + assert!( + fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()).is_empty(), + "the dropped future must not dispatch a physical batch" + ); + } + + #[cfg(unix)] + #[test] + fn stale_idle_group_cleanup_preserves_successor_registration() { + let temp_dir = tempdir().expect("fixture directory"); + let registry = DstDirFsyncGroupCommit::default(); + let (mut first_rx, first_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue first worker"); + let old_group = first_worker.expect("first waiter starts a worker"); + // W1 has completed its batch and marked G idle, but has not cleaned G up. + let first_waiter = old_group.inner.lock().pending.pop_front().expect("first batch waiter"); + registry.complete_batch(1); + old_group.inner.lock().worker_running = false; + + let (mut second_rx, second_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue second worker"); + let reused_group = second_worker.expect("idle G starts another worker"); + assert!(Arc::ptr_eq(&old_group, &reused_group)); + let second_waiter = reused_group.inner.lock().pending.pop_front().expect("second batch waiter"); + registry.complete_batch(1); + reused_group.inner.lock().worker_running = false; + registry.remove_idle_group(&reused_group); + assert_eq!(registry.counts_for_test(), (0, 0), "normal idle cleanup must remove G"); + assert!(second_waiter.result_tx.send(Ok(())).is_ok()); + assert!(second_rx.try_recv().expect("second worker reports completion").is_ok()); + + let (mut successor_rx, successor_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue successor"); + let successor = successor_worker.expect("successor starts a new group"); + assert!(!Arc::ptr_eq(&old_group, &successor)); + assert_eq!(registry.counts_for_test(), (1, 1)); + // W1 resumes with its old Arc after W2 removed G and W3 installed G2. + registry.remove_idle_group(&old_group); + assert!(first_waiter.result_tx.send(Ok(())).is_ok()); + assert!(first_rx.try_recv().expect("first worker reports completion").is_ok()); + assert!( + registry + .inner + .lock() + .groups + .get(&successor.key) + .is_some_and(|registered| Arc::ptr_eq(registered, &successor)), + "stale cleanup must retain the exact successor Arc" + ); + assert_eq!(registry.counts_for_test(), (1, 1)); + assert!(successor.inner.lock().worker_running); + assert_eq!(successor.inner.lock().pending.len(), 1); + assert!(matches!(successor_rx.try_recv(), Err(oneshot::error::TryRecvError::Empty))); + + let (_joined_rx, new_worker) = registry.enqueue_for_test(temp_dir.path()).expect("join successor"); + assert!(new_worker.is_none(), "a later waiter must join G2 instead of creating G3"); + assert_eq!(successor.inner.lock().pending.len(), 2); + assert_eq!(registry.counts_for_test(), (1, 2)); + } + + #[cfg(unix)] + #[tokio::test] + #[serial_test::serial(dst_dir_fsync_group_commit)] + async fn stale_idle_cleanup_then_unpolled_worker_drop_releases_waiter_budget() { + wait_for_dst_dir_fsync_group_commit_idle().await; + let temp_dir = tempdir().expect("fixture directory"); + let (old_rx, old_worker) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_for_test(temp_dir.path()) + .expect("enqueue old group"); + let old_group = old_worker.expect("old group starts a worker"); + tokio::time::timeout(Duration::from_secs(5), run_dst_dir_fsync_group_worker(old_group.clone())) + .await + .expect("old worker must finish its actual fsync"); + assert!(old_rx.await.expect("old worker reports completion").is_ok()); + assert!(fsync_dir_recorder::was_fsynced(temp_dir.path())); + assert_eq!(fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()), vec![1]); + assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0)); + + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let probe = Arc::downgrade(&owner); + let generation = ctx.namespace_commit_generation(); + let (rx, successor_worker) = DST_DIR_FSYNC_GROUP_COMMIT + .enqueue_opened( + OpenedDstDirFsyncGroup::open(temp_dir.path()).expect("open successor directory"), + Some(owner), + ) + .expect("enqueue successor owner"); + let successor = successor_worker.expect("successor starts a new group"); + assert!(!Arc::ptr_eq(&old_group, &successor)); + let worker = run_dst_dir_fsync_group_worker(successor.clone()); + // The stale Arc represents W1 resuming after another worker removed G. + DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&old_group); + assert!(ctx.namespace_commits_pending()); + assert!(probe.upgrade().is_some()); + drop(worker); + let channel_closed = tokio::time::timeout(Duration::from_secs(5), rx) + .await + .expect("dropping the unpolled worker must release its channel") + .is_err(); + let counts_after_drop = DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(); + let owner_released = probe.upgrade().is_none(); + let namespace_pending = ctx.namespace_commits_pending(); + let generation_after_drop = ctx.namespace_commit_generation(); + let successor_pending = successor.inner.lock().pending.len(); + let worker_running = successor.inner.lock().worker_running; + // Preserve the observed result before cleanup, so a RED run cannot leak + // its phantom count into unrelated tests in the same process. + clear_dst_dir_fsync_group_commit_for_test(); + assert!(channel_closed); + assert!(owner_released); + assert!(!namespace_pending); + assert!(generation_after_drop > generation); + assert_eq!(successor_pending, 0); + assert!(!worker_running); + assert_eq!( + fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()), + vec![1], + "dropping the successor before its first poll must not dispatch another fsync" + ); + assert_eq!(counts_after_drop, (0, 0), "stale cleanup must not strand a phantom waiter"); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(dst_dir_fsync_group_commit)] async fn dst_dir_fsync_group_commit_cancellation_releases_waiter_state() { diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 5c0b16a1f..4dc17ad34 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -3663,22 +3663,22 @@ async fn rollback_failed_rename( let object = object.to_string(); let disk_namespace_commit_guard = namespace_commit_guard.clone(); let task = tokio::spawn(async move { - let _namespace_commit_guard = disk_namespace_commit_guard; + let _namespace_commit_guard = disk_namespace_commit_guard.clone(); #[allow(clippy::let_unit_value)] let _task_guard = SetDisks::rename_fanout_task_guard(&object); SetDisks::rename_fanout_barrier(&object, disk_index, rename_fanout_barrier_phase::ROLLBACK).await; #[cfg(test)] rollback_fault_injection::before_undo(&object, disk_index)?; - disk.delete_version( + disk.undo_write_with_namespace_owner( &bucket, &object, fi, - false, DeleteOptions { undo_write: true, old_data_dir: rollback_dir, ..Default::default() }, + disk_namespace_commit_guard.map(|owner| owner as Arc), ) .await }); @@ -4238,7 +4238,7 @@ impl SetDisks { let successful_rename_completion_rank = successful_rename_completion_rank.clone(); let namespace_commit_guard = namespace_commit_guard.clone(); tasks.spawn(async move { - let _namespace_commit_guard = namespace_commit_guard; + let _namespace_commit_guard = namespace_commit_guard.clone(); let mut dispatch_state = RenameDispatchState::NotDispatched; let result = std::panic::AssertUnwindSafe(async { #[allow(clippy::let_unit_value)] @@ -4273,7 +4273,13 @@ impl SetDisks { &file_info, &dst_bucket, &dst_object, - scanner_publication_lease_token, + crate::disk::RenameDataGuards { + scanner_publication_lease_token, + namespace_owner: namespace_commit_guard + .clone() + .map(|owner| owner as Arc), + ..Default::default() + }, ) .await; let rejected_before_publication = observed.rejected_before_publication(); @@ -4602,7 +4608,7 @@ impl SetDisks { // Keep the storage-owned movement permit attached to the actual // fan-out owner, even if the caller future is cancelled. let _fanout_publication_scope = fanout_publication_scope; - let _namespace_commit_guard = fanout_namespace_commit_guard; + let _namespace_commit_guard = fanout_namespace_commit_guard.clone(); let successful_rename_completion_rank = rustfs_io_metrics::put_stage_metrics_enabled().then(|| Arc::new(AtomicUsize::new(0))); let futures = fanout_disks @@ -4617,6 +4623,7 @@ impl SetDisks { let dst_bucket = fanout_dst_bucket.clone(); let successful_rename_completion_rank = successful_rename_completion_rank.clone(); let publication_scope = scanner_publication_commit_scope.clone(); + let namespace_commit_guard = fanout_namespace_commit_guard.clone(); async move { let mut dispatch_state = RenameDispatchState::NotDispatched; @@ -4669,7 +4676,13 @@ impl SetDisks { file_info, &dst_bucket, &dst_object, - scanner_publication_lease_token, + crate::disk::RenameDataGuards { + scanner_publication_lease_token, + namespace_owner: namespace_commit_guard + .clone() + .map(|owner| owner as Arc), + ..Default::default() + }, ) .await; let rejected_before_publication = observed.rejected_before_publication(); @@ -10860,6 +10873,358 @@ mod tests { .await; } + #[cfg(not(windows))] + async fn assert_namespace_owner_survives_physical_publication_timeout(allow_early_ack: bool) { + use crate::disk::os; + use futures::FutureExt; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60")), + (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")), + ], + async { + const DISKS: usize = 4; + let bucket = "namespace-physical-tail"; + let object = "inline-overwrite"; + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + prepare_rename_source_dirs(&dirs, &disks, "source").await; + let mut old = metadata_test_fileinfo(object); + old.mod_time = Some(OffsetDateTime::now_utc()); + old.size = 15; + old.parts.clear(); + old.add_object_part(1, "old-etag".to_string(), 15, None, 15, None, None); + old.data = Some(Bytes::from_static(b"old-inline-body")); + old.set_inline_data(); + old.metadata.insert("etag".to_string(), "old-etag".to_string()); + let mut infos = rename_commit_fileinfos(object, DISKS, "new-etag"); + let mut hooks = Vec::new(); + let mut entered = Vec::new(); + let mut releases = Vec::new(); + let mut publication_paths = Vec::new(); + for (disk, info) in disks.iter().flatten().zip(&mut infos) { + disk.write_metadata(bucket, bucket, object, old.clone()) + .await + .expect("the old inline version must be readable before overwrite"); + info.size = 11; + info.parts.clear(); + info.add_object_part(1, "new-etag".to_string(), 11, None, 11, None, None); + let crate::disk::Disk::Local(local) = disk.as_ref() else { + panic!("physical publication fixture requires local disks"); + }; + // Linux IO paths use a mount FD, which is also the namespace lock key. + let destination = local + .get_disk() + .get_object_path_for_io(bucket, object) + .expect("the publication path must resolve through the disk's mount lease"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + hooks.push(os::prepared_publication_test_hooks::install( + &destination.join(STORAGE_FORMAT_FILE), + move || { + let _ = entered_tx.send(()); + // Sender drop also releases the syscall when an earlier assertion fails. + let _ = release_rx.recv(); + }, + )); + entered.push(entered_rx); + releases.push(release_tx); + publication_paths.push(destination); + } + let namespace_owner = ctx.begin_namespace_commit(); + let namespace_probe = Arc::downgrade(&namespace_owner); + let receipt = RenameRollbackReceipt::default(); + let mut rename = Box::pin(SetDisks::rename_data_owned_with_fence( + &disks, + (RUSTFS_META_TMP_BUCKET, "source"), + infos, + (bucket, object), + allow_early_ack, + RenameDataFenceOptions::new(3, None) + .with_rollback_receipt(receipt.clone()) + .with_namespace_commit_guard(Some(namespace_owner)), + )); + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + signals = join_all(entered) => { + assert!(signals.into_iter().all(|signal| signal.is_ok()), "all physical publishers must enter"); + } + _ = rename.as_mut() => panic!("rename must not finish before physical publication is paused"), + } + }) + .await + .expect("all four prepared metadata renames must reach their blocking syscall"); + assert!(ctx.namespace_commits_pending()); + assert_eq!(ctx.namespace_commit_generation(), 1); + + // Every wrapper timer exists before advancing; the physical closures stay blocked. + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let result = tokio::time::timeout(Duration::from_secs(5), rename) + .await + .expect("ordinary disk timeout must not wait for the physical rename"); + assert!(result.is_err(), "four timed-out disks cannot satisfy write quorum"); + let report = receipt.0.get().expect("failed fanout must finish rollback accounting"); + assert_eq!(report.disks.len(), DISKS); + assert!( + report + .disks + .iter() + .all(|disk| matches!(disk.outcome, RenameRollbackOutcome::Indeterminate(DiskError::Timeout))) + ); + let pending_before_release = ctx.namespace_commits_pending(); + let owner_alive_before_release = namespace_probe.upgrade().is_some(); + let old_snapshot_generation = ctx.namespace_commit_generation(); + for (disk, destination) in disks.iter().flatten().zip(&publication_paths) { + let root = disk.path(); + assert!( + os::acquire_rename_data_mutation_lease(&root, bucket, destination) + .now_or_never() + .is_none(), + "the physical publication must still own object serialization after the async timeout" + ); + assert!( + root.join(RUSTFS_META_TMP_BUCKET) + .join("source") + .join(STORAGE_FORMAT_FILE) + .exists() + ); + let stored = disk + .read_version( + "", + bucket, + object, + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("a scanner can still read the complete old metadata while publication is paused"); + assert_eq!(stored.size, 15); + assert_eq!(stored.data.as_deref(), Some(b"old-inline-body".as_slice())); + } + assert_eq!(ctx.namespace_commit_generation(), old_snapshot_generation); + + // Drain real syscalls before checking the regression, including on the RED run. + drop(releases); + for (disk, destination) in disks.iter().flatten().zip(&publication_paths) { + let root = disk.path(); + let lease = tokio::time::timeout( + Duration::from_secs(5), + os::acquire_rename_data_mutation_lease(&root, bucket, destination), + ) + .await + .expect("released physical publishers must drain"); + drop(lease); + } + for dir in &dirs { + let reopened = reopen_local_disk(dir).await; + let stored = reopened + .read_version( + "", + bucket, + object, + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("the detached prepared rename must actually publish after timeout"); + assert_eq!(stored.size, 11); + assert_eq!(stored.data.as_deref(), Some(b"inline-body".as_slice())); + } + // The lease releases its locks before dropping the namespace owner, and the + // owner's `Drop` runs after its `Weak` probe stops upgrading, so wait for the + // pending counter itself instead of asserting it right after the drain. + tokio::time::timeout(Duration::from_secs(5), async { + while ctx.namespace_commits_pending() || namespace_probe.upgrade().is_some() { + tokio::task::yield_now().await; + } + }) + .await + .expect("released physical publishers must release namespace ownership"); + let generation_after_publication = ctx.namespace_commit_generation(); + assert!(!ctx.namespace_commits_pending()); + assert!(namespace_probe.upgrade().is_none()); + assert!(receipt.is_incomplete(), "late publication must not erase failed-write recovery evidence"); + assert!( + pending_before_release && owner_alive_before_release, + "physical publication outlived namespace accounting: early_ack={allow_early_ack}, \ + pending={pending_before_release}, owner_alive={owner_alive_before_release}, \ + old_snapshot_generation={old_snapshot_generation}, after_late_publication={generation_after_publication}" + ); + assert!( + generation_after_publication > old_snapshot_generation, + "physical completion must invalidate the scanner's old metadata snapshot" + ); + }, + ) + .await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn rename_full_wait_timeout_keeps_namespace_owner_until_physical_publication() { + assert_namespace_owner_survives_physical_publication_timeout(false).await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn rename_early_ack_timeout_keeps_namespace_owner_until_physical_publication() { + assert_namespace_owner_survives_physical_publication_timeout(true).await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn successful_rename_ack_keeps_physical_tail_owner_after_caller_cancellation() { + use crate::disk::os; + temp_env::async_with_vars( + [ + (rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60")), + (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")), + ], + async { + let bucket = "physical-ack-tail"; + let object = "ack-object"; + let (dirs, disks) = call_counter_local_disks(bucket, 4).await; + prepare_rename_source_dirs(&dirs, &disks, "source").await; + let mut infos = rename_commit_fileinfos(object, 4, "new-etag"); + for info in &mut infos { + info.size = 11; + info.parts.clear(); + info.add_object_part(1, "new-etag".to_string(), 11, None, 11, None, None); + } + let disk = disks[3].as_ref().expect("tail disk"); + let crate::disk::Disk::Local(local) = disk.as_ref() else { + panic!("local fixture"); + }; + let destination = local.get_disk().get_object_path_for_io(bucket, object).expect("tail IO path"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = os::prepared_publication_test_hooks::install(&destination.join(STORAGE_FORMAT_FILE), move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let owner = ctx.begin_namespace_commit(); + let owner_probe = Arc::downgrade(&owner); + let receipt = RenameRollbackReceipt::default(); + let caller_receipt = receipt.clone(); + let caller_disks = disks.clone(); + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + let caller = tokio::spawn(async move { + let commit = SetDisks::rename_data_owned_with_fence( + &caller_disks, + (RUSTFS_META_TMP_BUCKET, "source"), + infos, + (bucket, object), + true, + RenameDataFenceOptions::new(3, None) + .with_namespace_commit_guard(Some(owner)) + .with_rollback_receipt(caller_receipt), + ) + .await + .expect("three real disk publications must produce a successful ACK"); + assert!(ack_tx.send(commit).is_ok(), "deliver successful ACK"); + std::future::pending::<()>().await; + }); + let mut commit = tokio::time::timeout(Duration::from_secs(10), async { + entered_rx.await.expect("physical tail entry"); + ack_rx + .await + .expect("ACK must arrive while the fourth disk is physically paused") + }) + .await + .expect("successful quorum ACK must not wait for its physical tail"); + assert_eq!(commit.online_disks.iter().flatten().count(), 3); + assert!(!destination.join(STORAGE_FORMAT_FILE).exists(), "tail has not published at ACK"); + let tail_drain = commit.tail_drain.take().expect("early ACK transfers a real tail handle"); + drop(commit); + caller.abort(); + assert!(caller.await.expect_err("cancel caller after it delivered ACK").is_cancelled()); + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let tail = tokio::time::timeout(Duration::from_secs(5), tail_drain) + .await + .expect("ordinary tail timeout stays bounded after ACK") + .expect("tail owner must not panic") + .expect("successful ACK keeps its convergence result"); + assert_eq!(tail.convergence, RenameConvergence::PartialCommit); + assert!(receipt.0.get().is_none(), "an acknowledged write must never enter rollback"); + let pending = ctx.namespace_commits_pending(); + let alive = owner_probe.upgrade().is_some(); + let generation = ctx.namespace_commit_generation(); + for disk in disks.iter().flatten().take(3) { + let stored = disk + .read_version( + "", + bucket, + object, + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("all ACK voters keep the new object after caller cancellation"); + assert_eq!(stored.data.as_deref(), Some(b"inline-body".as_slice())); + } + drop(release_tx); + let lease = tokio::time::timeout( + Duration::from_secs(5), + os::acquire_rename_data_mutation_lease(&disk.path(), bucket, &destination), + ) + .await + .expect("late physical tail drains"); + drop(lease); + tokio::time::timeout(Duration::from_secs(5), async { + while ctx.namespace_commits_pending() || owner_probe.upgrade().is_some() { + tokio::task::yield_now().await; + } + }) + .await + .expect("late physical tail must release namespace ownership"); + for dir in &dirs { + let stored = reopen_local_disk(dir) + .await + .read_version( + "", + bucket, + object, + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("successful ACK remains committed on every disk after late publication"); + assert_eq!(stored.data.as_deref(), Some(b"inline-body".as_slice())); + } + assert!( + pending && alive, + "physical ACK tail must retain namespace ownership after the coordinator exits" + ); + assert!(!ctx.namespace_commits_pending()); + assert!(owner_probe.upgrade().is_none()); + assert!(ctx.namespace_commit_generation() > generation); + assert!(receipt.0.get().is_none(), "late publication cannot change success into rollback"); + }, + ) + .await; + } + #[tokio::test] #[serial_test::serial(capacity_dirty_scope)] async fn rename_rollback_incomplete_receipt_waits_for_undo_barrier() {