store libsql-wal segments in the db directory

Instead of having a `wals` directory to manage all the wals for all the
dbs, we now store the wals for a specific db alonside the db itself, in
a `{db_name}-wal` directory. This servers multiple purposes
- it's easier to cleanup a db: just remove the dir
- the directory is placed where sqlite would expect the wal to be, but
  since it's a directory, it will refuse to open it, thus preventing
  accidental interferences between libsql-wal and sqlite
- when forking, we can create the fork in a separate tmp dir, and then
  swap the whole dir, in place with all the wal contents
- The regitry doesn't manage any of it's own files anymore, it'll be
  easier to integrate in embedded replicas.

Author: MarinPostma

libsql-server/src/bottomless_migrate.rs

@@ -50,7 +50,6 @@ pub async fn bottomless_migrate(
     let tmp = TempDir::new()?;
 
     tokio::fs::create_dir_all(tmp.path().join("dbs")).await?;
-    tokio::fs::create_dir_all(tmp.path().join("wals")).await?;
 
     let configs_stream = meta_store.namespaces();
     tokio::pin!(configs_stream);
@@ -68,7 +67,6 @@ pub async fn bottomless_migrate(
     });
 
     let tmp_registry = Arc::new(WalRegistry::new(
-        tmp.path().join("wals"),
         NoStorage.into(),
         sender,
     )?);
