diff --git a/crates/storage/src/multipart.rs b/crates/storage/src/multipart.rs index 92d6377..ffb96c5 100644 --- a/crates/storage/src/multipart.rs +++ b/crates/storage/src/multipart.rs @@ -150,6 +150,9 @@ impl Storage { let final_path = self.part_path(upload_id, part_number); fs::rename(&tmp_path, &final_path).await?; + if let Some(p) = final_path.parent() { + crate::sync::sync_dir(p).await?; + } Ok(etag) } @@ -316,6 +319,9 @@ impl Storage { fs::create_dir_all(parent).await?; } fs::rename(&tmp_path, &final_path).await?; + if let Some(p) = final_path.parent() { + crate::sync::sync_dir(p).await?; + } // Best-effort cleanup. If this fails, gc_multipart_uploads will handle it. if let Err(e) = fs::remove_dir_all(&dir).await { diff --git a/crates/storage/src/read.rs b/crates/storage/src/read.rs index 4b4ba69..4cb0893 100644 --- a/crates/storage/src/read.rs +++ b/crates/storage/src/read.rs @@ -97,7 +97,12 @@ impl Storage { validate_key(key)?; let path = self.data_dir.join(bucket).join(key); match fs::remove_file(&path).await { - Ok(()) => Ok(()), + Ok(()) => { + if let Some(p) = path.parent() { + crate::sync::sync_dir(p).await?; + } + Ok(()) + } Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), Err(e) => Err(e.into()), } @@ -160,6 +165,9 @@ impl Storage { fs::create_dir_all(p).await?; } fs::rename(&tmp_path, &dst).await?; + if let Some(p) = dst.parent() { + crate::sync::sync_dir(p).await?; + } Ok(attrs.etag) } @@ -190,6 +198,9 @@ impl Storage { fs::create_dir_all(p).await?; } fs::rename(&src, &dst).await?; + if let Some(p) = dst.parent() { + crate::sync::sync_dir(p).await?; + } Ok(()) } } diff --git a/crates/storage/src/sync.rs b/crates/storage/src/sync.rs index ce546a4..1fa7d91 100644 --- a/crates/storage/src/sync.rs +++ b/crates/storage/src/sync.rs @@ -1,4 +1,7 @@ use std::io; +use std::path::Path; +#[cfg(test)] +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use futures::future::join_all; @@ -6,6 +9,24 @@ use tokio::fs; use tokio::sync::{mpsc, oneshot}; use tokio::time::timeout_at; +#[cfg(test)] +pub(crate) static DIR_FSYNCS: AtomicUsize = AtomicUsize::new(0); + +/// fsync a directory so a rename/unlink dirent is durable (Pedra F17). +pub(crate) fn sync_dir_blocking(dir: &Path) -> io::Result<()> { + #[cfg(test)] + DIR_FSYNCS.fetch_add(1, Ordering::SeqCst); + let f = std::fs::File::open(dir)?; + f.sync_all() +} + +pub(crate) async fn sync_dir(dir: &Path) -> io::Result<()> { + let d = dir.to_path_buf(); + tokio::task::spawn_blocking(move || sync_dir_blocking(&d)) + .await + .unwrap_or_else(|e| Err(io::Error::other(e))) +} + pub(crate) struct SyncRequest { file: fs::File, tx: oneshot::Sender>, diff --git a/crates/storage/src/write.rs b/crates/storage/src/write.rs index c2df72e..30441f4 100644 --- a/crates/storage/src/write.rs +++ b/crates/storage/src/write.rs @@ -168,7 +168,11 @@ impl Storage { if cond.is_some() { commit_locked(&tmp, &dest, cond, bucket_owned, key_owned) } else { - std::fs::rename(&tmp, &dest).map_err(StorageError::Io) + std::fs::rename(&tmp, &dest).map_err(StorageError::Io)?; + if let Some(p) = dest.parent() { + crate::sync::sync_dir_blocking(p).map_err(StorageError::Io)?; + } + Ok(()) } }) .await @@ -261,6 +265,9 @@ fn commit_locked( } std::fs::rename(tmp, dest).map_err(StorageError::Io)?; + if let Some(p) = dest.parent() { + crate::sync::sync_dir_blocking(p).map_err(StorageError::Io)?; + } let _ = std::fs::remove_file(&lock_path); Ok(()) } @@ -345,6 +352,25 @@ mod tests { assert_eq!(info.etag, etag); } + #[tokio::test] + async fn write_object_fsyncs_the_parent_directory() { + crate::sync::DIR_FSYNCS.store(0, std::sync::atomic::Ordering::SeqCst); + let (s, _dir) = make_storage().await; + s.write_object( + "bucket", + "hello.txt", + Cursor::new(b"hello"), + ObjectMeta::default(), + None, + ) + .await + .unwrap(); + assert!( + crate::sync::DIR_FSYNCS.load(std::sync::atomic::Ordering::SeqCst) > 0, + "PUT rename must fsync the parent directory" + ); + } + #[tokio::test] async fn if_none_match_blocks_overwrite() { let (s, _dir) = make_storage().await;