Implement bitcache compact.

Author: artob

rust/lib/bitcache-core/src/repository.rs

@@ -160,6 +160,15 @@ pub trait Repository: Send + Sync {
         async { Ok(false) }
     }
 
+    /// Performs backend-specific repository maintenance.
+    ///
+    /// The default implementation is a no-op. Backends may override this to
+    /// compact or otherwise optimize their physical storage without changing
+    /// the repository's logical contents.
+    fn compact(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
+        async { Ok(()) }
+    }
+
     /// Removes all blobs, resetting the repository to an empty state.
     fn clear(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send;
 

rust/lib/bitcache-fs/src/repository.rs

@@ -417,6 +417,76 @@ impl FsRepository {
         }
     }
 
+    /// Rewrites an uncompressed blob using maximum XZ compression.
+    #[cfg(feature = "tokio")]
+    async fn compact_blob(&self, id: &Id) -> Result<(), RepositoryError> {
+        use async_compression::{Level, tokio::write::XzEncoder};
+        use bitcache_core::tokio::{
+            fs::File,
+            io::{AsyncWriteExt, copy},
+        };
+
+        let source_name = Self::path(id);
+        let source_file = match self.0.open(&source_name) {
+            Ok(file) => file,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+            Err(error) => return Err(error.into()),
+        };
+        let extended = self.read_extended(&source_name, &source_file)?;
+        let mut source_file = File::from_std(source_file.into_std());
+        let (temp_name, temp_file) = self.create_temp_file()?;
+        let temp_file = File::from_std(temp_file.into_std());
+        let mut encoder = XzEncoder::with_quality(temp_file, Level::Best);
+
+        let result: Result<(), RepositoryError> = async {
+            copy(&mut source_file, &mut encoder).await?;
+            encoder.shutdown().await?;
+            Ok(())
+        }
+        .await;
+        drop(encoder);
+        drop(source_file);
+
+        if let Err(error) = result {
+            let _ = self.0.remove_file(&temp_name);
+            return Err(error);
+        }
+
+        let temp_file = match self.0.open(&temp_name) {
+            Ok(file) => file,
+            Err(error) => {
+                let _ = self.0.remove_file(&temp_name);
+                return Err(error.into());
+            },
+        };
+        if let Err(error) = self.prepare_temp(&temp_name, &temp_file, &extended) {
+            drop(temp_file);
+            let _ = self.0.remove_file(&temp_name);
+            return Err(error);
+        }
+        drop(temp_file);
+
+        let compressed_name = Self::compressed_path(id);
+        match self.0.remove_file(&compressed_name) {
+            Ok(()) => (),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => (),
+            Err(error) => {
+                let _ = self.0.remove_file(&temp_name);
+                return Err(error.into());
+            },
+        }
+        if let Err(error) = self.0.hard_link(&temp_name, &self.0, &compressed_name) {
+            let _ = self.0.remove_file(&temp_name);
+            return Err(error.into());
+        }
+        self.0.remove_file(&temp_name)?;
+        match self.0.remove_file(source_name) {
+            Ok(()) => Ok(()),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+            Err(error) => Err(error.into()),
+        }
+    }
+
     /// Collects all physical blob IDs, including expired blobs.
     fn collect_physical_ids(&self) -> Result<Vec<Id>, RepositoryError> {
         let mut ids = Vec::new();
@@ -758,6 +828,19 @@ impl Repository for FsRepository {
         self.write_existing_extended(&blob.name, &blob.file, &blob.extended)
     }
 
+    async fn compact(&mut self) -> Result<(), Self::Error> {
+        #[cfg(feature = "tokio")]
+        {
+            for id in self.collect_physical_ids()? {
+                self.compact_blob(&id).await?;
+            }
+            Ok(())
+        }
+
+        #[cfg(not(feature = "tokio"))]
+        Ok(())
+    }
+
     async fn clear(&mut self) -> Result<(), Self::Error> {
         for id in self.collect_physical_ids()? {
             for encoding in [BlobEncoding::Uncompressed, BlobEncoding::Xz] {

rust/lib/bitcache-fs/tests/repository.rs

@@ -249,6 +249,54 @@ async fn test_compressed_storage_listing_and_uncompressed_preference() {
     assert!(!compressed_path.exists());
 }
 
+#[tokio::test]
+async fn test_compact_compresses_uncompressed_blobs() {
+    let temp_dir = TestDir::new();
+    let repository_path = temp_dir.path().join("repository");
+    let mut repository = FsRepository::create(repository_path.to_str().unwrap()).unwrap();
+    let data = b"legacy uncompressed blob\n".repeat(1_024);
+    let id = Id::of(&data);
+    let uncompressed_path = uncompressed_blob_path(&repository_path, &id);
+    let compressed_path = blob_path(&repository_path, &id);
+    fs::write(&uncompressed_path, &data).unwrap();
+    set_xattr(
+        &uncompressed_path,
+        "user.bitcache.media-type",
+        b"text/plain",
+    );
+
+    repository.compact().await.unwrap();
+
+    assert!(!uncompressed_path.exists());
+    assert!(compressed_path.exists());
+    let compacted_metadata = fs::metadata(&compressed_path).unwrap();
+    assert_eq!(compacted_metadata.permissions().mode() & 0o777, 0o444);
+    assert_eq!(read_blob_file(&repository, &id).await, data);
+    assert_eq!(
+        repository
+            .get(&id)
+            .await
+            .unwrap()
+            .unwrap()
+            .metadata()
+            .media_type(),
+        Some("text/plain")
+    );
+    assert!(fs::read_dir(&repository_path).unwrap().all(|entry| {
+        !entry
+            .unwrap()
+            .file_name()
+            .to_string_lossy()
+            .starts_with(".put-")
+    }));
+
+    repository.compact().await.unwrap();
+    assert_eq!(
+        fs::metadata(&compressed_path).unwrap().ino(),
+        compacted_metadata.ino()
+    );
+}
+
 #[tokio::test]
 async fn test_fs_repository_expiry_and_media_type_round_trip() {
     let temp_dir = TestDir::new();

rust/lib/bitcache/src/main.rs

@@ -134,6 +134,12 @@ enum Command {
         ids: Vec<Id>,
     },
 
+    /// Compact the repository's physical storage.
+    ///
+    /// Filesystem repositories rewrite uncompressed blobs using maximum XZ
+    /// compression. Other repository backends may perform no maintenance.
+    Compact {},
+
     /// Remove all blobs from the repository.
     ///
     /// As a safety measure, this requires the `--force` flag; without it,
@@ -474,6 +480,12 @@ pub async fn main() -> Result<(), SysexitsError> {
             Ok(())
         },
 
+        Command::Compact {} => {
+            let mut repository = bitcache::open_env("BITCACHE_URL", "file:.bitcache").await?;
+            repository.compact().await?;
+            Ok(())
+        },
+
         Command::Clear { force } => {
             if !force {
                 eprintln!("bitcache: refusing to clear the repository without --force");