changes to compactor trait

Author: MarinPostma

Cargo.lock

@@ -48,3 +48,4 @@ uuid = { version = "1.4.0", features = ["v4", "serde"] }
 
 [dev-dependencies]
 turmoil = "0.5.5"
+walkdir = "2.3.3"

libsqlx-server/Cargo.toml

@@ -171,15 +171,14 @@ impl Injector {
 
 #[cfg(test)]
 mod test {
-    use std::fs::File;
+    use std::path::PathBuf;
 
     use crate::database::libsql::injector::Injector;
     use crate::database::libsql::replication_log::logger::LogFile;
 
     #[test]
     fn test_simple_inject_frames() {
-        let file = File::open("assets/test/simple_wallog").unwrap();
-        let log = LogFile::new(file).unwrap();
+        let log = LogFile::new(PathBuf::from("assets/test/simple_wallog")).unwrap();
         let temp = tempfile::tempdir().unwrap();
 
         let mut injector = Injector::new(temp.path(), Box::new(()), 10).unwrap();
@@ -199,8 +198,7 @@ mod test {
 
     #[test]
     fn test_inject_frames_split_txn() {
-        let file = File::open("assets/test/simple_wallog").unwrap();
-        let log = LogFile::new(file).unwrap();
+        let log = LogFile::new(PathBuf::from("assets/test/simple_wallog")).unwrap();
         let temp = tempfile::tempdir().unwrap();
 
         // inject one frame at a time
@@ -221,8 +219,7 @@ mod test {
 
     #[test]
     fn test_inject_partial_txn_isolated() {
-        let file = File::open("assets/test/simple_wallog").unwrap();
-        let log = LogFile::new(file).unwrap();
+        let log = LogFile::new(PathBuf::from("assets/test/simple_wallog")).unwrap();
         let temp = tempfile::tempdir().unwrap();
 
         // inject one frame at a time

libsqlx/src/database/libsql/injector/mod.rs

@@ -19,7 +19,6 @@ use self::replication_log::logger::FrameNotifierCb;
 
 pub use connection::LibsqlConnection;
 pub use replication_log::logger::{LogCompactor, LogFile};
-pub use replication_log::merger::SnapshotMerger;
 
 mod connection;
 mod injector;
@@ -196,12 +195,12 @@ impl InjectableDatabase for LibsqlDatabase<ReplicaType> {
 
 #[cfg(test)]
 mod test {
-    use std::fs::File;
     use std::sync::atomic::AtomicBool;
     use std::sync::atomic::Ordering::Relaxed;
 
     use parking_lot::Mutex;
     use rusqlite::types::Value;
+    use uuid::Uuid;
 
     use crate::connection::Connection;
     use crate::database::libsql::replication_log::logger::LogFile;
@@ -238,8 +237,7 @@ mod test {
             .unwrap();
         assert!(row.lock().is_empty());
 
-        let file = File::open("assets/test/simple_wallog").unwrap();
-        let log = LogFile::new(file).unwrap();
+        let log = LogFile::new(PathBuf::from("assets/test/simple_wallog")).unwrap();
         let mut injector = db.injector().unwrap();
         log.frames_iter().unwrap().for_each(|f| {
             injector.inject(f.unwrap()).unwrap();
@@ -312,14 +310,16 @@ mod test {
             }
 
             fn compact(
-                &self,
-                _file: LogFile,
-                _path: PathBuf,
-                _size_after: u32,
+                &mut self,
+                _id: Uuid,
             ) -> Result<(), Box<dyn std::error::Error + Sync + Send + 'static>> {
                 self.0.store(true, Relaxed);
                 Ok(())
             }
+
+            fn snapshot_dir(&self) -> PathBuf {
+                todo!();
+            }
         }
 
         let temp = tempfile::tempdir().unwrap();
@@ -353,13 +353,15 @@ mod test {
             }
 
             fn compact(
-                &self,
-                _file: LogFile,
-                _path: PathBuf,
-                _size_after: u32,
+                &mut self,
+                _id: Uuid,
             ) -> Result<(), Box<dyn std::error::Error + Sync + Send + 'static>> {
                 unreachable!()
             }
+
+            fn snapshot_dir(&self) -> PathBuf {
+                todo!()
+            }
         }
 
         let temp = tempfile::tempdir().unwrap();

libsqlx/src/database/libsql/mod.rs

@@ -26,7 +26,6 @@ use crate::database::frame::{Frame, FrameHeader};
 #[cfg(feature = "bottomless")]
 use crate::libsql::ffi::SQLITE_IOERR_WRITE;
 
-use super::snapshot::{find_snapshot_file, SnapshotFile};
 use super::{FrameNo, CRC_64_GO_ISO, WAL_MAGIC, WAL_PAGE_SIZE};
 
 init_static_wal_method!(REPLICATION_METHODS, ReplicationLoggerHook);
@@ -767,15 +766,18 @@ pub trait LogCompactor: Sync + Send + 'static {
 impl LogCompactor for () {
     fn compact(
         &mut self,
-        log_name: String,
-        path: PathBuf,
+        _log_id: Uuid,
     ) -> Result<(), Box<dyn std::error::Error + Sync + Send + 'static>> {
         Ok(())
     }
 
     fn should_compact(&self, _file: &LogFile) -> bool {
         false
     }
+
+    fn snapshot_dir(&self) -> PathBuf {
+        todo!()
+    }
 }
 
 pub type FrameNotifierCb = Box<dyn Fn(FrameNo) + Send + Sync + 'static>;
@@ -784,7 +786,6 @@ pub struct ReplicationLogger {
     pub generation: Generation,
     pub log_file: RwLock<LogFile>,
     compactor: Box<Mutex<dyn LogCompactor + Send>>,
-    db_path: PathBuf,
     /// a notifier channel other tasks can subscribe to, and get notified when new frames become
     /// available.
     pub new_frame_notifier: FrameNotifierCb,
@@ -821,17 +822,11 @@ impl ReplicationLogger {
         if should_recover {
             Self::recover(log_file, data_path, compactor, new_frame_notifier)
         } else {
-            Self::from_log_file(
-                db_path.to_path_buf(),
-                log_file,
-                compactor,
-                new_frame_notifier,
-            )
+            Self::from_log_file(log_file, compactor, new_frame_notifier)
         }
     }
 
     fn from_log_file(
-        db_path: PathBuf,
         log_file: LogFile,
         compactor: impl LogCompactor,
         new_frame_notifier: FrameNotifierCb,
@@ -843,7 +838,6 @@ impl ReplicationLogger {
             generation: Generation::new(generation_start_frame_no),
             compactor: Box::new(Mutex::new(compactor)),
             log_file: RwLock::new(log_file),
-            db_path,
             new_frame_notifier,
         })
     }
@@ -885,7 +879,7 @@ impl ReplicationLogger {
 
         assert!(data_path.pop());
 
-        Self::from_log_file(data_path, log_file, compactor, new_frame_notifier)
+        Self::from_log_file(log_file, compactor, new_frame_notifier)
     }
 
     pub fn database_id(&self) -> anyhow::Result<Uuid> {
@@ -930,10 +924,6 @@ impl ReplicationLogger {
             .expect("there should be at least one frame after commit"))
     }
 
-    pub fn get_snapshot_file(&self, from: FrameNo) -> anyhow::Result<Option<SnapshotFile>> {
-        find_snapshot_file(&self.db_path, from)
-    }
-
     pub fn get_frame(&self, frame_no: FrameNo) -> Result<Frame, LogReadError> {
         self.log_file.read().frame(frame_no)
     }
@@ -1036,8 +1026,8 @@ mod test {
 
     #[test]
     fn log_file_test_rollback() {
-        let f = tempfile::tempfile().unwrap();
-        let mut log_file = LogFile::new(f).unwrap();
+        let f = tempfile::NamedTempFile::new().unwrap();
+        let mut log_file = LogFile::new(f.path().to_path_buf()).unwrap();
         (0..5)
             .map(|i| WalPage {
                 page_no: i,

libsqlx/src/database/libsql/replication_log/logger.rs

@@ -1,8 +1,7 @@
 use crc::Crc;
 
 pub mod logger;
-pub mod merger;
-pub mod snapshot;
+// pub mod merger;
 
 pub const WAL_PAGE_SIZE: i32 = 4096;
 pub const WAL_MAGIC: u64 = u64::from_le_bytes(*b"SQLDWAL\0");