@@ -110,7 +108,6 @@ pub async fn bottomless_migrate(
     // doesn't exist, then we restore it, otherwise, we delete it.
     tokio::fs::rename(&base_dbs_dir, &base_dbs_dir_tmp).await?;
     tokio::fs::rename(tmp.path().join("dbs"), base_dbs_dir).await?;
-    tokio::fs::rename(tmp.path().join("wals"), base_config.base_path.join("wals")).await?;
     tokio::fs::remove_dir_all(base_config.base_path.join("_dbs")).await?;
 
     Ok(())

libsql-server/src/lib.rs

@@ -835,20 +835,6 @@ where
         scripted_backup: Option<ScriptBackupManager>,
         meta_store: MetaStore,
     ) -> anyhow::Result<(NamespaceConfigurators, MakeReplicationSvc)> {
-        let wal_path = base_config.base_path.join("wals");
-        let enable_libsql_wal_test = {
-            let is_primary = self.rpc_server_config.is_some();
-            let is_libsql_wal_test = std::env::var("LIBSQL_WAL_TEST").is_ok();
-            is_primary && is_libsql_wal_test
-        };
-        let use_libsql_wal =
-            self.use_custom_wal == Some(CustomWAL::LibsqlWal) || enable_libsql_wal_test;
-        if !use_libsql_wal {
-            if wal_path.try_exists()? {
-                anyhow::bail!("database was previously setup to use libsql-wal");
-            }
-        }
-
         #[cfg(feature = "durable-wal")]
         if let Some(CustomWAL::DurableWal) = self.use_custom_wal {
             if self.db_config.bottomless_replication.is_some() {
@@ -864,7 +850,6 @@ where
                     task_manager,
                     migration_scheduler_handle,
                     scripted_backup,
-                    wal_path,
                     meta_store,
                 )
                 .await
@@ -895,7 +880,6 @@ where
         task_manager: &mut TaskManager,
         migration_scheduler_handle: SchedulerHandle,
         scripted_backup: Option<ScriptBackupManager>,
-        wal_path: PathBuf,
         meta_store: MetaStore,
     ) -> anyhow::Result<(NamespaceConfigurators, MakeReplicationSvc)> {
         tracing::info!("using libsql wal");
@@ -964,7 +948,7 @@ where
             anyhow::bail!("replication without bottomless not supported yet");
         }
 
-        let registry = Arc::new(WalRegistry::new(wal_path, storage, sender)?);
+        let registry = Arc::new(WalRegistry::new(storage, sender)?);
         let checkpointer = LibsqlCheckpointer::new(registry.clone(), receiver, 8);
         task_manager.spawn_with_shutdown_notify(|_| async move {
             checkpointer.run().await;
@@ -1288,15 +1272,16 @@ where
     }
 
     fn check_previous_migration_success(&self) -> anyhow::Result<bool> {
-        let wals_path = self.path.join("wals");
-        if !wals_path.try_exists()? {
-            return Ok(false);
-        }
-
-        let dir = std::fs::read_dir(&wals_path)?;
-
-        // wals dir exist and is not empty
-        Ok(dir.count() != 0)
+        todo!("not usings wal directory anymore");
+        // let wals_path = self.path.join("wals");
+        // if !wals_path.try_exists()? {
+        //     return Ok(false);
+        // }
+        //
+        // let dir = std::fs::read_dir(&wals_path)?;
+        //
+        // // wals dir exist and is not empty
+        // Ok(dir.count() != 0)
     }
 }
 

libsql-server/src/namespace/configurator/helpers.rs

@@ -494,28 +494,6 @@ pub async fn cleanup_libsql(
         let _ = tokio::fs::remove_dir_all(ns_db_path).await;
     }
 
-    let ns_wals_path = base_path.join("wals").join(namespace.as_str());
-    if ns_wals_path.try_exists()? {
-        tracing::debug!("removing database directory: {}", ns_wals_path.display());
-        if let Err(e) = tokio::fs::remove_dir_all(ns_wals_path).await {
-            // what can go wrong?:
-            match e.kind() {
-                // alright, there's nothing to delete anyway
-                std::io::ErrorKind::NotFound => (),
-                _ => {
-                    // something unexpected happened, this namespaces is in a bad state.
-                    // The entry will not be removed from the registry to prevent another
-                    // namespace with the same name to be reuse the same wal files. a
-                    // manual intervention is necessary
-                    // FIXME: on namespace creation, we could ensure that this directory is
-                    // clean.
-                    tracing::error!("error deleting `{namespace}` wal directory, manual intervention may be necessary: {e}");
-                    return Err(e.into());
-                }
-            }
-        }
-    }
-
     // when all is cleaned, leave place for next one
     registry.remove(&namespace).await;
 

libsql-wal/benches/benchmarks.rs

@@ -61,7 +61,7 @@ fn with_libsql_conn(f: impl FnOnce(&mut Connection<LibsqlWal<StdIO>>)) {
 
     let (sender, _) = tokio::sync::mpsc::channel(12);
     let registry =
-        Arc::new(WalRegistry::new(tmp.path().join("wals"), NoStorage.into(), sender).unwrap());
+        Arc::new(WalRegistry::new(NoStorage.into(), sender).unwrap());
     let wal_manager = LibsqlWalManager::new(registry.clone(), Arc::new(resolver));
 
     let mut conn = libsql_sys::Connection::open(

libsql-wal/src/lib.rs

@@ -112,7 +112,6 @@ pub mod test {
             let registry = Arc::new(
                 WalRegistry::new_with_io(
                     io.clone(),
-                    tmp.path().join("test/wals"),
                     TestStorage::new_io(store, io).into(),
                     sender,
                 )

libsql-wal/src/registry.rs

@@ -1,6 +1,6 @@
 use std::io;
 use std::num::NonZeroU64;
-use std::path::{Path, PathBuf};
+use std::path::Path;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -47,7 +47,6 @@ enum Slot<IO: Io> {
 /// Wal Registry maintains a set of shared Wal, and their respective set of files.
 pub struct WalRegistry<IO: Io, S> {
     io: Arc<IO>,
-    path: PathBuf,
     shutdown: AtomicBool,
     opened: DashMap<NamespaceName, Slot<IO>>,
     storage: Arc<S>,
@@ -56,25 +55,21 @@ pub struct WalRegistry<IO: Io, S> {
 
 impl<S> WalRegistry<StdIO, S> {
     pub fn new(
-        path: PathBuf,
         storage: Arc<S>,
         checkpoint_notifier: mpsc::Sender<CheckpointMessage>,
     ) -> Result<Self> {
-        Self::new_with_io(StdIO(()), path, storage, checkpoint_notifier)
+        Self::new_with_io(StdIO(()), storage, checkpoint_notifier)
     }
 }
 
 impl<IO: Io, S> WalRegistry<IO, S> {
     pub fn new_with_io(
         io: IO,
-        path: PathBuf,
         storage: Arc<S>,
         checkpoint_notifier: mpsc::Sender<CheckpointMessage>,
     ) -> Result<Self> {
-        io.create_dir_all(&path)?;
         let registry = Self {
             io: io.into(),
-            path,
             opened: Default::default(),
             shutdown: Default::default(),
             storage,
@@ -246,10 +241,14 @@ where
 
         let mut checkpointed_frame_no = footer.map(|f| f.replication_index.get()).unwrap_or(0);
 
-        let path = self.path.join(namespace.as_str());
-        self.io.create_dir_all(&path)?;
+        // the trick here to prevent sqlite to open our db is to create a dir <db-name>-wal. Sqlite
+        // will think that this is a wal file, but it's in fact a directory and it will not like
+        // it.
+        let mut wals_path = db_path.to_owned();
+        wals_path.set_file_name(format!("{}-wal", db_path.file_name().unwrap().to_str().unwrap()));
+        self.io.create_dir_all(&wals_path)?;
         // TODO: handle that with abstract io
-        let dir = walkdir::WalkDir::new(&path).sort_by_file_name().into_iter();
+        let dir = walkdir::WalkDir::new(&wals_path).sort_by_file_name().into_iter();
 
         // we only checkpoint durable frame_no so this is a good first estimate without an actual
         // network call.
@@ -323,7 +322,7 @@ where
                 None => (0, NonZeroU64::new(1).unwrap()),
             });
 
-        let current_segment_path = path.join(format!("{namespace}:{next_frame_no:020}.seg"));
+        let current_segment_path = wals_path.join(format!("{next_frame_no:020}.seg"));
 
         let segment_file = self.io.open(true, true, true, &current_segment_path)?;
         let salt = self.io.with_rng(|rng| rng.gen());
@@ -368,6 +367,7 @@ where
             checkpoint_notifier: self.checkpoint_notifier.clone(),
             io: self.io.clone(),
             swap_strategy,
+            wals_path: wals_path.to_owned(),
         });
 
         self.opened
@@ -524,10 +524,9 @@ where
             return Ok(());
         }
         let start_frame_no = current.next_frame_no();
-        let path = self
-            .path
-            .join(shared.namespace().as_str())
-            .join(format!("{}:{start_frame_no:020}.seg", shared.namespace()));
+        let path = shared
+            .wals_path
+            .join(format!("{start_frame_no:020}.seg"));
 
         let segment_file = self.io.open(true, true, true, &path)?;
         let salt = self.io.with_rng(|rng| rng.gen());

libsql-wal/src/shared_wal.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
@@ -55,6 +56,7 @@ pub struct SharedWal<IO: Io> {
     pub(crate) checkpoint_notifier: mpsc::Sender<CheckpointMessage>,
     pub(crate) io: Arc<IO>,
     pub(crate) swap_strategy: Box<dyn SegmentSwapStrategy>,
+    pub(crate) wals_path: PathBuf,
 }
 
 impl<IO: Io> SharedWal<IO> {

libsql-wal/tests/flaky_fs.rs

@@ -213,14 +213,13 @@ async fn flaky_fs() {
     let registry = Arc::new(
         WalRegistry::new_with_io(
             io.clone(),
-            tmp.path().join("test/wals"),
             TestStorage::new_io(false, io).into(),
             sender,
         )
         .unwrap(),
     );
     let wal_manager = LibsqlWalManager::new(registry.clone(), Arc::new(resolver));
-
+    tokio::fs::create_dir_all(tmp.path().join("test")).await.unwrap();
     let conn = libsql_sys::Connection::open(
         tmp.path().join("test/data").clone(),
         OpenFlags::SQLITE_OPEN_CREATE | OpenFlags::SQLITE_OPEN_READ_WRITE,

libsql-wal/tests/oracle.rs

@@ -11,7 +11,7 @@ use std::sync::Arc;
 use libsql_sys::ffi::{sqlite3_finalize, sqlite3_prepare, Sqlite3DbHeader};
 use libsql_sys::name::NamespaceName;
 use libsql_sys::rusqlite::OpenFlags;
-use libsql_sys::wal::{Sqlite3WalManager, Wal};
+use libsql_sys::wal::{Sqlite3WalManager, Wal, WalManager};
 use libsql_sys::Connection;
 use libsql_wal::registry::WalRegistry;
 use libsql_wal::storage::TestStorage;
@@ -67,7 +67,7 @@ async fn run_test_sample(path: &Path) -> Result {
 
     let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(42);
     let before = std::time::Instant::now();
-    let sqlite_results = run_script(&sqlite_conn, &script, &mut rng).collect::<Vec<_>>();
+    let sqlite_results = run_script(&sqlite_conn, &script, &mut rng, Sqlite3WalManager::default()).collect::<Vec<_>>();
     println!("ran sqlite in {:?}", before.elapsed());
     drop(sqlite_conn);
 
@@ -95,13 +95,13 @@ async fn run_test_sample(path: &Path) -> Result {
     let (sender, _receiver) = tokio::sync::mpsc::channel(64);
     let registry = Arc::new(
         WalRegistry::new(
-            tmp.path().join("test/wals"),
             TestStorage::new().into(),
             sender,
         )
         .unwrap(),
     );
     let wal_manager = LibsqlWalManager::new(registry.clone(), Arc::new(resolver));
+    tokio::fs::create_dir_all(tmp.path().join("test")).await.unwrap();
     let db_path = tmp.path().join("test/data").clone();
     let libsql_conn = libsql_sys::Connection::open(
         &db_path,
@@ -114,7 +114,7 @@ async fn run_test_sample(path: &Path) -> Result {
 
     let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(42);
     let before = std::time::Instant::now();
-    let libsql_results = run_script(&libsql_conn, &script, &mut rng).collect::<Vec<_>>();
+    let libsql_results = run_script(&libsql_conn, &script, &mut rng, wal_manager.clone()).collect::<Vec<_>>();
     println!("ran libsql in {:?}", before.elapsed());
 
     for ((a, _), (b, _)) in sqlite_results.iter().zip(libsql_results.iter()) {
@@ -204,12 +204,13 @@ fn run_script<'a, T: Wal>(
     conn: &'a Connection<T>,
     script: &'a str,
     rng: &'a mut ChaCha8Rng,
+    wal_manager: impl WalManager + Clone + 'static,
 ) -> impl Iterator<Item = (String, String)> + 'a {
     let mut stmts = split_statements(conn, script);
     std::iter::from_fn(move || {
         let stmt_str = patch_randomness(stmts.next()?, rng);
         if stmt_str.trim_start().starts_with("ATTACH") {
-            patch_attach(&stmt_str);
+            patch_attach(&stmt_str, wal_manager.clone());
         }
         let mut stmt = conn.prepare(&stmt_str).unwrap();
 
@@ -227,14 +228,14 @@ fn run_script<'a, T: Wal>(
     })
 }
 
-fn patch_attach(s: &str) {
+fn patch_attach(s: &str, wal: impl WalManager) {
     let mut split = s.split_whitespace();
     let name = split.nth(1).unwrap();
     let name = name.trim_matches('\'');
     let _ = libsql_sys::Connection::open(
         name,
         OpenFlags::SQLITE_OPEN_CREATE | OpenFlags::SQLITE_OPEN_READ_WRITE,
-        Sqlite3WalManager::default(),
+        wal,
         100000,
         None,
     )