From d646f85778ca371461591701e1626edd6201f2bc Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Mon, 28 Sep 2026 00:55:32 +0800 Subject: [PATCH] perf(ecstore): reuse pinned directory handles for object I/O --- crates/ecstore/src/disk/fs.rs | 28 - crates/ecstore/src/disk/local.rs | 190 ++--- crates/ecstore/src/disk/local/commit.rs | 82 +- crates/ecstore/src/disk/os.rs | 822 ++++++++++++++++++-- crates/ecstore/src/set_disk/ops/object.rs | 28 +- crates/ecstore/tests/fd_relative_io_test.rs | 243 ++++++ crates/ecstore/tests/storage_api.rs | 6 + 7 files changed, 1172 insertions(+), 227 deletions(-) create mode 100644 crates/ecstore/tests/fd_relative_io_test.rs diff --git a/crates/ecstore/src/disk/fs.rs b/crates/ecstore/src/disk/fs.rs index 7dbbaf14d..15a52ef7b 100644 --- a/crates/ecstore/src/disk/fs.rs +++ b/crates/ecstore/src/disk/fs.rs @@ -223,12 +223,6 @@ pub fn rename_std(from: impl AsRef, to: impl AsRef) -> io::Result<() std::fs::rename(from, to) } -#[cfg(any(not(windows), test))] -#[tracing::instrument(level = "debug", skip_all)] -pub async fn read_file(path: impl AsRef) -> io::Result> { - fs::read(path.as_ref()).await -} - // Bucket existence cache - reduces statx syscalls for repeated bucket checks /// Cache for bucket directory existence checks. @@ -614,28 +608,6 @@ mod tests { assert_eq!(content, "test content"); } - #[tokio::test] - async fn test_read_file() { - let temp_dir = TempDir::new().unwrap(); - let file_path = temp_dir.path().join("test_read.txt"); - - let test_content = b"This is test content for reading"; - tokio::fs::write(&file_path, test_content).await.unwrap(); - - // Read file - let read_content = read_file(&file_path).await.unwrap(); - assert_eq!(read_content, test_content); - } - - #[tokio::test] - async fn test_read_file_nonexistent() { - let temp_dir = TempDir::new().unwrap(); - let file_path = temp_dir.path().join("nonexistent.txt"); - - // Should fail for non-existent file - assert!(read_file(&file_path).await.is_err()); - } - #[tokio::test] async fn test_same_file() { let temp_dir = TempDir::new().unwrap(); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 89e03caaa..23b54adfe 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -189,8 +189,10 @@ struct ListingMetadataRead { has_namespace_child_candidate: bool, } -fn read_all_data_std(path: &Path) -> core::result::Result<(Vec, Option), ReadAllError> { - let mut file = std::fs::File::open(path).map_err(ReadAllError::Open)?; +fn read_all_data_std( + opened: std::io::Result, +) -> core::result::Result<(Vec, Option), ReadAllError> { + let mut file = opened.map_err(ReadAllError::Open)?; let metadata = file.metadata().map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?; if metadata.is_dir() { @@ -5309,24 +5311,6 @@ fn mount_id_from_mountinfo_contents(mountinfo: &str, path: &Path) -> Option } impl LocalDisk { - #[cfg(target_os = "linux")] - fn open_mount_lease(root: &Path) -> Result<(std::fs::File, PathBuf, Option)> { - use rustix::fs::{Mode, OFlags, open}; - use std::os::fd::AsRawFd as _; - - let fd = open( - root, - OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, - Mode::empty(), - ) - .map_err(std::io::Error::from) - .map_err(DiskError::from)?; - let lease = std::fs::File::from(fd); - let io_root = PathBuf::from(format!("/proc/self/fd/{}/.", lease.as_raw_fd())); - let mount_id = mount_id_for_fd(&lease); - Ok((lease, io_root, mount_id)) - } - #[cfg(not(target_os = "linux"))] fn open_mount_lease(root: &Path) -> Result { Ok(root.to_path_buf()) @@ -5410,7 +5394,11 @@ impl LocalDisk { let root = publication_root.path().to_path_buf(); #[cfg(target_os = "linux")] - let (mount_lease, io_root, mount_lease_mount_id) = Self::open_mount_lease(&root)?; + let (mount_lease, io_root, mount_lease_mount_id) = ( + publication_root.directory().clone(), + publication_root.io_path().to_path_buf(), + mount_id_for_fd(publication_root.directory()), + ); #[cfg(not(target_os = "linux"))] let io_root = Self::open_mount_lease(&root)?; @@ -5458,9 +5446,11 @@ impl LocalDisk { state = "format_path_resolved", "Local disk format path resolved" ); - let (format_data, format_meta) = read_file_exists(&io_format_path).await.inspect_err(|err| { - log_startup_disk_error("read_format_json", &io_format_path, err); - })?; + let (format_data, format_meta) = read_file_exists(&io_format_path, &publication_root) + .await + .inspect_err(|err| { + log_startup_disk_error("read_format_json", &io_format_path, err); + })?; let mut id = None; // let mut format_legacy = false; @@ -5551,7 +5541,7 @@ impl LocalDisk { publication_root, io_root: io_root.clone(), #[cfg(target_os = "linux")] - mount_lease: Arc::new(mount_lease), + mount_lease, #[cfg(target_os = "linux")] mount_lease_mount_id, endpoint: ep.clone(), @@ -6008,6 +5998,19 @@ impl LocalDisk { self.local_disk_object_path(self.io_root(), bucket, key) } + /// Only for operations that open through PublicationRoot or a commit guard: + /// containment and symlink checks happen with the actual open, off the runtime. + fn io_get_object_open_path(&self, bucket: &str, key: &str) -> Result { + #[cfg(target_os = "linux")] + { + let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key); + check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?; + Ok(path) + } + #[cfg(not(target_os = "linux"))] + self.io_get_object_path(bucket, key) + } + fn io_get_bucket_path(&self, bucket: &str) -> Result { self.local_disk_bucket_path(self.io_root(), bucket) } @@ -6166,7 +6169,7 @@ impl LocalDisk { let mut meta = FileMeta::new(); if !fi.fresh { - let (buf, _) = read_file_exists(&p).await?; + let (buf, _) = read_file_exists(&p, &self.publication_root).await?; if !buf.is_empty() { let _ = meta.unmarshal_msg(&buf).map_err(|_| { meta = FileMeta::new(); @@ -7017,9 +7020,10 @@ impl LocalDisk { // - metadata failure -> to_file_error // - parse failure -> propagated verbatim from read_xl_meta_no_data_sync (`?`) let path = file_path.as_ref().to_path_buf(); + let root = self.publication_root.clone(); let (data, modtime) = tokio::task::spawn_blocking(move || -> Result<(Vec, Option)> { // Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)). - let mut f = std::fs::File::open(&path).map_err(to_file_error)?; + let mut f = root.open_readonly(&path).map_err(to_file_error)?; let meta = f.metadata().map_err(to_file_error)?; @@ -7072,6 +7076,7 @@ impl LocalDisk { let metadata_path = self.io_get_object_path(bucket, path.to_string_lossy().as_ref()); // A part's existence check, metadata read and decode share one dispatch. // Keep open errors unmapped for the existing missing-volume fallback. + let root = self.publication_root.clone(); let result = tokio::task::spawn_blocking(move || -> Result<_> { let part_error = |error: String| ObjectPartInfo { number: num, @@ -7084,7 +7089,7 @@ impl LocalDisk { // Invalid metadata paths remain request errors, but missing data wins // first, as it does in the serial reader. let metadata_path = metadata_path?; - Ok(read_all_data_std(&metadata_path) + Ok(read_all_data_std(root.open_readonly(&metadata_path)) .map(|(data, _)| ObjectPartInfo::unmarshal(&data).unwrap_or_else(|err| part_error(err.to_string())))) }) .await; @@ -7103,8 +7108,9 @@ impl LocalDisk { let object_dir = self.io_get_object_path(volume, object_name)?; let metadata_path = object_dir.join(STORAGE_FORMAT_FILE); let volume_dir = self.io_get_bucket_path(volume)?; + let root = self.publication_root.clone(); let result = tokio::task::spawn_blocking(move || { - let (bytes, _) = read_all_data_std(&metadata_path)?; + let (bytes, _) = read_all_data_std(root.open_readonly(&metadata_path))?; let file_meta = FileMeta::load(&bytes).ok(); let data_dirs: HashSet = file_meta .as_ref() @@ -7194,7 +7200,8 @@ impl LocalDisk { // gating the volume fallback on the open error alone is equivalent to // the original code, where the fallback lived solely in the open match arm. let path = file_path.as_ref().to_path_buf(); - let res = tokio::task::spawn_blocking(move || read_all_data_std(&path)) + let root = self.publication_root.clone(); + let res = tokio::task::spawn_blocking(move || read_all_data_std(root.open_readonly(&path))) .await .map_err(DiskError::from)?; @@ -7926,16 +7933,16 @@ impl LocalDisk { os::make_dir_all(parent, skip_parent).await?; } + let root = self.publication_root.clone(); tokio::task::spawn_blocking(move || { #[cfg(test)] run_owned_file_write_before_open(&path); - let mut file = std::fs::OpenOptions::new() - .create(true) - .write(true) - .truncate(true) - .open(&path) - .map_err(to_file_error)?; + let (mut file, parent_handle) = if sync == SyncMode::FileAndDir { + root.create_truncate_with_parent(&path).map_err(to_file_error)? + } else { + (root.create_truncate(&path).map_err(to_file_error)?, None) + }; std::io::Write::write_all(&mut file, buf.as_ref()).map_err(to_file_error)?; if sync != SyncMode::None { file.sync_data().map_err(to_file_error)?; @@ -7944,7 +7951,7 @@ impl LocalDisk { if sync == SyncMode::FileAndDir && let Some(parent) = path.parent() { - os::fsync_dir_std(parent).map_err(to_file_error)?; + os::fsync_directory_handle_std(parent, parent_handle.as_ref()).map_err(to_file_error)?; } } Ok::<_, std::io::Error>(()) @@ -7969,7 +7976,18 @@ impl LocalDisk { os::make_dir_all(parent, skip_parent).await?; } - let f = super::fs::open_file(path.as_ref(), mode).await.map_err(to_file_error)?; + let f = if mode == O_CREATE | O_WRONLY | O_TRUNC { + let path = path.as_ref().to_path_buf(); + let root = self.publication_root.clone(); + File::from_std( + tokio::task::spawn_blocking(move || root.create_truncate(&path)) + .await + .map_err(DiskError::from)? + .map_err(to_file_error)?, + ) + } else { + super::fs::open_file(path.as_ref(), mode).await.map_err(to_file_error)? + }; Ok(f) } @@ -8835,9 +8853,9 @@ fn file_meta_counts_toward_limit(meta: &FileMeta) -> bool { } // Filter std::io::ErrorKind::NotFound -async fn read_file_exists(path: impl AsRef) -> Result<(Bytes, Option)> { +async fn read_file_exists(path: impl AsRef, root: &os::PublicationRoot) -> Result<(Bytes, Option)> { let p = path.as_ref(); - let (data, meta) = match read_file_all(&p).await { + let (data, meta) = match read_file_all(&p, root).await { Ok((data, meta)) => (data, Some(meta)), Err(e) => { if e == Error::FileNotFound { @@ -8856,32 +8874,28 @@ async fn read_file_exists(path: impl AsRef) -> Result<(Bytes, Option) -> Result<(Bytes, Metadata)> { - let p = path.as_ref(); - let meta = read_file_metadata(&path).await?; - - let data = fs::read(&p) - .await - .inspect_err(|err| { - log_startup_disk_io_error("read_file_all", p, err); - }) - .map_err(to_file_error)?; - - Ok((data.into(), meta)) -} - -async fn read_file_metadata(p: impl AsRef) -> Result { - let path = p.as_ref(); - let meta = fs::metadata(path) - .await - .inspect_err(|err| { - if err.kind() != ErrorKind::NotFound { - log_startup_disk_io_error("read_file_metadata", path, err); - } - }) - .map_err(to_file_error)?; - - Ok(meta) +async fn read_file_all(path: impl AsRef, root: &os::PublicationRoot) -> Result<(Bytes, Metadata)> { + let path = path.as_ref().to_path_buf(); + let root = root.clone(); + tokio::task::spawn_blocking(move || { + let result = (|| { + let mut file = root.open_readonly(&path)?; + let metadata = file.metadata()?; + let mut data = Vec::new(); + std::io::Read::read_to_end(&mut file, &mut data)?; + Ok::<_, std::io::Error>((Bytes::from(data), metadata)) + })(); + result + .inspect_err(|err| { + if err.kind() != ErrorKind::NotFound { + log_startup_disk_io_error("read_file_all", &path, err); + } + }) + .map_err(to_file_error) + .map_err(DiskError::from) + }) + .await + .map_err(DiskError::from)? } fn skip_access_checks(p: impl AsRef) -> bool { @@ -8963,13 +8977,19 @@ fn check_local_disk_valid_path(root: &Path, path: impl AsRef) -> Result<() #[cfg(target_os = "linux")] fn check_local_disk_valid_object_path_at(root: &Path, root_fd: &std::fs::File, bucket_path: &Path, path: &Path) -> Result<()> { + check_local_disk_object_path_components(root, bucket_path, path)?; + reject_local_disk_symlink_components_at(root, root_fd, &normalize_path_components(path)) +} + +#[cfg(target_os = "linux")] +fn check_local_disk_object_path_components(root: &Path, bucket_path: &Path, path: &Path) -> Result<()> { let bucket_path = normalize_path_components(bucket_path); let path = normalize_path_components(path); if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) { return Err(DiskError::InvalidPath); } - reject_local_disk_symlink_components_at(root, root_fd, &path) + Ok(()) } #[cfg(target_os = "linux")] @@ -9519,9 +9539,9 @@ impl DiskAPI for LocalDisk { } } - let p = self.io_get_object_path(volume, path)?; + let p = self.io_get_object_open_path(volume, path)?; - let (data, _) = read_file_all(&p).await?; + let (data, _) = read_file_all(&p, &self.publication_root).await?; Ok(data) } @@ -11120,7 +11140,7 @@ impl DiskAPI for LocalDisk { }; // if req.metadata_only {} - match read_file_all(&fpath).await { + match read_file_all(&fpath, &self.publication_root).await { Ok((data, meta)) => { found += 1; @@ -21382,10 +21402,12 @@ mod test { #[tokio::test] async fn test_read_file_exists() { - let test_file = "./test_read_exists.txt"; + let directory = tempfile::tempdir().expect("test directory"); + let root = os::PublicationRoot::new(directory.path()).expect("publication root"); + let test_file = &directory.path().join("read_exists"); // Test non-existent file - let (data, metadata) = read_file_exists(test_file).await.expect("operation should succeed"); + let (data, metadata) = read_file_exists(test_file, &root).await.expect("operation should succeed"); assert!(data.is_empty()); assert!(metadata.is_none()); @@ -21393,7 +21415,7 @@ mod test { fs::write(test_file, b"test content").await.expect("operation should succeed"); // Test existing file - let (data, metadata) = read_file_exists(test_file).await.expect("operation should succeed"); + let (data, metadata) = read_file_exists(test_file, &root).await.expect("operation should succeed"); assert_eq!(data.as_ref(), b"test content"); assert!(metadata.is_some()); @@ -21403,14 +21425,16 @@ mod test { #[tokio::test] async fn test_read_file_all() { - let test_file = "./test_read_all.txt"; + let directory = tempfile::tempdir().expect("test directory"); + let root = os::PublicationRoot::new(directory.path()).expect("publication root"); + let test_file = &directory.path().join("read_all"); let test_content = b"test content for read_all"; // Create test file fs::write(test_file, test_content).await.expect("operation should succeed"); // Test reading file - let (data, metadata) = read_file_all(test_file).await.expect("operation should succeed"); + let (data, metadata) = read_file_all(test_file, &root).await.expect("operation should succeed"); assert_eq!(data.as_ref(), test_content); assert!(metadata.is_file()); assert_eq!(metadata.len(), test_content.len() as u64); @@ -21419,22 +21443,6 @@ mod test { let _ = fs::remove_file(test_file).await; } - #[tokio::test] - async fn test_read_file_metadata() { - let test_file = "./test_metadata.txt"; - - // Create test file - fs::write(test_file, b"test").await.expect("operation should succeed"); - - // Test reading metadata - let metadata = read_file_metadata(test_file).await.expect("operation should succeed"); - assert!(metadata.is_file()); - assert_eq!(metadata.len(), 4); // "test" is 4 bytes - - // Clean up - let _ = fs::remove_file(test_file).await; - } - #[test] fn test_is_root_path() { // Unix root path diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs index 6ca3e14e8..03e2051d1 100644 --- a/crates/ecstore/src/disk/local/commit.rs +++ b/crates/ecstore/src/disk/local/commit.rs @@ -173,6 +173,7 @@ pub(super) fn create_local_inline_rollback_backup( Ok(backup_path) } +#[cfg(test)] pub(super) async fn lock_rename_commit_directories( source_parent: &Path, destination_parent: &Path, @@ -180,7 +181,7 @@ pub(super) async fn lock_rename_commit_directories( publication_root: &os::PublicationRoot, mutation_lease: Arc, ) -> Result { - #[cfg(windows)] + #[cfg(any(windows, target_os = "linux"))] let result = { let source_parent = source_parent.to_path_buf(); let destination_parent = destination_parent.to_path_buf(); @@ -188,7 +189,7 @@ pub(super) async fn lock_rename_commit_directories( let publication_root = publication_root.clone(); os::run_blocking_namespace_operation(mutation_lease, move || { let result = os::prepare_rename_commit_guard(&source_parent, &destination_parent, &base_dir, &publication_root); - #[cfg(test)] + #[cfg(all(test, windows))] if result.is_ok() { run_destination_commit_directory_preparation(&destination_parent); } @@ -196,7 +197,7 @@ pub(super) async fn lock_rename_commit_directories( }) .await }; - #[cfg(not(windows))] + #[cfg(not(any(windows, target_os = "linux")))] let result = { let _ = mutation_lease; os::prepare_rename_commit_guard(source_parent, destination_parent, base_dir, publication_root) @@ -210,33 +211,42 @@ pub(super) async fn lock_rename_commit_directories( result.map_err(to_file_error).map_err(DiskError::from) } -async fn read_rename_destination_metadata( +async fn prepare_rename_destination_metadata( + source_parent: &Path, file_path: &Path, - rename_commit_guard: &os::RenameCommitGuard, + base_dir: &Path, + publication_root: &os::PublicationRoot, mutation_lease: Arc, -) -> Result> { - #[cfg(windows)] - let result = { - let file_path = file_path.to_path_buf(); - let rename_commit_guard = rename_commit_guard.clone(); - os::run_blocking_namespace_operation(mutation_lease, move || { - os::read_destination_file_with_commit_guard(&file_path, &rename_commit_guard) - }) - .await - }; - #[cfg(not(windows))] - let _ = (rename_commit_guard, mutation_lease); - #[cfg(not(windows))] - let result = match super::super::fs::read_file(file_path).await { - Ok(data) => Ok(Some(data)), - Err(err) if err.kind() == ErrorKind::NotFound => Ok(None), - Err(err) => Err(err), - }; - - result - .map(|data| data.map(Bytes::from)) - .map_err(to_file_error) - .map_err(DiskError::from) +) -> Result<(os::RenameCommitGuard, Option)> { + let source_parent = source_parent.to_path_buf(); + let file_path = file_path.to_path_buf(); + let base_dir = base_dir.to_path_buf(); + let publication_root = publication_root.clone(); + os::run_blocking_namespace_operation(mutation_lease, move || { + let parent = file_path + .parent() + .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "missing metadata parent"))?; + let guard = os::prepare_rename_commit_guard(&source_parent, parent, &base_dir, &publication_root).map_err(|err| { + match std::fs::symlink_metadata(&base_dir) { + Err(base_err) if base_err.kind() == ErrorKind::NotFound => base_err, + _ => err, + } + })?; + #[cfg(all(test, windows))] + run_destination_commit_directory_preparation(parent); + #[cfg(any(windows, target_os = "linux"))] + let data = os::read_destination_file_with_commit_guard(&file_path, &guard)?; + #[cfg(not(any(windows, target_os = "linux")))] + let data = match std::fs::read(&file_path) { + Ok(data) => Some(data), + Err(err) if err.kind() == ErrorKind::NotFound => None, + Err(err) => return Err(err), + }; + Ok((guard, data.map(Bytes::from))) + }) + .await + .map_err(to_file_error) + .map_err(DiskError::from) } async fn restore_renamed_data_source( @@ -419,7 +429,7 @@ impl LocalDisk { // xl.meta path let src_file_path = self.io_get_object_path(src_volume, format!("{}/{}", src_path, STORAGE_FORMAT_FILE).as_str())?; - let dst_file_path = self.io_get_object_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?; + let dst_file_path = self.io_get_object_open_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?; // data_dir path let has_data_dir_path = { @@ -473,18 +483,16 @@ impl LocalDisk { if !no_inline { fs::create_dir_all(src_file_parent).await.map_err(to_file_error)?; } - // Acquire the common trees before reading destination metadata. On - // Windows this pins the object directory identity across metadata - // preparation, data publication, rollback backup, and final commit. - let rename_commit_guard = lock_rename_commit_directories( + // Open the commit directories and read the old metadata in one blocking + // operation. The returned guard owns the same identities through publish. + let (rename_commit_guard, has_dst_buf) = prepare_rename_destination_metadata( src_file_parent, - dst_file_parent, + &dst_file_path, &dst_volume_dir, &self.publication_root, mutation_lease.clone(), ) .await?; - let has_dst_buf = read_rename_destination_metadata(&dst_file_path, &rename_commit_guard, mutation_lease.clone()).await?; if no_inline { // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta @@ -858,7 +866,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, Some(mutation_lease.clone())).await { + if let Err(err) = os::fsync_commit_directory(parent, &rename_commit_guard, mutation_lease.clone(), None).await { rustfs_io_metrics::record_put_object_stage_duration_from( rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, fsync_started, @@ -1222,7 +1230,7 @@ impl LocalDisk { { let fsync_started = rustfs_io_metrics::put_stage_timer(); if let Err(err) = - os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission) + os::fsync_commit_directory(dst_parent, &rename_commit_guard, mutation_lease.clone(), Some(admission)) .await { rustfs_io_metrics::record_put_object_stage_duration_from( diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index e8127b6a2..80ab24e77 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -486,6 +486,10 @@ pub(crate) mod prepare_rename_test_hooks { /// Fsync a directory so recently created or renamed entries survive power loss. /// No-op on non-Unix platforms where directories cannot be opened for syncing. pub fn fsync_dir_std(dir: impl AsRef) -> io::Result<()> { + fsync_directory_handle_std(dir.as_ref(), None) +} + +pub(crate) fn fsync_directory_handle_std(dir: &Path, opened: Option<&std::fs::File>) -> io::Result<()> { #[cfg(test)] fsync_dir_recorder::record(dir.as_ref()); #[cfg(unix)] @@ -494,10 +498,13 @@ pub fn fsync_dir_std(dir: impl AsRef) -> io::Result<()> { if let Some(kind) = fsync_dir_recorder::take_failure(dir.as_ref()) { return Err(io::Error::from(kind)); } - std::fs::File::open(dir.as_ref())?.sync_all()?; + match opened { + Some(file) => file.sync_all()?, + None => std::fs::File::open(dir)?.sync_all()?, + } } #[cfg(not(unix))] - let _ = dir; + let _ = (dir, opened); Ok(()) } @@ -735,6 +742,17 @@ impl OpenedDstDirFsyncGroup { Ok(Self { key, dir }) } } + + #[cfg(target_os = "linux")] + fn from_file(dir: &Path, file: Arc) -> io::Result { + let key = DstDirFsyncGroupKey::from_metadata(dir.canonicalize()?, file.metadata()?)?; + Ok(Self { + #[cfg(test)] + dir: key.canonical_path.clone(), + key, + dir_file: file, + }) + } } struct DstDirFsyncWaiter { @@ -1008,19 +1026,45 @@ fn run_dst_dir_fsync_group_worker(group: Arc) -> impl std::fut } } +#[cfg(any(test, not(target_os = "linux")))] async fn fsync_dst_dir_group_commit_with_enabled( dir: impl AsRef, enabled: bool, namespace_owner: Option>, +) -> io::Result<()> { + fsync_dst_dir_group_commit_opened(dir.as_ref(), enabled, namespace_owner, None).await +} + +async fn fsync_dst_dir_group_commit_opened( + dir: &Path, + enabled: bool, + namespace_owner: Option>, + opened: Option>, ) -> io::Result<()> { if !enabled { - return fsync_dir_with_owner(dir.as_ref(), namespace_owner).await; + if let Some(file) = opened { + let dir = dir.to_path_buf(); + return fsync_spawn_blocking(move || { + let _owner = namespace_owner; + fsync_directory_handle_std(&dir, Some(&file)) + }) + .await?; + } + return fsync_dir_with_owner(dir, 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 dir = dir.to_path_buf(); + let opened = tokio::task::spawn_blocking(move || { + #[cfg(target_os = "linux")] + if let Some(file) = opened { + return OpenedDstDirFsyncGroup::from_file(&dir, file); + } + #[cfg(not(target_os = "linux"))] + let _ = opened; + 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, namespace_owner)?; if let Some(group) = worker { tokio::spawn(run_dst_dir_fsync_group_worker(group)); @@ -1033,6 +1077,49 @@ async fn fsync_dst_dir_group_commit_with_enabled( } } +/// Reuse the published directory without bypassing group commit or sync admission. +pub(crate) async fn fsync_commit_directory( + dir: &Path, + guard: &RenameCommitGuard, + lease: Arc, + admission: Option<&FileSyncAdmission>, +) -> io::Result<()> { + #[cfg(target_os = "linux")] + { + if dir != guard.directories.destination_path { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "sync directory does not match commit guard")); + } + let file = guard + .directories + .destination + .get() + .ok_or_else(|| io::Error::other("commit directory was not published"))? + .file + .clone(); + if !dst_dir_fsync_group_commit_enabled() + && let Some(admission) = admission + { + let dir = dir.to_path_buf(); + return run_blocking_namespace_file_sync_operation(lease, admission, move || { + #[cfg(test)] + fsync_dir_recorder::record_limited(&dir); + fsync_directory_handle_std(&dir, Some(&file)) + }) + .await; + } + fsync_dst_dir_group_commit_opened(dir, dst_dir_fsync_group_commit_enabled(), Some(lease), Some(file)).await + } + #[cfg(not(target_os = "linux"))] + { + let _ = guard; + match admission { + Some(admission) => fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dir, lease, admission).await, + None => fsync_dst_dir_group_commit(dir, Some(lease)).await, + } + } +} + +#[cfg(not(target_os = "linux"))] pub(crate) async fn fsync_dst_dir_group_commit( dir: impl AsRef, namespace_owner: Option>, @@ -1040,6 +1127,7 @@ pub(crate) async fn fsync_dst_dir_group_commit( fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled(), namespace_owner).await } +#[cfg(not(target_os = "linux"))] pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit( dir: impl AsRef, lease: Arc, @@ -2291,6 +2379,20 @@ pub(crate) fn create_prepared_rename_source_with_commit_guard( #[cfg(not(windows))] { let _ = (dst_file_path, commit_guard); + #[cfg(target_os = "linux")] + let source = { + use rustix::fs::{Mode, OFlags}; + let directories = &commit_guard.directories; + let name = directories.file_name(src_file_path, true)?; + directories.file_name(dst_file_path, false)?; + open_linux_relative( + &directories.source.file, + Path::new(name), + OFlags::CREATE | OFlags::TRUNC | OFlags::WRONLY, + Mode::from_raw_mode(0o666), + )? + }; + #[cfg(not(target_os = "linux"))] let source = std::fs::OpenOptions::new() .create(true) .write(true) @@ -2311,18 +2413,39 @@ pub(crate) fn create_prepared_rename_source_with_commit_guard( } } -#[cfg(windows)] +#[cfg(any(windows, target_os = "linux"))] pub(crate) fn read_destination_file_with_commit_guard( file_path: &Path, commit_guard: &RenameCommitGuard, ) -> io::Result>> { - if file_path.parent() != Some(commit_guard.destination_parent.as_path()) { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "destination file parent does not match its commit guard", - )); + #[cfg(target_os = "linux")] + { + use rustix::fs::{Mode, OFlags}; + let directories = &commit_guard.directories; + let name = directories.file_name(file_path, false)?; + let Some(parent) = directories.destination.get() else { + return Ok(None); + }; + return match open_linux_relative(&parent.file, Path::new(name), OFlags::RDONLY, Mode::empty()) { + Ok(mut file) => { + let mut data = Vec::new(); + std::io::Read::read_to_end(&mut file, &mut data)?; + Ok(Some(data)) + } + Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None), + Err(err) => Err(err), + }; + } + #[cfg(windows)] + { + if file_path.parent() != Some(commit_guard.destination_parent.as_path()) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "destination file parent does not match its commit guard", + )); + } + read_windows_relative_file(file_path, &commit_guard.destination_parent_guard) } - read_windows_relative_file(file_path, &commit_guard.destination_parent_guard) } #[cfg(windows)] @@ -2360,23 +2483,40 @@ pub(crate) async fn rename_all_with_prepared_source( dst_file_path: impl AsRef, base_dir: impl AsRef, publication_root: &PublicationRoot, - _commit_guard: &RenameCommitGuard, + commit_guard: &RenameCommitGuard, lease: Arc, ) -> Result<()> { let src_file_path = src_file_path.as_ref().to_path_buf(); let dst_file_path = dst_file_path.as_ref().to_path_buf(); let base_dir = base_dir.as_ref().to_path_buf(); + #[cfg(not(target_os = "linux"))] let publication_root = publication_root.clone(); + #[cfg(target_os = "linux")] + let _ = publication_root; + let commit_guard = commit_guard.clone(); let operation = { let src_file_path = src_file_path.clone(); let dst_file_path = dst_file_path.clone(); + #[cfg(not(target_os = "linux"))] let base_dir = base_dir.clone(); 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(any(test, feature = "test-util"))] - prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path); - rename_prepared(&src_file_path, &dst_file_path, &preparation) + #[cfg(target_os = "linux")] + { + #[cfg(any(test, feature = "test-util"))] + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path); + return commit_guard + .directories + .rename(&src_file_path, &dst_file_path, Some(&prepared_source)); + } + #[cfg(not(target_os = "linux"))] + { + let _ = commit_guard; + 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(any(test, feature = "test-util"))] + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path); + rename_prepared(&src_file_path, &dst_file_path, &preparation) + } } }; let result = run_blocking_namespace_operation(lease, operation).await; @@ -2393,10 +2533,35 @@ pub(crate) async fn rename_all_with_commit_guard( dst_file_path: impl AsRef, base_dir: impl AsRef, publication_root: &PublicationRoot, - _commit_guard: &RenameCommitGuard, + commit_guard: &RenameCommitGuard, lease: Arc, ) -> Result<()> { - rename_all_with_lease(src_file_path, dst_file_path, base_dir, publication_root, lease).await + #[cfg(target_os = "linux")] + { + let source = src_file_path.as_ref().to_path_buf(); + let destination = dst_file_path.as_ref().to_path_buf(); + let guard = commit_guard.clone(); + let _ = (base_dir, publication_root); + return run_blocking_namespace_operation(lease, move || { + guard.directories.prepare_destination()?; + #[cfg(test)] + prepared_publication_test_hooks::run_rename_destination(&source, &destination); + #[cfg(any(test, feature = "test-util"))] + { + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &source); + prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &destination); + } + guard.directories.rename(&source, &destination, None) + }) + .await + .map_err(to_file_error) + .map_err(DiskError::from); + } + #[cfg(not(target_os = "linux"))] + { + let _ = commit_guard; + rename_all_with_lease(src_file_path, dst_file_path, base_dir, publication_root, lease).await + } } #[tracing::instrument(level = "debug", skip_all)] @@ -2524,7 +2689,7 @@ async fn reliable_rename_inner_with_lease( result } -#[cfg(not(windows))] +#[cfg(not(any(windows, target_os = "linux")))] fn validate_prepared_rename_source(prepared_source: &PreparedRenameSource, src_file_path: &Path) -> io::Result<()> { if prepared_source.path != src_file_path { return Err(io::Error::new( @@ -2724,6 +2889,8 @@ pub(crate) async fn fsync_dir_with_namespace_file_sync_limit( struct RenamePreparation { parent_guard: Option, + #[cfg(target_os = "linux")] + publication_root: PublicationRoot, #[cfg(windows)] _source_parent_guard: ExistingBaseDirectoryGuard, #[cfg(windows)] @@ -2732,15 +2899,32 @@ struct RenamePreparation { #[cfg(not(windows))] fn prepare_rename_with_retry( - src_file_path: &Path, + _src_file_path: &Path, dst_file_path: &Path, base_dir: &Path, publication_root: &PublicationRoot, ) -> io::Result { - let prune_budget = prepare_prune_budget(dst_file_path, base_dir); + let parent_guard = dst_file_path + .parent() + .map(|parent| prepare_destination_parent_with_retry(parent, base_dir, publication_root)) + .transpose()?; + Ok(RenamePreparation { + parent_guard, + #[cfg(target_os = "linux")] + publication_root: publication_root.clone(), + }) +} + +#[cfg(not(windows))] +fn prepare_destination_parent_with_retry( + parent: &Path, + base_dir: &Path, + publication_root: &PublicationRoot, +) -> io::Result { + let prune_budget = parent_prune_budget(parent, base_dir); let mut attempt = 0; loop { - match prepare_rename(src_file_path, dst_file_path, base_dir, publication_root) { + match mkdir_all_below_existing_base_std(parent, base_dir, publication_root) { Ok(preparation) => return Ok(preparation), Err(err) if should_retry_prepare(&err, attempt, prune_budget) => { attempt += 1; @@ -2830,20 +3014,6 @@ fn prepare_rename_with_retry( }) } -#[cfg(not(windows))] -fn prepare_rename( - _src_file_path: &Path, - dst_file_path: &Path, - base_dir: &Path, - publication_root: &PublicationRoot, -) -> io::Result { - let parent_guard = dst_file_path - .parent() - .map(|parent| mkdir_all_below_existing_base_std(parent, base_dir, publication_root)) - .transpose()?; - Ok(RenamePreparation { parent_guard }) -} - /// Publish a prepared rename. The retry budget starts fresh here: preparation /// keeps its own counter, so a chain rebuilt after a concurrent prune must not /// cost the rename its one retry. @@ -2859,9 +3029,23 @@ fn rename_prepared(_src_file_path: &Path, dst_file_path: &Path, preparation: &Re #[cfg(not(windows))] { + // A missing source parent is not a destination-prune race. Open it once + // outside both retry loops, retaining it through any rename retry. + #[cfg(target_os = "linux")] + let source_parent = preparation.publication_root.open_directory( + _src_file_path + .parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing rename source parent"))?, + )?; let mut attempt = 0; loop { - let rename_result = rename_into_existing_parent(_src_file_path, dst_file_path, preparation.parent_guard.as_ref()); + let rename_result = rename_into_existing_parent( + _src_file_path, + dst_file_path, + preparation.parent_guard.as_ref(), + #[cfg(target_os = "linux")] + &source_parent, + ); match rename_result { Ok(()) => return Ok(()), Err(err) if should_retry_rename(&err, attempt) => { @@ -2900,8 +3084,9 @@ fn rename_into_existing_parent( src_file_path: &Path, dst_file_path: &Path, parent_guard: Option<&ExistingBaseDirectoryGuard>, + #[cfg(target_os = "linux")] source_parent: &std::fs::File, ) -> io::Result<()> { - use rustix::fs::{Mode, OFlags, open, renameat}; + use rustix::fs::renameat; let Some(parent_guard) = parent_guard else { let rename_started = rustfs_io_metrics::put_stage_timer(); @@ -2912,6 +3097,7 @@ fn rename_into_existing_parent( ); return result; }; + #[cfg(not(target_os = "linux"))] let src_parent = src_file_path .parent() .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "rename source must have a parent directory"))?; @@ -2921,10 +3107,11 @@ fn rename_into_existing_parent( let dst_name = dst_file_path .file_name() .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "rename destination must have a file name"))?; - let src_parent = open( + #[cfg(not(target_os = "linux"))] + let src_parent = rustix::fs::open( src_parent, - OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, - Mode::empty(), + rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::DIRECTORY | rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::CLOEXEC, + rustix::fs::Mode::empty(), ) .map_err(io::Error::from)?; let dst_parent = parent_guard @@ -2932,6 +3119,9 @@ fn rename_into_existing_parent( .ok_or_else(|| io::Error::other("rename destination parent guard is empty"))?; let rename_started = rustfs_io_metrics::put_stage_timer(); + #[cfg(target_os = "linux")] + let result = renameat(source_parent, src_name, dst_parent, dst_name).map_err(io::Error::from); + #[cfg(not(target_os = "linux"))] let result = renameat(&src_parent, src_name, dst_parent, dst_name).map_err(io::Error::from); rustfs_io_metrics::record_put_object_stage_duration_from( rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL, @@ -3114,16 +3304,18 @@ struct WindowsDirectoryHandle { /// Stable root for namespace-changing disk operations. /// -/// Windows opens the configured endpoint once and keeps that directory identity -/// pinned for the lifetime of the disk. Publication then resolves every source -/// and destination component relative to this handle instead of re-entering the -/// mutable pathname namespace. Other platforms retain the path so callers use a -/// uniform API while their existing `openat`/`renameat` guards remain unchanged. +/// Linux and Windows pin the configured directory for the lifetime of the disk. +/// Linux also exposes a procfs alias for operations not yet using relative IO; +/// both access forms share one root handle and therefore one mount identity. #[derive(Clone)] pub(crate) struct PublicationRoot { - path: PathBuf, + path: Arc, + #[cfg(target_os = "linux")] + directory: Arc, + #[cfg(target_os = "linux")] + io_path: Arc, #[cfg(windows)] - configured_path: PathBuf, + configured_path: Arc, #[cfg(windows)] directory: WindowsDirectoryHandle, } @@ -3137,15 +3329,34 @@ impl PublicationRoot { #[cfg(windows)] let (resolved_path, directory) = open_windows_publication_root(path)?; + #[cfg(target_os = "linux")] + let directory = { + use rustix::fs::{Mode, OFlags, open}; + Arc::new(std::fs::File::from(open( + path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?)) + }; + #[cfg(target_os = "linux")] + let io_path = { + use std::os::fd::AsRawFd; + Arc::new(PathBuf::from(format!("/proc/self/fd/{}/.", directory.as_raw_fd()))) + }; + Ok(Self { #[cfg(not(windows))] - path: path.to_path_buf(), + path: Arc::new(path.to_path_buf()), #[cfg(windows)] - path: resolved_path, + path: Arc::new(resolved_path), #[cfg(windows)] - configured_path: path.to_path_buf(), + configured_path: Arc::new(path.to_path_buf()), #[cfg(windows)] directory, + #[cfg(target_os = "linux")] + directory, + #[cfg(target_os = "linux")] + io_path, }) } @@ -3153,15 +3364,170 @@ impl PublicationRoot { &self.path } + #[cfg(target_os = "linux")] + pub(crate) fn directory(&self) -> &Arc { + &self.directory + } + + #[cfg(target_os = "linux")] + pub(crate) fn io_path(&self) -> &Path { + &self.io_path + } + + pub(crate) fn open_readonly(&self, path: &Path) -> io::Result { + #[cfg(target_os = "linux")] + return self.open_relative(path, rustix::fs::OFlags::RDONLY, rustix::fs::Mode::empty()); + #[cfg(not(target_os = "linux"))] + std::fs::File::open(path) + } + + pub(crate) fn create_truncate(&self, path: &Path) -> io::Result { + #[cfg(target_os = "linux")] + { + use rustix::fs::{Mode, OFlags}; + self.open_relative(path, OFlags::CREATE | OFlags::WRONLY | OFlags::TRUNC, Mode::from_raw_mode(0o666)) + } + #[cfg(not(target_os = "linux"))] + std::fs::OpenOptions::new().create(true).write(true).truncate(true).open(path) + } + + pub(crate) fn create_truncate_with_parent(&self, path: &Path) -> io::Result<(std::fs::File, Option)> { + #[cfg(target_os = "linux")] + { + use rustix::fs::{Mode, OFlags}; + let parent = self.open_directory( + path.parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file parent"))?, + )?; + let name = path + .file_name() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file name"))?; + let file = open_linux_relative( + &parent, + Path::new(name), + OFlags::CREATE | OFlags::WRONLY | OFlags::TRUNC, + Mode::from_raw_mode(0o666), + )?; + Ok((file, Some(parent))) + } + #[cfg(not(target_os = "linux"))] + Ok((self.create_truncate(path)?, None)) + } + + #[cfg(target_os = "linux")] + fn open_relative(&self, path: &Path, flags: rustix::fs::OFlags, mode: rustix::fs::Mode) -> io::Result { + let relative = path + .strip_prefix(self.io_path.as_path()) + .or_else(|_| path.strip_prefix(self.path.as_path())) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path must remain below its publication root"))?; + open_linux_relative(&self.directory, relative, flags, mode) + } + + #[cfg(target_os = "linux")] + fn open_directory(&self, path: &Path) -> io::Result { + self.open_relative( + path, + rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::DIRECTORY, + rustix::fs::Mode::empty(), + ) + } + #[cfg(windows)] fn relative_path<'a>(&self, path: &'a Path) -> io::Result<&'a Path> { // The configured path only derives a suffix; traversal stays rooted at the pinned directory handle. - path.strip_prefix(&self.path) - .or_else(|_| path.strip_prefix(&self.configured_path)) + path.strip_prefix(self.path.as_path()) + .or_else(|_| path.strip_prefix(self.configured_path.as_path())) .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path must remain below its publication root")) } } +/// Open the actual IO handle with the same containment policy as path validation. +/// Older kernels retain fd-relative traversal; only ENOSYS enables the fallback. +#[cfg(target_os = "linux")] +fn open_linux_relative( + directory: &std::fs::File, + relative: &Path, + flags: rustix::fs::OFlags, + mode: rustix::fs::Mode, +) -> io::Result { + use rustix::fs::{OFlags, ResolveFlags, openat2}; + if relative + .components() + .any(|component| !matches!(component, Component::Normal(_) | Component::CurDir)) + { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid relative disk path")); + } + let relative = if relative.as_os_str().is_empty() { + Path::new(".") + } else { + relative + }; + let flags = flags | OFlags::CLOEXEC | OFlags::NOFOLLOW; + match openat2(directory, relative, flags, mode, ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS) { + Ok(fd) => Ok(fd.into()), + Err(rustix::io::Errno::NOSYS) => open_linux_relative_legacy(directory, relative, flags, mode), + Err(rustix::io::Errno::LOOP | rustix::io::Errno::XDEV) => Err(DiskError::InvalidPath.into()), + Err(err) => Err(err.into()), + } +} + +#[cfg(target_os = "linux")] +fn open_linux_relative_legacy( + directory: &std::fs::File, + relative: &Path, + flags: rustix::fs::OFlags, + mode: rustix::fs::Mode, +) -> io::Result { + use rustix::fs::OFlags; + + let mut parent = None; + let mut components = relative + .components() + .filter_map(|component| match component { + Component::Normal(name) => Some(name), + _ => None, + }) + .peekable(); + while let Some(name) = components.next() { + let current = parent.as_ref().unwrap_or(directory); + if components.peek().is_none() { + return open_linux_relative_component(current, name, flags, mode); + } + parent = Some(open_linux_relative_component( + current, + name, + OFlags::PATH | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + rustix::fs::Mode::empty(), + )?); + } + open_linux_relative_component(directory, std::ffi::OsStr::new("."), flags, mode) +} + +#[cfg(target_os = "linux")] +fn open_linux_relative_component( + directory: &std::fs::File, + name: &std::ffi::OsStr, + flags: rustix::fs::OFlags, + mode: rustix::fs::Mode, +) -> io::Result { + use rustix::fs::{AtFlags, FileType, openat, statat}; + use rustix::io::Errno; + match openat(directory, name, flags, mode) { + Ok(fd) => Ok(fd.into()), + Err(Errno::LOOP) => Err(DiskError::InvalidPath.into()), + // O_DIRECTORY | O_NOFOLLOW reports ENOTDIR for an intermediate link. + // Inspect only this failure so ordinary non-directory errors retain + // their classification without adding a stat to successful opens. + Err(Errno::NOTDIR) + if statat(directory, name, AtFlags::SYMLINK_NOFOLLOW) + .is_ok_and(|metadata| FileType::from_raw_mode(metadata.st_mode) == FileType::Symlink) => + { + Err(DiskError::InvalidPath.into()) + } + Err(err) => Err(err.into()), + } +} + #[cfg(windows)] #[derive(Clone)] pub(crate) struct ExistingBaseDirectoryGuard { @@ -3199,6 +3565,8 @@ pub(crate) type ExistingBaseDirectoryGuard = (); #[derive(Clone)] pub(crate) struct RenameCommitGuard { + #[cfg(target_os = "linux")] + directories: Arc, #[cfg(windows)] source_parent: PathBuf, #[cfg(windows)] @@ -3209,6 +3577,108 @@ pub(crate) struct RenameCommitGuard { destination_parent_guard: ExistingBaseDirectoryGuard, } +#[cfg(target_os = "linux")] +struct LinuxCommitDirectory { + file: Arc, + device: u64, + inode: u64, +} + +#[cfg(target_os = "linux")] +impl LinuxCommitDirectory { + fn new(file: std::fs::File) -> io::Result { + use std::os::unix::fs::MetadataExt; + let metadata = file.metadata()?; + Ok(Self { + file: Arc::new(file), + device: metadata.dev(), + inode: metadata.ino(), + }) + } +} + +#[cfg(target_os = "linux")] +struct LinuxCommitDirectories { + root: PublicationRoot, + source_path: PathBuf, + destination_path: PathBuf, + source: LinuxCommitDirectory, + destination: std::sync::OnceLock, + destination_base: PathBuf, +} + +#[cfg(target_os = "linux")] +impl LinuxCommitDirectories { + fn prepare_destination(&self) -> io::Result<&LinuxCommitDirectory> { + if let Some(directory) = self.destination.get() { + return Ok(directory); + } + let mut parents = prepare_destination_parent_with_retry(&self.destination_path, &self.destination_base, &self.root)?; + let directory = std::fs::File::from( + parents + .pop() + .ok_or_else(|| io::Error::other("missing destination directory guard"))?, + ); + let directory = LinuxCommitDirectory::new(directory)?; + Ok(self.destination.get_or_init(|| directory)) + } + + fn file_name<'a>(&self, path: &'a Path, source: bool) -> io::Result<&'a std::ffi::OsStr> { + let parent = if source { &self.source_path } else { &self.destination_path }; + if path.parent() != Some(parent) { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "file parent does not match its commit guard")); + } + path.file_name() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file name")) + } + + fn validate(&self) -> io::Result<()> { + use std::os::unix::fs::MetadataExt; + // A pinned directory may have been unlinked or replaced. Do not report a + // successful publication into a detached object tree. + for (path, directory) in [ + (&self.source_path, &self.source), + (&self.destination_path, self.prepare_destination()?), + ] { + let current = self.root.open_directory(path)?.metadata()?; + if (current.dev(), current.ino()) != (directory.device, directory.inode) { + return Err(io::Error::new(io::ErrorKind::InvalidData, "commit directory identity changed")); + } + } + Ok(()) + } + + fn rename(&self, source: &Path, destination: &Path, prepared: Option<&PreparedRenameSource>) -> io::Result<()> { + let source_name = self.file_name(source, true)?; + let destination_name = self.file_name(destination, false)?; + let destination_parent = self.prepare_destination()?; + self.validate()?; + let mut attempt = 0; + loop { + if let Some(prepared) = prepared { + let metadata = rustix::fs::statat(&*self.source.file, source_name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW)?; + if prepared.path != source || metadata.st_dev != prepared.device || metadata.st_ino != prepared.inode { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "rename source identity changed while publication was prepared", + )); + } + } + let started = rustfs_io_metrics::put_stage_timer(); + let result = rustix::fs::renameat(&*self.source.file, source_name, &*destination_parent.file, destination_name) + .map_err(io::Error::from); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL, + started, + ); + match result { + Err(err) if should_retry_rename(&err, attempt) => attempt += 1, + result => return result, + } + } + } +} + pub(crate) struct RenameDestinationPathGuard { #[cfg(windows)] directory: PathBuf, @@ -3408,6 +3878,29 @@ pub(crate) fn prepare_rename_commit_guard( destination_base: &Path, publication_root: &PublicationRoot, ) -> io::Result { + #[cfg(target_os = "linux")] + { + existing_base_directory_suffix(destination_parent, destination_base)?; + let source = LinuxCommitDirectory::new(publication_root.open_directory(source_parent)?)?; + let destination = std::sync::OnceLock::new(); + match publication_root.open_directory(destination_parent) { + Ok(directory) => { + let _ = destination.set(LinuxCommitDirectory::new(directory)?); + } + Err(err) if err.kind() == io::ErrorKind::NotFound => {} + Err(err) => return Err(err), + } + return Ok(RenameCommitGuard { + directories: Arc::new(LinuxCommitDirectories { + root: publication_root.clone(), + source_path: source_parent.to_path_buf(), + destination_path: destination_parent.to_path_buf(), + source, + destination, + destination_base: destination_base.to_path_buf(), + }), + }); + } #[cfg(windows)] { // A same-directory rename must use one shared-write parent handle: @@ -3436,7 +3929,7 @@ pub(crate) fn prepare_rename_commit_guard( }) } - #[cfg(not(windows))] + #[cfg(not(any(windows, target_os = "linux")))] { let _ = (source_parent, destination_parent, destination_base, publication_root); Ok(RenameCommitGuard {}) @@ -4232,11 +4725,7 @@ fn windows_rename_source_is_allowed(attributes: u32, reparse_tag: u32) -> bool { attributes & FILE_ATTRIBUTE_REPARSE_POINT == 0 || reparse_tag == IO_REPARSE_TAG_DEDUP } -pub(crate) fn mkdir_all_below_existing_base_std( - dir_path: &Path, - base_dir: &Path, - publication_root: &PublicationRoot, -) -> io::Result { +fn existing_base_directory_suffix<'a>(dir_path: &'a Path, base_dir: &Path) -> io::Result<&'a Path> { let relative = dir_path .strip_prefix(base_dir) .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory"))?; @@ -4249,15 +4738,31 @@ pub(crate) fn mkdir_all_below_existing_base_std( } } + Ok(relative) +} + +pub(crate) fn mkdir_all_below_existing_base_std( + dir_path: &Path, + base_dir: &Path, + publication_root: &PublicationRoot, +) -> io::Result { + let relative = existing_base_directory_suffix(dir_path, base_dir)?; + #[cfg(unix)] { - let _ = publication_root; - use rustix::fs::{Mode, OFlags, mkdirat, open, openat}; + use rustix::fs::{Mode, OFlags, mkdirat, openat}; use rustix::io::Errno; let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC; let mode = Mode::RWXU | Mode::RWXG | Mode::RWXO; - let mut parents = vec![open(base_dir, flags, Mode::empty()).map_err(io::Error::from)?]; + #[cfg(target_os = "linux")] + let base = publication_root.open_directory(base_dir)?.into(); + #[cfg(not(target_os = "linux"))] + let base = { + let _ = publication_root; + rustix::fs::open(base_dir, flags, Mode::empty()).map_err(io::Error::from)? + }; + let mut parents = vec![base]; #[cfg(test)] let mut walked_path = base_dir.to_path_buf(); @@ -4377,10 +4882,18 @@ fn should_retry_rename(err: &io::Error, attempt: usize) -> bool { /// steal an attempt by making that same upward progress. A destination whose /// parent *is* the base gets a budget of zero, keeping `NotFound` immediately /// terminal for speculative cleanup renames. +#[cfg(any(windows, test))] fn prepare_prune_budget(dst_file_path: &Path, base_dir: &Path) -> usize { dst_file_path .parent() - .and_then(|parent| parent.strip_prefix(base_dir).ok()) + .map(|parent| parent_prune_budget(parent, base_dir)) + .unwrap_or(0) +} + +fn parent_prune_budget(parent: &Path, base_dir: &Path) -> usize { + parent + .strip_prefix(base_dir) + .ok() .map(|relative| relative.components().count()) .unwrap_or(0) } @@ -4570,7 +5083,11 @@ mod tests { .first() .expect("test publication root requires at least one path") .to_path_buf(); - while !paths.iter().all(|path| path.starts_with(&common)) { + // The disk root must exist independently of a missing or symlinked + // rename base; those are inputs whose rejection the tests exercise. + while !paths.iter().all(|path| path.starts_with(&common)) + || !std::fs::symlink_metadata(&common).is_ok_and(|metadata| metadata.is_dir()) + { assert!(common.pop(), "test paths must share an absolute root"); } PublicationRoot::new(&common).expect("test publication root should open") @@ -7814,3 +8331,172 @@ mod tests { assert_eq!(err.kind(), io::ErrorKind::NotFound); } } + +#[cfg(all(test, target_os = "linux"))] +mod fd_relative_tests { + use super::*; + use std::io::{Read, Write}; + + #[tokio::test] + async fn guarded_first_publication_recovers_from_a_concurrent_parent_prune() { + let temp = tempfile::tempdir().expect("temporary disk"); + let root = PublicationRoot::new(temp.path()).expect("pinned root"); + let source = temp.path().join("tmp"); + let base = temp.path().join("bucket"); + let shared = base.join("prefix"); + let destination = shared.join("object"); + std::fs::create_dir(&source).expect("staging directory"); + std::fs::create_dir_all(&shared).expect("shared prefix"); + let guard = prepare_rename_commit_guard(&source, &destination, &base, &root).expect("commit directories"); + let src = source.join("xl.meta"); + let dst = destination.join("xl.meta"); + let mut staged = create_prepared_rename_source_with_commit_guard(&src, &dst, &guard).expect("staged metadata"); + staged.write_all(b"metadata", true).expect("write metadata"); + let pruned = shared.clone(); + prepare_rename_test_hooks::queue_after_component_opened(&shared, move || { + std::fs::remove_dir(pruned).expect("prune empty prefix"); + }); + let lease = acquire_namespace_mutation_lease(&dst).await; + rename_all_with_prepared_source(staged, &src, &dst, &base, &root, &guard, lease) + .await + .expect("publication must rebuild the pruned parent"); + assert_eq!(std::fs::read(dst).expect("published metadata"), b"metadata"); + assert!(!src.exists()); + } + + #[test] + fn legacy_relative_open_rejects_symlinks_and_opens_existing_components() { + use rustix::fs::{Mode, OFlags}; + let temp = tempfile::tempdir().expect("temporary disk"); + std::fs::create_dir(temp.path().join("parent")).expect("parent directory"); + std::fs::write(temp.path().join("parent/data"), b"data").expect("file data"); + std::os::unix::fs::symlink("parent", temp.path().join("link")).expect("directory link"); + std::os::unix::fs::symlink("data", temp.path().join("parent/link")).expect("file link"); + let root = PublicationRoot::new(temp.path()).expect("pinned root"); + let flags = OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC; + let mut bytes = Vec::new(); + open_linux_relative_legacy(root.directory(), Path::new("parent/data"), flags, Mode::empty()) + .expect("legacy open") + .read_to_end(&mut bytes) + .expect("legacy read"); + assert_eq!(bytes, b"data"); + for relative in ["link/data", "parent/link"] { + assert!(open_linux_relative_legacy(root.directory(), Path::new(relative), flags, Mode::empty()).is_err()); + } + } + + #[test] + fn root_relative_io_retains_mount_identity_after_path_replacement() { + let temp = tempfile::tempdir().expect("temporary disk"); + let disk = temp.path().join("disk"); + std::fs::create_dir(&disk).expect("disk root"); + std::fs::write(disk.join("data"), b"original").expect("original data"); + let root = PublicationRoot::new(&disk).expect("pinned root"); + let original = temp.path().join("original"); + std::fs::rename(&disk, &original).expect("move root"); + std::fs::create_dir(&disk).expect("replacement root"); + std::fs::write(disk.join("data"), b"replacement").expect("replacement data"); + for path in [disk.join("data"), root.io_path().join("data")] { + let mut actual = Vec::new(); + root.open_readonly(&path) + .expect("pinned read") + .read_to_end(&mut actual) + .expect("read bytes"); + assert_eq!(actual, b"original"); + } + let (mut file, parent) = root.create_truncate_with_parent(&disk.join("new")).expect("pinned creation"); + file.write_all(b"new").expect("write bytes"); + file.sync_data().expect("file sync"); + fsync_directory_handle_std(&disk, parent.as_ref()).expect("pinned parent sync"); + assert_eq!(std::fs::read(original.join("new")).expect("original disk file"), b"new"); + assert!(!disk.join("new").exists()); + } + + #[test] + fn root_relative_io_rejects_links_and_parent_escape_without_truncation() { + let temp = tempfile::tempdir().expect("temporary disk"); + let disk = temp.path().join("disk"); + std::fs::create_dir(&disk).expect("disk root"); + let outside = temp.path().join("outside"); + std::fs::write(&outside, b"preserved").expect("outside data"); + std::os::unix::fs::symlink(&outside, disk.join("link")).expect("file link"); + std::os::unix::fs::symlink(temp.path(), disk.join("parent")).expect("directory link"); + let root = PublicationRoot::new(&disk).expect("pinned root"); + for path in [disk.join("link"), disk.join("parent/outside"), disk.join("../outside")] { + assert!(root.open_readonly(&path).is_err(), "unsafe read: {path:?}"); + assert!(root.create_truncate(&path).is_err(), "unsafe write: {path:?}"); + } + assert_eq!(std::fs::read(&outside).expect("outside remains readable"), b"preserved"); + } + + #[tokio::test] + async fn guarded_publication_reuses_parents_and_syncs_without_changing_bytes() { + for grouped in [false, true] { + let _grouping = set_dst_dir_fsync_group_commit_for_test(grouped); + let temp = tempfile::tempdir().expect("temporary disk"); + let root = PublicationRoot::new(temp.path()).expect("pinned root"); + let source = temp.path().join("tmp"); + let destination = temp.path().join("bucket/object"); + let base = temp.path().join("bucket"); + std::fs::create_dir(&source).expect("staging directory"); + std::fs::create_dir(&base).expect("bucket directory"); + let guard = prepare_rename_commit_guard(&source, &destination, &base, &root).expect("commit directories"); + assert!(!destination.exists(), "preparation must not create the object directory"); + let src = source.join("xl.meta"); + let dst = destination.join("xl.meta"); + assert!( + read_destination_file_with_commit_guard(&dst, &guard) + .expect("absent read") + .is_none() + ); + let mut staged = create_prepared_rename_source_with_commit_guard(&src, &dst, &guard).expect("staged metadata"); + staged.write_all(b"metadata", true).expect("write metadata"); + let lease = acquire_namespace_mutation_lease(&dst).await; + rename_all_with_prepared_source(staged, &src, &dst, &base, &root, &guard, lease.clone()) + .await + .expect("publish"); + fsync_commit_directory(&destination, &guard, lease, None) + .await + .expect("directory sync"); + assert_eq!( + read_destination_file_with_commit_guard(&dst, &guard).expect("committed read"), + Some(b"metadata".to_vec()) + ); + assert!(!src.exists()); + assert_eq!(std::fs::read_dir(&destination).expect("object directory").count(), 1); + } + } + + #[tokio::test] + async fn guarded_publication_rejects_replaced_directory_and_source() { + for replace_directory in [false, true] { + let temp = tempfile::tempdir().expect("temporary disk"); + let root = PublicationRoot::new(temp.path()).expect("pinned root"); + let source = temp.path().join("tmp"); + let destination = temp.path().join("object"); + std::fs::create_dir(&source).expect("staging directory"); + std::fs::create_dir(&destination).expect("existing object directory"); + let guard = prepare_rename_commit_guard(&source, &destination, temp.path(), &root).expect("commit directories"); + let src = source.join("xl.meta"); + let dst = destination.join("xl.meta"); + std::fs::write(&dst, b"old").expect("old metadata"); + let mut staged = create_prepared_rename_source_with_commit_guard(&src, &dst, &guard).expect("staged metadata"); + staged.write_all(b"new", false).expect("new metadata"); + if replace_directory { + std::fs::rename(&destination, temp.path().join("detached")).expect("replace destination"); + std::fs::create_dir(&destination).expect("replacement directory"); + std::fs::write(&dst, b"old").expect("replacement metadata"); + } else { + std::fs::rename(&src, source.join("detached")).expect("replace staged file"); + std::fs::write(&src, b"unrelated").expect("replacement file"); + } + let lease = acquire_namespace_mutation_lease(&dst).await; + assert!( + rename_all_with_prepared_source(staged, &src, &dst, temp.path(), &root, &guard, lease) + .await + .is_err() + ); + assert_eq!(std::fs::read(&dst).expect("old metadata survives"), b"old"); + } + } +} diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 390938652..4a3ab5192 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3804,6 +3804,10 @@ impl SetDisks { ); } + fn needs_old_data_cleanup_receipt(old_data_dir: Uuid, committed_data_dir: Option, epoch: Option) -> bool { + epoch.is_some() && committed_data_dir != Some(old_data_dir) + } + pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts( &self, disks: &[Option], @@ -3813,10 +3817,10 @@ impl SetDisks { committed_data_dir: Option, epoch: Option, ) { - let Some(epoch) = epoch else { return }; - if committed_data_dir == Some(old_data_dir) { + if !Self::needs_old_data_cleanup_receipt(old_data_dir, committed_data_dir, epoch) { return; } + let Some(epoch) = epoch else { return }; let receipt = OldDataCleanupReceipt::new(epoch, old_data_dir, committed_data_dir); let Ok(encoded) = receipt.encode() else { return; @@ -5540,7 +5544,9 @@ impl SetDisks { let rename_stage_elapsed = rename_stage_start.elapsed(); let rename_stage_ms = rename_stage_elapsed.as_millis() as u64; - if let Some(old_dir) = op_old_dir { + if let Some(old_dir) = op_old_dir + && Self::needs_old_data_cleanup_receipt(old_dir, committed_data_dir, transaction_epoch) + { rustfs_io_metrics::record_put_object_stage_duration( rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_CLEANUP_RECEIPT_TRIGGERED, 1.0, @@ -23696,3 +23702,19 @@ mod single_delete_namespace_owner_tests { assert_single_delete_physical_owner("rollback").await; } } + +#[cfg(test)] +mod cleanup_receipt_admission_tests { + use super::*; + + #[test] + fn receipt_admission_skips_only_work_that_the_writer_would_skip() { + let old = Uuid::new_v4(); + let current = Uuid::new_v4(); + let epoch = Uuid::new_v4(); + assert!(!SetDisks::needs_old_data_cleanup_receipt(old, Some(current), None)); + assert!(!SetDisks::needs_old_data_cleanup_receipt(old, Some(old), Some(epoch))); + assert!(SetDisks::needs_old_data_cleanup_receipt(old, Some(current), Some(epoch))); + assert!(SetDisks::needs_old_data_cleanup_receipt(old, None, Some(epoch))); + } +} diff --git a/crates/ecstore/tests/fd_relative_io_test.rs b/crates/ecstore/tests/fd_relative_io_test.rs new file mode 100644 index 000000000..0383333e0 --- /dev/null +++ b/crates/ecstore/tests/fd_relative_io_test.rs @@ -0,0 +1,243 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#![cfg(all(target_os = "linux", feature = "test-util"))] + +mod storage_api; + +use bytes::Bytes; +use rustfs_filemeta::{FileInfo, ObjectPartInfo}; +use std::{path::Path, time::Duration}; +use storage_api::fd_relative::{ + DiskAPI, DiskError, DiskOption, DiskStore, Endpoint, LocalPublicationPause, LocalPublicationStage, ReadOptions, new_disk, +}; +use time::OffsetDateTime; +use uuid::Uuid; + +const RUSTFS_META_TMP_BUCKET: &str = ".rustfs.sys/tmp"; + +async fn disk_at(path: &Path) -> DiskStore { + new_disk( + &Endpoint::try_from(path.to_str().expect("UTF-8 test path")).expect("endpoint"), + &DiskOption::default(), + ) + .await + .expect("local disk") +} + +fn metadata(data: Bytes, inline: bool, directory: Uuid) -> FileInfo { + let mut fi = FileInfo::new("object", 1, 0); + fi.erasure.index = 1; + fi.version_id = Some(Uuid::nil()); + fi.data_dir = Some(directory); + fi.size = i64::try_from(data.len()).expect("test data size"); + fi.parts = vec![ObjectPartInfo { + number: 1, + size: data.len(), + actual_size: fi.size, + ..Default::default() + }]; + fi.mod_time = Some(OffsetDateTime::now_utc()); + fi.data = inline.then_some(data); + if inline { + fi.set_inline_data(); + } + fi +} + +async fn put(disk: &DiskStore, staging: &str, data: Bytes, inline: bool) -> Uuid { + let directory = Uuid::new_v4(); + if !inline { + disk.write_all(RUSTFS_META_TMP_BUCKET, &format!("{staging}/{directory}/part.1"), data.clone()) + .await + .expect("stage shard"); + } + disk.rename_data(RUSTFS_META_TMP_BUCKET, staging, metadata(data, inline, directory), "bucket", "object") + .await + .expect("commit object"); + directory +} + +#[tokio::test] +async fn pinned_root_reads_and_writes_original_disk_after_path_replacement() { + let temp = tempfile::tempdir().expect("test directory"); + let root = temp.path().join("disk"); + std::fs::create_dir(&root).expect("disk root"); + let disk = disk_at(&root).await; + disk.make_volume("bucket").await.expect("bucket"); + disk.write_all("bucket", "metadata", Bytes::from_static(b"original")) + .await + .expect("initial write"); + let moved = temp.path().join("original"); + std::fs::rename(&root, &moved).expect("move mount path"); + std::fs::create_dir_all(root.join("bucket")).expect("replacement disk"); + std::fs::write(root.join("bucket/metadata"), b"replacement").expect("replacement sentinel"); + assert_eq!( + disk.read_all("bucket", "metadata").await.expect("pinned read"), + Bytes::from_static(b"original") + ); + disk.write_all("bucket", "metadata", Bytes::from_static(b"updated")) + .await + .expect("pinned write"); + assert_eq!(std::fs::read(moved.join("bucket/metadata")).expect("original disk"), b"updated"); + assert_eq!(std::fs::read(root.join("bucket/metadata")).expect("replacement disk"), b"replacement"); +} + +#[tokio::test] +async fn relative_reads_and_writes_reject_symlink_escape() { + let temp = tempfile::tempdir().expect("test directory"); + let disk = disk_at(temp.path()).await; + disk.make_volume("bucket").await.expect("bucket"); + let outside = tempfile::tempdir().expect("outside directory"); + std::fs::write(outside.path().join("sentinel"), b"preserved").expect("sentinel"); + std::os::unix::fs::symlink(outside.path(), temp.path().join("bucket/link")).expect("directory link"); + std::os::unix::fs::symlink(outside.path().join("sentinel"), temp.path().join("bucket/file-link")).expect("file link"); + for path in ["link/sentinel", "file-link"] { + assert!(matches!(disk.read_all("bucket", path).await, Err(DiskError::InvalidPath))); + assert!(matches!( + disk.write_all("bucket", path, Bytes::from_static(b"unsafe")).await, + Err(DiskError::InvalidPath) + )); + } + assert_eq!(std::fs::read(outside.path().join("sentinel")).expect("outside bytes"), b"preserved"); +} + +#[tokio::test] +async fn inline_and_non_inline_overwrites_remain_readable_after_reopen() { + for inline in [true, false] { + let temp = tempfile::tempdir().expect("test directory"); + let disk = disk_at(temp.path()).await; + disk.make_volume("bucket").await.expect("bucket"); + put(&disk, "first", Bytes::from_static(b"old bytes"), inline).await; + let directory = put(&disk, "second", Bytes::from_static(b"new bytes"), inline).await; + drop(disk); + let disk = disk_at(temp.path()).await; + let fi = disk + .read_version( + "bucket", + "bucket", + "object", + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("committed metadata after reopen"); + assert_eq!(fi.data_dir, Some(directory)); + if inline { + assert_eq!(fi.data, Some(Bytes::from_static(b"new bytes"))); + } else { + assert_eq!( + disk.read_all("bucket", &format!("object/{directory}/part.1")) + .await + .expect("shard bytes"), + Bytes::from_static(b"new bytes") + ); + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn prepared_publication_rejects_source_or_directory_replacement() { + for replace_directory in [false, true] { + let temp = tempfile::tempdir().expect("test directory"); + let disk = disk_at(temp.path()).await; + disk.make_volume("bucket").await.expect("bucket"); + put(&disk, "first", Bytes::from_static(b"old bytes"), true).await; + let old = std::fs::read(temp.path().join("bucket/object/xl.meta")).expect("old metadata"); + let mut pause = LocalPublicationPause::install(&disk, "bucket", "object/xl.meta", LocalPublicationStage::PreparedRename) + .expect("publication pause"); + let writer = disk.clone(); + let task = tokio::spawn(async move { + writer + .rename_data( + RUSTFS_META_TMP_BUCKET, + "second", + metadata(Bytes::from_static(b"new bytes"), true, Uuid::new_v4()), + "bucket", + "object", + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), pause.entered()) + .await + .expect("publication reached pause") + .expect("pause entered"); + if replace_directory { + std::fs::rename(temp.path().join("bucket/object"), temp.path().join("detached")).expect("detach object"); + std::fs::create_dir(temp.path().join("bucket/object")).expect("replacement directory"); + std::fs::write(temp.path().join("bucket/object/xl.meta"), &old).expect("replacement metadata"); + } else { + let src = temp.path().join(RUSTFS_META_TMP_BUCKET).join("second/xl.meta"); + std::fs::rename(&src, src.with_extension("detached")).expect("detach staged metadata"); + std::fs::write(src, b"unrelated").expect("replacement staged file"); + } + drop(pause); + assert!(task.await.expect("writer task").is_err()); + assert_eq!( + std::fs::read(temp.path().join("bucket/object/xl.meta")).expect("old metadata survives"), + old + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cancelled_waiter_keeps_publication_serialized_until_physical_completion() { + let temp = tempfile::tempdir().expect("test directory"); + let disk = disk_at(temp.path()).await; + disk.make_volume("bucket").await.expect("bucket"); + put(&disk, "first", Bytes::from_static(b"old bytes"), true).await; + let mut pause = LocalPublicationPause::install(&disk, "bucket", "object/xl.meta", LocalPublicationStage::PreparedRename) + .expect("publication pause"); + let writer = disk.clone(); + let pending = tokio::spawn(async move { put(&writer, "shared-stage", Bytes::from_static(b"cancelled waiter"), true).await }); + tokio::time::timeout(Duration::from_secs(30), pause.entered()) + .await + .expect("publication reached pause") + .expect("pause entered"); + pending.abort(); + assert!(pending.await.expect_err("waiter aborted").is_cancelled()); + let writer = disk.clone(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let mut retry = tokio::spawn(async move { + let _ = started_tx.send(()); + put(&writer, "shared-stage", Bytes::from_static(b"latest bytes"), true).await + }); + started_rx.await.expect("retry started"); + assert!( + tokio::time::timeout(Duration::from_millis(200), &mut retry).await.is_err(), + "a retry must not publish while the cancelled writer still owns physical IO" + ); + drop(pause); + tokio::time::timeout(Duration::from_secs(30), retry) + .await + .expect("retry completed") + .expect("retry task"); + let fi = disk + .read_version( + "bucket", + "bucket", + "object", + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("latest metadata"); + assert_eq!(fi.data, Some(Bytes::from_static(b"latest bytes"))); +} diff --git a/crates/ecstore/tests/storage_api.rs b/crates/ecstore/tests/storage_api.rs index 84717f088..531b543d2 100644 --- a/crates/ecstore/tests/storage_api.rs +++ b/crates/ecstore/tests/storage_api.rs @@ -67,3 +67,9 @@ pub(crate) mod cleanup_isolation { pub(crate) use super::{DiskAPI, DiskOption, Endpoint, new_disk}; pub(crate) use rustfs_ecstore::api::disk::{DeleteOptions, DiskError}; } + +#[cfg(all(feature = "test-util", target_os = "linux"))] +pub(crate) mod fd_relative { + pub(crate) use super::{DiskAPI, DiskOption, DiskStore, Endpoint, new_disk}; + pub(crate) use rustfs_ecstore::api::disk::{DiskError, LocalPublicationPause, LocalPublicationStage, ReadOptions}; +}