Merge branch 'vacuum'

Author: madadam

CHANGELOG.md

@@ -5,7 +5,9 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-<!-- ## [Unreleased](https://github.com/equalitie/ouisync/compare/v0.10.0...master) -->
+## [Unreleased](https://github.com/equalitie/ouisync/compare/v0.10.0...master)
+
+- Fix repository size not decreasing after deleting files
 
 ## [v0.10.0](https://github.com/equalitie/ouisync/compare/v0.9.3...v0.10.0) - 2025-12-29
 

lib/src/db/mod.rs

@@ -13,10 +13,10 @@ use tracing::Span;
 use deadlock::ExpectShortLifetime;
 use ref_cast::RefCast;
 use sqlx::{
-    Row, SqlitePool, TransactionManager,
+    ConnectOptions, Row, SqlitePool, TransactionManager,
     sqlite::{
-        Sqlite, SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous,
-        SqliteTransactionManager,
+        Sqlite, SqliteAutoVacuum, SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions,
+        SqliteSynchronous, SqliteTransactionManager,
     },
 };
 use std::{
@@ -48,6 +48,20 @@ pub struct Pool {
 
 impl Pool {
     async fn create(conn_options: SqliteConnectOptions) -> Result<Self, sqlx::Error> {
+        if fs::try_exists(conn_options.get_filename())
+            .await
+            .unwrap_or(false)
+        {
+            // Try to enable auto-vacuum, but if it fails [^1] it's not a critical failure as we can
+            // keep using the db without auto-vacuum [^2]. Just log the error and keep going.
+            //
+            // [^1]: for example, because of low disk space
+            // [^2]: with the downside that disk space won't be reclaimed after data deletion
+            if let Err(error) = enable_auto_vacuum(conn_options.get_filename()).await {
+                tracing::warn!(?error, "failed to enable auto vacuum");
+            }
+        }
+
         let conn_options = conn_options
             .journal_mode(SqliteJournalMode::Wal)
             .synchronous(SqliteSynchronous::Normal)
@@ -63,7 +77,12 @@ impl Pool {
         let write = pool_options
             .clone()
             .max_connections(1)
-            .connect_with(conn_options.clone().optimize_on_close(true, Some(1000)))
+            .connect_with(
+                conn_options
+                    .clone()
+                    .auto_vacuum(SqliteAutoVacuum::Full)
+                    .optimize_on_close(true, Some(1000)),
+            )
             .await?;
 
         let reads = pool_options
@@ -316,6 +335,18 @@ impl std::fmt::Debug for WriteTransaction {
 
 impl_executor_by_deref!(WriteTransaction);
 
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("failed to create database directory")]
+    CreateDirectory(#[source] io::Error),
+    #[error("database already exists")]
+    Exists,
+    #[error("failed to open database")]
+    Open(#[source] sqlx::Error),
+    #[error("failed to execute database query")]
+    Query(#[from] sqlx::Error),
+}
+
 /// Creates a new database and opens a connection to it.
 pub(crate) async fn create(path: impl AsRef<Path>) -> Result<Pool, Error> {
     let path = path.as_ref();
@@ -386,16 +417,30 @@ pub(crate) const fn encode_u64(u: u64) -> i64 {
     u as i64
 }
 
-#[derive(Debug, Error)]
-pub enum Error {
-    #[error("failed to create database directory")]
-    CreateDirectory(#[source] io::Error),
-    #[error("database already exists")]
-    Exists,
-    #[error("failed to open database")]
-    Open(#[source] sqlx::Error),
-    #[error("failed to execute database query")]
-    Query(#[from] sqlx::Error),
+// Enable auto-vacuum on the given database unless already enabled
+async fn enable_auto_vacuum(db_path: &Path) -> Result<(), Error> {
+    let mut conn = SqliteConnectOptions::new()
+        .filename(db_path)
+        .connect()
+        .await?;
+
+    let auto_vacuum: u32 = sqlx::query("PRAGMA auto_vacuum")
+        .fetch_one(&mut conn)
+        .await?
+        .get(0);
+
+    if auto_vacuum != 0 {
+        return Ok(());
+    }
+
+    sqlx::query("PRAGMA auto_vacuum=FULL")
+        .execute(&mut conn)
+        .await?;
+
+    // Execute `VACUUM` command for the `auto_vacuum` pragma to take effect.
+    sqlx::query("VACUUM").execute(&mut conn).await?;
+
+    Ok(())
 }
 
 async fn get_pragma(conn: &mut Connection, name: &str) -> Result<u32, Error> {

lib/src/repository/tests.rs

@@ -13,6 +13,7 @@ use tokio::{
     sync::broadcast::Receiver,
     time::{self, Duration, timeout},
 };
+use tokio_stream::wrappers::ReadDirStream;
 use tracing::instrument;
 
 #[tokio::test(flavor = "multi_thread")]
@@ -1190,16 +1191,13 @@ async fn aux_db_files_are_deleted_on_close() {
     test_utils::init_log();
 
     let (temp_dir, repo) = setup().await;
-
     repo.close().await.unwrap();
 
-    let mut read_dir = fs::read_dir(temp_dir.path()).await.unwrap();
-    let mut entries = Vec::new();
-
-    while let Some(entry) = read_dir.next_entry().await.unwrap() {
-        entries.push(entry.path());
-    }
-
+    let entries: Vec<_> = ReadDirStream::new(fs::read_dir(temp_dir.path()).await.unwrap())
+        .map_ok(|entry| entry.path())
+        .try_collect()
+        .await
+        .unwrap();
     assert_eq!(entries, [temp_dir.path().join(DEFAULT_REPO_NAME)]);
 }
 
@@ -1255,6 +1253,63 @@ async fn export() {
     assert_eq!(dst_repo.access_mode(), AccessMode::Read);
 }
 
+#[tokio::test]
+async fn repository_size_decreases_after_delete() {
+    let (_base_dir, repo_path) = {
+        let (base_dir, repo) = setup().await;
+        let repo_path = base_dir.path().join(DEFAULT_REPO_NAME);
+        repo.close().await.unwrap();
+        (base_dir, repo_path)
+    };
+    let file_size: u64 = 1024 * 1024;
+
+    let repo_size_initial = fs::metadata(&repo_path).await.unwrap().len();
+    tracing::info!(?repo_size_initial);
+
+    {
+        let repo = Repository::open(&RepositoryParams::new(&repo_path), None, AccessMode::Write)
+            .await
+            .unwrap();
+        let mut file = repo.create_file("data").await.unwrap();
+        write_random_data(&mut file, file_size as usize).await;
+        file.flush().await.unwrap();
+        repo.close().await.unwrap();
+    }
+
+    ensure_aux_db_files_are_deleted(&repo_path).await;
+
+    let repo_size_after_create = fs::metadata(&repo_path).await.unwrap().len();
+    tracing::info!(?repo_size_after_create);
+    assert!(
+        repo_size_after_create >= repo_size_initial + file_size,
+        "actual size: {}, expected min size: {}",
+        repo_size_after_create,
+        repo_size_initial + file_size
+    );
+
+    {
+        let repo = Repository::open(&RepositoryParams::new(&repo_path), None, AccessMode::Write)
+            .await
+            .unwrap();
+        repo.remove_entry("data").await.unwrap();
+
+        wait_for(&repo, async || repo.count_blocks().await.unwrap() <= 1).await;
+
+        repo.close().await.unwrap();
+    }
+
+    ensure_aux_db_files_are_deleted(&repo_path).await;
+
+    let repo_size_after_delete = fs::metadata(&repo_path).await.unwrap().len();
+    tracing::info!(?repo_size_after_delete);
+    assert!(
+        repo_size_after_delete <= repo_size_after_create - file_size,
+        "actual size: {}, expected max size: {}",
+        repo_size_after_delete,
+        repo_size_after_create - file_size
+    );
+}
+
 const DEFAULT_REPO_NAME: &str = "repo.db";
 
 async fn setup() -> (TempDir, Repository) {
@@ -1311,6 +1366,19 @@ fn random_bytes(size: usize) -> Vec<u8> {
     buffer
 }
 
+async fn write_random_data(file: &mut File, size: usize) {
+    let mut buffer = [0u8; 4 * 1024];
+    let mut remaining = size;
+    let mut rng = rand::thread_rng();
+
+    while remaining > 0 {
+        let chunk_size = buffer.len().min(remaining);
+        rng.fill(&mut buffer[..chunk_size]);
+        file.write_all(&buffer).await.unwrap();
+        remaining -= chunk_size;
+    }
+}
+
 async fn wait_for_notification(rx: &mut Receiver<Event>) {
     match timeout(Duration::from_secs(5), rx.recv()).await {
         Ok(Ok(_)) => (),
@@ -1341,3 +1409,31 @@ where
     .await
     .expect("timeout waiting for condition")
 }
+
+// HACK: Due to a [bug in sqlx][1], the db aux files are not always deleted after a repository is
+// closed. As a workaround, this function reopens and closes the repository until the aux files are
+// gone.
+//
+// [1]: https://github.com/launchbadge/sqlx/issues/3217
+async fn ensure_aux_db_files_are_deleted(repo_path: &Path) {
+    let expected_files = [repo_path.to_owned()];
+
+    loop {
+        if ReadDirStream::new(fs::read_dir(repo_path.parent().unwrap()).await.unwrap())
+            .map_ok(|entry| entry.path())
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap()
+            == expected_files
+        {
+            break;
+        }
+
+        Repository::open(&RepositoryParams::new(repo_path), None, AccessMode::Blind)
+            .await
+            .unwrap()
+            .close()
+            .await
+            .unwrap();
+    }
+}