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() {