diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index c853a4da..e500a214 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 [`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 { - 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)> { + async fn publish(self, mut directory: DraftDirectory) -> Result<(u64, Option)> { 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(&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 { + 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, 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(); @@ -820,6 +951,7 @@ impl Draft { path, target: target.to_path_buf(), preamble_len, + directory, }) } @@ -850,12 +982,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", @@ -863,7 +1000,10 @@ impl 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]