fix(local-fs): Remove empty directories after deleting objects

objectstore-service/src/backend/local_fs.rs

@@ -44,6 +44,10 @@
 //! observe either the previous or next complete file. Unpublished drafts are removed automatically
 //! when dropped.
 //!
+//! Deleting an object also removes its directories while they are empty. This is not synchronized
+//! with writers: directory removal only succeeds on empty directories, and writers recreate a
+//! directory removed before they placed their file in it.
+//!
 //! To avoid races on metadata, expiry, and upload updates, this backend uses locks placed under
 //! `.locks/` to synchronize mutations across backend instances and cooperating processes. The first
 //! two bytes of the BLAKE3 hash of an object's storage path select one of 65,536 permanent lock
@@ -97,6 +101,9 @@ use crate::id::ObjectId;
 use crate::resumable::{BackendToken, Session};
 use crate::stream::{self, ClientStream};
 
+/// Number of times [`DraftDirectory::run`] creates the directories before giving up.
+const MAX_DIRECTORY_ATTEMPTS: usize = 3;
+
 /// Options for owner-only files on Unix, further restricted by umask.
 fn file_options() -> std::fs::OpenOptions {
     let mut options = std::fs::OpenOptions::new();
@@ -164,13 +171,53 @@ fn encode_key(key: &str) -> String {
     encoded
 }
 
-/// Creates owner-only directories on Unix without changing existing permissions.
-async fn create_directories(path: &Path) -> io::Result<()> {
-    let mut builder = tokio::fs::DirBuilder::new();
+/// Builder for owner-only directories on Unix that leaves existing permissions unchanged.
+fn directory_builder() -> std::fs::DirBuilder {
+    let mut builder = std::fs::DirBuilder::new();
     builder.recursive(true);
     #[cfg(unix)]
-    builder.mode(0o700);
-    builder.create(path).await
+    {
+        use std::os::unix::fs::DirBuilderExt as _;
+        builder.mode(0o700);
+    }
+    builder
+}
+
+/// Creates owner-only directories on Unix without changing existing permissions.
+async fn create_directories(path: &Path) -> io::Result<()> {
+    let path = path.to_path_buf();
+    tokio::task::spawn_blocking(move || directory_builder().create(path))
+        .await
+        .map_err(io::Error::other)?
+}
+
+/// Removes the directories of a deleted object below `root` as long as they are empty.
+///
+/// Runs [`remove_empty_directories_sync`] on the blocking thread pool.
+async fn remove_empty_directories(root: &Path, object_path: &Path) {
+    let root = root.to_path_buf();
+    let object_path = object_path.to_path_buf();
+    let task = tokio::task::spawn_blocking(move || {
+        remove_empty_directories_sync(&root, &object_path);
+    });
+
+    // A leftover empty directory is harmless, so a failed cleanup must not fail the delete.
+    let _ = task.await;
+}
+
+/// Removes the directories of an object below `root` as long as they are empty.
+///
+/// Stops at the first directory that cannot be removed, typically because it still holds other
+/// objects. Removal is atomic and only succeeds on empty directories, so this never races with
+/// stored objects. Writers handle directories removed under them with [`DraftDirectory::run`].
+fn remove_empty_directories_sync(root: &Path, object_path: &Path) {
+    let mut object_dirs = object_path
+        .ancestors()
+        .skip(1)
+        .take_while(|dir| *dir != root && dir.starts_with(root));
+
+    // A leftover empty directory is harmless, so cleanup ignores failures.
+    let _ = object_dirs.try_for_each(std::fs::remove_dir);
 }
 
 /// Configuration for [`LocalFsBackend`].
@@ -275,12 +322,8 @@ impl Backend for LocalFsBackend {
         self.change_stream
             .begin_write(id, metadata.time_expires)
             .await?;
-        create_directories(path.parent().unwrap()).await.context(
-            ErrorKind::BackendFailure,
-            "creating local-fs object directory",
-        )?;
 
-        let mut draft = Draft::create(&path, metadata).await?;
+        let mut draft = Draft::create(&self.path, &path, metadata).await?;
         let mut reader = pin!(StreamReader::new(stream));
         let payload_size = tokio::io::copy(&mut reader, draft.writer())
             .await
@@ -384,7 +427,7 @@ impl Backend for LocalFsBackend {
         metadata.time_expires = Some(expire_at);
         self.change_stream.begin_update(id, Some(expire_at)).await?;
 
-        let mut draft = Draft::create(&path, &metadata).await?;
+        let mut draft = Draft::create(&self.path, &path, &metadata).await?;
         tokio::io::copy(&mut reader, draft.writer()).await.context(
             ErrorKind::BackendFailure,
             "copying local-fs object payload for expiry extension",
@@ -404,12 +447,16 @@ impl Backend for LocalFsBackend {
         id: &ObjectId,
         _access_time: Timestamp,
     ) -> Result<DeleteResponse> {
-        let _guard = self.locks.acquire(id).await?;
+        let guard = self.locks.acquire(id).await?;
 
         objectstore_log::debug!("Deleting from local_fs backend");
         let path = self.path(id);
-        match tokio::fs::remove_file(path).await {
-            Ok(()) => self.change_stream.commit_delete(id).await,
+        match tokio::fs::remove_file(&path).await {
+            Ok(()) => {
+                self.change_stream.commit_delete(id).await;
+                drop(guard);
+                remove_empty_directories(&self.path, &path).await;
+            }
             Err(error) if error.kind() == io::ErrorKind::NotFound => {
                 objectstore_log::debug!("Object not found");
             }
@@ -481,14 +528,8 @@ impl Backend for LocalFsBackend {
             });
         }
 
-        let object_path = self.path(&session.object_id);
-        create_directories(object_path.parent().unwrap())
-            .await
-            .context(
-                ErrorKind::BackendFailure,
-                "creating local-fs object directory",
-            )?;
-        let (stored_size, expires_at) = upload.publish(object_path).await?;
+        let directory = DraftDirectory::new(&self.path, &self.path(&session.object_id));
+        let (stored_size, expires_at) = upload.publish(directory).await?;
         self.change_stream
             .commit_write(&session.object_id, stored_size, expires_at)
             .await;
@@ -696,7 +737,7 @@ impl UploadFile {
     }
 
     /// Publishes the upload; requires a successful `append()` with no subsequent writes.
-    async fn publish(self, target: PathBuf) -> Result<(u64, Option<Timestamp>)> {
+    async fn publish(self, mut directory: DraftDirectory) -> Result<(u64, Option<Timestamp>)> {
         let Self {
             file,
             path,
@@ -705,10 +746,17 @@ impl UploadFile {
             ..
         } = self;
         drop(file);
-        tokio::fs::rename(path, target).await.context(
-            ErrorKind::BackendFailure,
-            "publishing local-fs resumable upload",
-        )?;
+
+        let target = directory.object_path().to_path_buf();
+        directory
+            .run(move || std::fs::rename(&path, &target))
+            .await
+            .context(
+                ErrorKind::BackendFailure,
+                "publishing local-fs resumable upload",
+            )?;
+
+        directory.keep();
         Ok((stored_size, expires_at))
     }
 }
@@ -792,23 +840,106 @@ impl ObjectFile {
     }
 }
 
+/// The directories of an object that is not published yet.
+///
+/// Writers create the directories and place their entry with [`run`](Self::run). When dropped
+/// before [`keep`](Self::keep), for example after a failed or cancelled write, this removes the
+/// directories again as long as they are empty.
+struct DraftDirectory {
+    root: PathBuf,
+    object_path: PathBuf,
+    kept: bool,
+}
+
+impl DraftDirectory {
+    fn new(root: &Path, object_path: &Path) -> Self {
+        Self {
+            root: root.to_path_buf(),
+            object_path: object_path.to_path_buf(),
+            kept: false,
+        }
+    }
+
+    /// Returns the path the object is published at.
+    fn object_path(&self) -> &Path {
+        &self.object_path
+    }
+
+    /// Creates the directories and runs `f` on the blocking thread pool to place a new entry in them.
+    ///
+    /// [`remove_empty_directories`] can remove the directories while they are being created, or
+    /// before `f` places its entry. When either step fails with [`io::ErrorKind::NotFound`], both
+    /// are retried, up to [`MAX_DIRECTORY_ATTEMPTS`] times in total.
+    async fn run<T, F>(&self, mut f: F) -> io::Result<T>
+    where
+        T: Send + 'static,
+        F: FnMut() -> io::Result<T> + Send + 'static,
+    {
+        let dir = self.object_path.parent().map(Path::to_path_buf);
+        let task = tokio::task::spawn_blocking(move || {
+            let mut attempt = || {
+                if let Some(dir) = &dir {
+                    directory_builder().create(dir)?;
+                }
+                f()
+            };
+
+            // A concurrent delete can remove the empty directory before `f` places its entry in it.
+            for _ in 1..MAX_DIRECTORY_ATTEMPTS {
+                match attempt() {
+                    Err(error) if error.kind() == io::ErrorKind::NotFound => continue,
+                    result => return result,
+                }
+            }
+
+            attempt()
+        });
+
+        task.await.map_err(io::Error::other)?
+    }
+
+    /// Keeps the directories, since the published object now lives in them.
+    fn keep(&mut self) {
+        self.kept = true;
+    }
+}
+
+impl Drop for DraftDirectory {
+    fn drop(&mut self) {
+        if self.kept {
+            return;
+        }
+
+        let root = std::mem::take(&mut self.root);
+        let object_path = std::mem::take(&mut self.object_path);
+        let cleanup = move || remove_empty_directories_sync(&root, &object_path);
+
+        // Drafts are usually dropped on an async worker, which must not block on filesystem calls.
+        // Without a runtime, there is no blocking pool, so clean up in place.
+        match tokio::runtime::Handle::try_current() {
+            Ok(runtime) => drop(runtime.spawn_blocking(cleanup)),
+            Err(_) => cleanup(),
+        }
+    }
+}
+
 struct Draft {
     writer: BufWriter<tokio::fs::File>,
     path: tempfile::TempPath,
     target: PathBuf,
     /// Bytes the metadata preamble occupies, counted as it is written.
     preamble_len: u64,
+    /// Declared after `path`, so an unpublished draft's file is removed before its directories.
+    directory: DraftDirectory,
 }
 
 impl Draft {
-    async fn create(target: &Path, metadata: &Metadata) -> Result<Self> {
+    async fn create(root: &Path, target: &Path, metadata: &Metadata) -> Result<Self> {
+        let directory = DraftDirectory::new(root, target);
         let parent = target.parent().unwrap().to_path_buf();
-        let tempfile = tokio::task::spawn_blocking(move || create_tempfile(&parent))
+        let tempfile = directory
+            .run(move || create_tempfile(&parent))
             .await
-            .context(
-                ErrorKind::Internal,
-                "waiting for local-fs object draft creation",
-            )?
             .context(ErrorKind::BackendFailure, "creating local-fs object draft")?;
         let (file, path) = tempfile.into_parts();
 
@@ -820,6 +951,7 @@ impl Draft {
             path,
             target: target.to_path_buf(),
             preamble_len,
+            directory,
         })
     }
 
@@ -850,20 +982,28 @@ impl Draft {
             writer,
             path,
             target,
+            mut directory,
             ..
         } = self;
 
         drop(writer);
-        tokio::task::spawn_blocking(move || path.persist(target))
-            .await
+        let task = tokio::task::spawn_blocking(move || {
+            // The error contains the draft file. Drop it first so the directory gets cleaned up.
+            path.persist(target).map_err(|error| error.error)
+        });
+
+        task.await
             .context(
                 ErrorKind::Internal,
                 "waiting to publish local-fs object draft",
             )?
             .context(
                 ErrorKind::BackendFailure,
                 "publishing local-fs object draft",
-            )
+            )?;
+
+        directory.keep();
+        Ok(())
     }
 }
 
@@ -1331,6 +1471,13 @@ mod tests {
             .unwrap()
             .unwrap();
         assert_eq!(stream::read_to_vec(payload).await.unwrap(), b"a/b");
+
+        // Deleting the remaining objects removes their empty directories up to the root.
+        for id in [&child, &suffixed] {
+            backend.delete_object(id, Timestamp::now()).await.unwrap();
+        }
+        assert!(!backend.path.join("testing").exists());
+        assert!(backend.path.exists());
     }
 
     #[test]