Fsync the parent directory after object rename and delete

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 {

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

crates/storage/src/sync.rs

@@ -1,11 +1,32 @@
 use std::io;
+use std::path::Path;
+#[cfg(test)]
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 
 use futures::future::join_all;
 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<io::Result<()>>,

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;