diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 774ab8779..3fe4d432f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -515,6 +515,28 @@ pub enum InternalBuf<'a> { Owned(Bytes), } +/// Durability mode for `write_all_internal`. +/// +/// `FileOnly` is reserved for tmp files the caller immediately renames away. +/// The safe-rename recipe (file content fdatasync -> rename -> fsync of the +/// destination parent directory) never needs the tmp directory entry to be +/// durable: the rename removes it, and a crash before the rename means the +/// operation was never acknowledged, so there is nothing to recover. Files +/// that stay where they are written (format.json via `write_all_public`, the +/// old-metadata rollback backup in `rename_data`, ...) must use `FileAndDir` +/// so both the contents and the new directory entry survive power loss. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum SyncMode { + /// No fsync; durability is not required (or drive sync is disabled). + None, + /// fdatasync the file contents, then fsync its parent directory. + FileAndDir, + /// fdatasync only the file contents. Only valid when the caller renames + /// the file away right after the write and fsyncs the rename + /// destination's parent directory before acknowledging. + FileOnly, +} + struct FileCacheReclaimWriter { inner: File, reclaim_len: usize, @@ -2462,7 +2484,12 @@ impl LocalDisk { let tmp_volume_dir = self.get_bucket_path(super::RUSTFS_META_TMP_BUCKET)?; let tmp_file_path = self.get_object_path(super::RUSTFS_META_TMP_BUCKET, Uuid::new_v4().to_string().as_str())?; - self.write_all_internal(&tmp_file_path, InternalBuf::Ref(buf), sync, &tmp_volume_dir) + // The tmp file is renamed to its final location right below, so only + // its contents must be durable here (SyncMode::FileOnly): the rename + // drops the tmp directory entry, and the destination parent directory + // is fsynced after the rename. + let tmp_sync = if sync { SyncMode::FileOnly } else { SyncMode::None }; + self.write_all_internal(&tmp_file_path, InternalBuf::Ref(buf), tmp_sync, &tmp_volume_dir) .await?; rename_all(tmp_file_path, &file_path, volume_dir).await?; @@ -2486,14 +2513,17 @@ impl LocalDisk { let volume_dir = self.get_bucket_path(volume)?; - self.write_all_private(volume, path, data, true, &volume_dir).await?; + // Files written here (format.json, ...) stay where they land — no + // rename follows — so the new directory entry must be fsynced too. + self.write_all_private(volume, path, data, SyncMode::FileAndDir, &volume_dir) + .await?; Ok(()) } // write_all_private with check_path_length #[tracing::instrument(level = "debug", skip_all)] - async fn write_all_private(&self, volume: &str, path: &str, buf: Bytes, sync: bool, skip_parent: &Path) -> Result<()> { + async fn write_all_private(&self, volume: &str, path: &str, buf: Bytes, sync: SyncMode, skip_parent: &Path) -> Result<()> { let file_path = self.get_object_path(volume, path)?; check_path_length(file_path.to_string_lossy().as_ref())?; @@ -2503,24 +2533,33 @@ impl LocalDisk { Ok(()) } // write_all_internal do write file - async fn write_all_internal(&self, file_path: &Path, data: InternalBuf<'_>, sync: bool, skip_parent: &Path) -> Result<()> { + async fn write_all_internal( + &self, + file_path: &Path, + data: InternalBuf<'_>, + sync: SyncMode, + skip_parent: &Path, + ) -> Result<()> { let skip_parent = if skip_parent.as_os_str().is_empty() { self.root.as_path() } else { skip_parent }; - let sync = sync && drive_sync_enabled(); + let sync = if drive_sync_enabled() { sync } else { SyncMode::None }; match data { InternalBuf::Ref(buf) => { let mut f = self.open_file(file_path, O_CREATE | O_WRONLY | O_TRUNC, skip_parent).await?; f.write_all(buf).await.map_err(to_file_error)?; - if sync { + if sync != SyncMode::None { f.sync_data().await.map_err(to_file_error)?; // Persist the directory entry too, so a freshly created file - // survives power loss along with its contents. - if let Some(parent) = file_path.parent() { + // survives power loss along with its contents. Skipped for + // FileOnly: the caller renames the file away immediately. + if sync == SyncMode::FileAndDir + && let Some(parent) = file_path.parent() + { os::fsync_dir(parent).await.map_err(to_file_error)?; } } @@ -2542,9 +2581,14 @@ impl LocalDisk { .map_err(to_file_error)?; std::io::Write::write_all(&mut f, buf.as_ref()).map_err(to_file_error)?; - if sync { + if sync != SyncMode::None { f.sync_data().map_err(to_file_error)?; - if let Some(parent) = path.parent() { + // See the Ref branch above: FileOnly callers rename the + // file away, so the tmp directory entry never needs to + // become durable. + if sync == SyncMode::FileAndDir + && let Some(parent) = path.parent() + { os::fsync_dir_std(parent).map_err(to_file_error)?; } } @@ -4080,11 +4124,16 @@ impl DiskAPI for LocalDisk { let new_dst_buf = xlmeta.marshal_msg()?; let src_file_parent = src_file_path.parent().unwrap_or(src_volume_dir.as_path()); + // This tmp xl.meta is renamed onto dst_file_path at the commit + // point below, so only its contents must be durable before the + // rename (SyncMode::FileOnly); the dst parent directory is fsynced + // after the commit rename, and a crash before the rename means the + // PUT was never acknowledged. self.write_all_private( src_volume, &format!("{}/{}", &src_path, STORAGE_FORMAT_FILE), new_dst_buf.into(), - true, + SyncMode::FileOnly, src_file_parent, ) .await?; @@ -4133,6 +4182,9 @@ impl DiskAPI for LocalDisk { return Err(DiskError::Unexpected); } + // The rollback backup stays where it is written (no rename) and is + // the sole restore source for a later undo_write, so it keeps + // SyncMode::FileAndDir: contents and directory entry both durable. if let Some(old_data_dir) = has_old_data_dir && let Some(dst_buf) = has_dst_buf.as_ref() && let Err(err) = self @@ -4140,7 +4192,7 @@ impl DiskAPI for LocalDisk { dst_volume, &format!("{}/{}/{}", &dst_path, &old_data_dir, STORAGE_FORMAT_FILE_BACKUP), dst_buf.clone().into(), - true, + SyncMode::FileAndDir, &skip_parent, ) .await @@ -5041,6 +5093,160 @@ mod test { ); } + #[tokio::test] + async fn test_write_all_meta_skips_tmp_parent_dir_fsync_but_fsyncs_dst_parent() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + assert!(drive_sync_enabled(), "test requires the default drive-sync configuration"); + + let bucket = "sync-meta-bucket"; + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + + let meta_path = format!("dir/object/{STORAGE_FORMAT_FILE}"); + disk.write_all_meta(bucket, &meta_path, b"payload", true) + .await + .expect("write_all_meta should succeed"); + + let dst_file_path = disk.get_object_path(bucket, &meta_path).expect("dst path should resolve"); + assert_eq!( + tokio::fs::read(&dst_file_path).await.expect("xl.meta should exist"), + b"payload", + "renamed xl.meta must carry the written contents" + ); + + let tmp_parent = disk + .get_bucket_path(RUSTFS_META_TMP_BUCKET) + .expect("tmp bucket path should resolve"); + assert!( + !os::fsync_dir_recorder::was_fsynced(&tmp_parent), + "tmp parent dir must not be fsynced for a write-then-rename tmp file" + ); + + let dst_parent = dst_file_path.parent().expect("dst file should have a parent").to_path_buf(); + assert!( + os::fsync_dir_recorder::was_fsynced(&dst_parent), + "destination parent dir must be fsynced after the commit rename" + ); + } + + #[tokio::test] + async fn test_write_all_public_still_fsyncs_parent_dir() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + assert!(drive_sync_enabled(), "test requires the default drive-sync configuration"); + + let bucket = "sync-public-bucket"; + ensure_test_volume(&disk, bucket).await; + + disk.write_all(bucket, "config/settings.json", Bytes::from_static(b"payload")) + .await + .expect("write_all should succeed"); + + let file_path = disk + .get_object_path(bucket, "config/settings.json") + .expect("file path should resolve"); + let parent = file_path.parent().expect("file should have a parent").to_path_buf(); + assert!( + os::fsync_dir_recorder::was_fsynced(&parent), + "direct (non-renamed) writes must keep fsyncing their parent dir" + ); + } + + #[tokio::test] + async fn test_rename_data_non_inline_skips_tmp_parent_dir_fsync() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + assert!(drive_sync_enabled(), "test requires the default drive-sync configuration"); + + let bucket = "sync-rename-bucket"; + let object = "dir/object"; + let tmp_object = "tmp-sync-write"; + let version_id = Uuid::parse_str("44444444-4444-4444-4444-444444444444").expect("version id should parse"); + let old_data_dir = Uuid::parse_str("55555555-5555-5555-5555-555555555555").expect("old data dir should parse"); + let new_data_dir = Uuid::parse_str("66666666-6666-6666-6666-666666666666").expect("new data dir should parse"); + + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + + let old_fi = test_file_info(object, version_id, Some(old_data_dir), None); + let dst_object_dir = dir.path().join(bucket).join(object); + fs::create_dir_all(dst_object_dir.join(old_data_dir.to_string())) + .await + .expect("old data dir should be created"); + fs::write(dst_object_dir.join(STORAGE_FORMAT_FILE), test_meta(old_fi)) + .await + .expect("old metadata should be written"); + + let tmp_data_dir = dir + .path() + .join(RUSTFS_META_TMP_BUCKET) + .join(tmp_object) + .join(new_data_dir.to_string()); + fs::create_dir_all(&tmp_data_dir) + .await + .expect("new tmp data dir should be created"); + fs::write(tmp_data_dir.join("part.1"), b"new-data") + .await + .expect("new tmp data should be written"); + + let new_fi = test_file_info(object, version_id, Some(new_data_dir), None); + disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object) + .await + .expect("rename_data should commit"); + + // The tmp xl.meta write point uses SyncMode::FileOnly: its parent dir + // ({tmp}/{tmp_object}) must not be fsynced. + let tmp_meta_parent = disk + .get_object_path(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{STORAGE_FORMAT_FILE}")) + .expect("tmp meta path should resolve") + .parent() + .expect("tmp meta should have a parent") + .to_path_buf(); + assert!( + !os::fsync_dir_recorder::was_fsynced(&tmp_meta_parent), + "tmp xl.meta parent dir must not be fsynced for the write-then-rename tmp file" + ); + + // The commit sequence itself is untouched: the destination parent dir + // is fsynced after the commit rename, ... + let dst_meta_parent = disk + .get_object_path(bucket, &format!("{object}/{STORAGE_FORMAT_FILE}")) + .expect("dst meta path should resolve") + .parent() + .expect("dst meta should have a parent") + .to_path_buf(); + assert!( + os::fsync_dir_recorder::was_fsynced(&dst_meta_parent), + "destination parent dir must be fsynced after the commit rename" + ); + + // ... and the rollback backup (which stays in place, no rename) still + // fsyncs its parent dir (SyncMode::FileAndDir). + let backup_parent = disk + .get_object_path(bucket, &format!("{object}/{old_data_dir}/{STORAGE_FORMAT_FILE_BACKUP}")) + .expect("backup path should resolve") + .parent() + .expect("backup should have a parent") + .to_path_buf(); + assert!( + os::fsync_dir_recorder::was_fsynced(&backup_parent), + "old-metadata rollback backup must keep fsyncing its parent dir" + ); + } + #[tokio::test] async fn test_rename_data_writes_old_metadata_backup_for_inline_overwrite() { use tempfile::tempdir; diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index b8b19541d..2ba2b4cb5 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -63,9 +63,33 @@ pub fn check_path_length(path_name: &str) -> Result<()> { Ok(()) } +/// Test-only recorder of every directory passed to [`fsync_dir_std`]. +/// +/// Durability regressions are invisible to ordinary behavior tests (the data +/// is on disk either way), so unit tests assert directly on which directories +/// were fsynced. Paths are recorded globally; tests must match on paths under +/// their own unique tempdir to stay robust against parallel test execution. +#[cfg(test)] +pub(crate) mod fsync_dir_recorder { + use std::path::{Path, PathBuf}; + use std::sync::Mutex; + + static RECORDED: Mutex> = Mutex::new(Vec::new()); + + pub(crate) fn record(dir: &Path) { + RECORDED.lock().expect("fsync dir recorder poisoned").push(dir.to_path_buf()); + } + + pub(crate) fn was_fsynced(dir: &Path) -> bool { + RECORDED.lock().expect("fsync dir recorder poisoned").iter().any(|p| p == dir) + } +} + /// 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<()> { + #[cfg(test)] + fsync_dir_recorder::record(dir.as_ref()); #[cfg(unix)] { std::fs::File::open(dir.as_ref())?.sync_all()?;