From c7c31dc9c9ca583b846352a33f9083a2a3f626a2 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 3 Nov 2026 14:37:19 +0100 Subject: [PATCH 1/6] fix(local-fs): Remove empty directories after deleting objects --- objectstore-service/src/backend/local_fs.rs | 131 ++++++++++++++++---- 1 file changed, 108 insertions(+), 23 deletions(-) diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index c853a4da..7fec2596 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/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 [`run_in_directory`] creates the directory 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,73 @@ 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)? +} + +/// Creates `dir` and runs `f` to place a new entry in it. +/// +/// [`remove_empty_directories`] can remove `dir` after it is created but before `f` places its +/// entry. When `f` fails with [`io::ErrorKind::NotFound`], the directory is created again and `f` +/// retried, up to [`MAX_DIRECTORY_ATTEMPTS`] times. +fn run_in_directory(dir: &Path, mut f: impl FnMut() -> io::Result) -> io::Result { + let mut attempt = 1; + loop { + directory_builder().create(dir)?; + match f() { + Err(error) if error.kind() == io::ErrorKind::NotFound => { + if attempt == MAX_DIRECTORY_ATTEMPTS { + return Err(error); + } + attempt += 1; + } + result => return result, + } + } +} + +/// Removes the directories of a deleted 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 [`run_in_directory`]. +async fn remove_empty_directories(root: &Path, object_path: &Path) -> Result<()> { + let root = root.to_path_buf(); + let object_path = object_path.to_path_buf(); + let task = tokio::task::spawn_blocking(move || { + object_path + .ancestors() + .skip(1) + .take_while(|dir| *dir != root && dir.starts_with(&root)) + .try_for_each(std::fs::remove_dir) + }); + + let result = task.await.context( + ErrorKind::Internal, + "waiting to remove empty local-fs directories", + )?; + + if let Err(error) = result { + objectstore_log::debug!(!!&error, "Stopped removing empty local-fs directories"); + } + + Ok(()) } /// Configuration for [`LocalFsBackend`]. @@ -275,10 +342,6 @@ 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 reader = pin!(StreamReader::new(stream)); @@ -404,12 +467,16 @@ impl Backend for LocalFsBackend { id: &ObjectId, _access_time: Timestamp, ) -> Result { - 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"); } @@ -482,12 +549,6 @@ 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?; self.change_stream .commit_write(&session.object_id, stored_size, expires_at) @@ -705,10 +766,27 @@ impl UploadFile { .. } = self; drop(file); - tokio::fs::rename(path, target).await.context( - ErrorKind::BackendFailure, - "publishing local-fs resumable upload", - )?; + + let task = tokio::task::spawn_blocking(move || { + let rename = || std::fs::rename(&path, &target); + // A concurrent delete can remove the empty target directory before the rename. + if let Some(parent) = target.parent() { + run_in_directory(parent, rename) + } else { + rename() + } + }); + + task.await + .context( + ErrorKind::Internal, + "waiting to publish local-fs resumable upload", + )? + .context( + ErrorKind::BackendFailure, + "publishing local-fs resumable upload", + )?; + Ok((stored_size, expires_at)) } } @@ -871,7 +949,7 @@ fn create_tempfile(parent: &Path) -> io::Result { let mut builder = tempfile::Builder::new(); builder.suffix(".draft"); - builder.tempfile_in(parent) + run_in_directory(parent, || builder.tempfile_in(parent)) } #[cfg(test)] @@ -1331,6 +1409,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] From f5f0d24fbd40f1dd88fe416da4c33ee6d8be0c15 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 3 Nov 2026 15:32:31 +0100 Subject: [PATCH 2/6] fix(local-fs): Retry when directory creation races with cleanup --- objectstore-service/src/backend/local_fs.rs | 23 +++++++++------------ 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index 7fec2596..e8f7e9bc 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -193,23 +193,20 @@ async fn create_directories(path: &Path) -> io::Result<()> { /// Creates `dir` and runs `f` to place a new entry in it. /// -/// [`remove_empty_directories`] can remove `dir` after it is created but before `f` places its -/// entry. When `f` fails with [`io::ErrorKind::NotFound`], the directory is created again and `f` -/// retried, up to [`MAX_DIRECTORY_ATTEMPTS`] times. +/// [`remove_empty_directories`] can remove `dir` or one of its parents 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. fn run_in_directory(dir: &Path, mut f: impl FnMut() -> io::Result) -> io::Result { - let mut attempt = 1; - loop { - directory_builder().create(dir)?; - match f() { - Err(error) if error.kind() == io::ErrorKind::NotFound => { - if attempt == MAX_DIRECTORY_ATTEMPTS { - return Err(error); - } - attempt += 1; - } + let mut attempt = || directory_builder().create(dir).and_then(|()| f()); + + for _ in 1..MAX_DIRECTORY_ATTEMPTS { + match attempt() { + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, result => return result, } } + + attempt() } /// Removes the directories of a deleted object below `root` as long as they are empty. From 8faf2f362b5ee555888472ca6b25696b0fcbe786 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 3 Nov 2026 17:51:46 +0100 Subject: [PATCH 3/6] ref(local-fs): Explain the final directory attempt --- objectstore-service/src/backend/local_fs.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index e8f7e9bc..82458a32 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -199,6 +199,7 @@ async fn create_directories(path: &Path) -> io::Result<()> { fn run_in_directory(dir: &Path, mut f: impl FnMut() -> io::Result) -> io::Result { let mut attempt = || directory_builder().create(dir).and_then(|()| f()); + // Retry on NotFound; the final attempt below returns its result as is, including NotFound. for _ in 1..MAX_DIRECTORY_ATTEMPTS { match attempt() { Err(error) if error.kind() == io::ErrorKind::NotFound => continue, From 7afcc5391ee8f0871a04d08969b116dccb93ee67 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 3 Nov 2026 17:52:38 +0100 Subject: [PATCH 4/6] ref(local-fs): Explain why directory creation is retried --- objectstore-service/src/backend/local_fs.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index 82458a32..9b1a4216 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -199,7 +199,7 @@ async fn create_directories(path: &Path) -> io::Result<()> { fn run_in_directory(dir: &Path, mut f: impl FnMut() -> io::Result) -> io::Result { let mut attempt = || directory_builder().create(dir).and_then(|()| f()); - // Retry on NotFound; the final attempt below returns its result as is, including NotFound. + // 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, From c19dfc42ea3eb99e979033a48626f84ce409c256 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 3 Nov 2026 18:30:02 +0100 Subject: [PATCH 5/6] fix(local-fs): Remove directories of failed writes --- objectstore-service/src/backend/local_fs.rs | 191 +++++++++++++------- 1 file changed, 122 insertions(+), 69 deletions(-) diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index 9b1a4216..38b82ddc 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -101,7 +101,7 @@ use crate::id::ObjectId; use crate::resumable::{BackendToken, Session}; use crate::stream::{self, ClientStream}; -/// Number of times [`run_in_directory`] creates the directory before giving up. +/// 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. @@ -191,51 +191,33 @@ async fn create_directories(path: &Path) -> io::Result<()> { .map_err(io::Error::other)? } -/// Creates `dir` and runs `f` to place a new entry in it. -/// -/// [`remove_empty_directories`] can remove `dir` or one of its parents 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. -fn run_in_directory(dir: &Path, mut f: impl FnMut() -> io::Result) -> io::Result { - let mut attempt = || directory_builder().create(dir).and_then(|()| 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() -} - /// Removes the directories of a deleted 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 [`run_in_directory`]. -async fn remove_empty_directories(root: &Path, object_path: &Path) -> Result<()> { +/// 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 || { - object_path - .ancestors() - .skip(1) - .take_while(|dir| *dir != root && dir.starts_with(&root)) - .try_for_each(std::fs::remove_dir) + remove_empty_directories_sync(&root, &object_path); }); - let result = task.await.context( - ErrorKind::Internal, - "waiting to remove empty local-fs directories", - )?; - - if let Err(error) = result { - objectstore_log::debug!(!!&error, "Stopped removing empty local-fs directories"); - } + // A leftover empty directory is harmless, so a failed cleanup must not fail the delete. + let _ = task.await; +} - Ok(()) +/// 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`]. @@ -341,7 +323,7 @@ impl Backend for LocalFsBackend { .begin_write(id, metadata.time_expires) .await?; - 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 @@ -445,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", @@ -473,7 +455,7 @@ impl Backend for LocalFsBackend { Ok(()) => { self.change_stream.commit_delete(id).await; drop(guard); - remove_empty_directories(&self.path, &path).await?; + remove_empty_directories(&self.path, &path).await; } Err(error) if error.kind() == io::ErrorKind::NotFound => { objectstore_log::debug!("Object not found"); @@ -546,8 +528,8 @@ impl Backend for LocalFsBackend { }); } - let object_path = self.path(&session.object_id); - 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; @@ -755,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)> { + async fn publish(self, mut directory: DraftDirectory) -> Result<(u64, Option)> { let Self { file, path, @@ -765,26 +747,16 @@ impl UploadFile { } = self; drop(file); - let task = tokio::task::spawn_blocking(move || { - let rename = || std::fs::rename(&path, &target); - // A concurrent delete can remove the empty target directory before the rename. - if let Some(parent) = target.parent() { - run_in_directory(parent, rename) - } else { - rename() - } - }); - - task.await - .context( - ErrorKind::Internal, - "waiting to publish 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)) } } @@ -868,23 +840,95 @@ 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(&self, mut f: F) -> io::Result + where + T: Send + 'static, + F: FnMut() -> io::Result + 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 { + remove_empty_directories_sync(&self.root, &self.object_path); + } + } +} + struct Draft { writer: BufWriter, 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 { + async fn create(root: &Path, target: &Path, metadata: &Metadata) -> Result { + 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(); @@ -896,6 +940,7 @@ impl Draft { path, target: target.to_path_buf(), preamble_len, + directory, }) } @@ -926,12 +971,17 @@ 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", @@ -939,7 +989,10 @@ impl Draft { .context( ErrorKind::BackendFailure, "publishing local-fs object draft", - ) + )?; + + directory.keep(); + Ok(()) } } @@ -947,7 +1000,7 @@ fn create_tempfile(parent: &Path) -> io::Result { let mut builder = tempfile::Builder::new(); builder.suffix(".draft"); - run_in_directory(parent, || builder.tempfile_in(parent)) + builder.tempfile_in(parent) } #[cfg(test)] From 13a2be5ff97950ed626a1635cf6cfb9925fecf45 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Thu, 5 Nov 2026 15:44:28 +0100 Subject: [PATCH 6/6] ref(local-fs): Remove draft directories on the blocking pool Dropping an unpublished draft removed its empty directories synchronously, usually on an async worker. Inside a Tokio runtime, the removal now runs on the blocking thread pool. Without a runtime, it still runs in place. --- objectstore-service/src/backend/local_fs.rs | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index 38b82ddc..e500a214 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -906,8 +906,19 @@ impl DraftDirectory { impl Drop for DraftDirectory { fn drop(&mut self) { - if !self.kept { - remove_empty_directories_sync(&self.root, &self.object_path); + 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(), } } }