simplified startup

Author: Migorithm

duva/src/adapters/loggers/disk_based.rs

@@ -414,10 +414,7 @@ impl TWriteAheadLog for FileOpLogs {
     }
 
     /// Replays all existing operations in the op_logs, invoking a callback for each.
-    fn replay<F>(&mut self, mut replay_handler: F) -> Result<()>
-    where
-        F: FnMut(WriteOperation) + Send,
-    {
+    fn replay(&mut self, mut replay_handler: &mut dyn FnMut(WriteOperation)) -> Result<()> {
         // Replay all segments in order with streaming to avoid loading everything into memory
         for segment in self.segments.iter_mut().chain(std::iter::once(&mut self.active_segment)) {
             if segment.lookups.is_empty() {
@@ -610,7 +607,7 @@ mod tests {
         let mut op_logs = FileOpLogs::new(&path)?;
         let mut ops = Vec::new();
 
-        op_logs.replay(|op| {
+        op_logs.replay(&mut |op| {
             ops.push(op);
         })?;
 
@@ -657,7 +654,7 @@ mod tests {
 
         assert!(
             op_logs
-                .replay(|op| {
+                .replay(&mut |op| {
                     ops.push(op);
                 })
                 .is_err()
@@ -775,7 +772,7 @@ mod tests {
 
         // Verify we can read operations from both segments
         let mut ops = Vec::new();
-        op_logs.replay(|op| ops.push(op))?;
+        op_logs.replay(&mut |op| ops.push(op))?;
         assert_eq!(ops.len(), 101);
 
         Ok(())

duva/src/adapters/loggers/memory_based.rs

@@ -14,10 +14,7 @@ impl TWriteAheadLog for MemoryOpLogs {
         Ok(())
     }
 
-    fn replay<F>(&mut self, mut f: F) -> Result<()>
-    where
-        F: FnMut(WriteOperation) + Send,
-    {
+    fn replay(&mut self, f: &mut dyn FnMut(WriteOperation)) -> Result<()> {
         for op in self.writer.iter() {
             f(op.clone());
         }

duva/src/adapters/loggers/mod.rs

@@ -5,25 +5,15 @@ mod memory_based;
 pub mod op_logs;
 
 trait TWriteAheadLog: Send + Sync + 'static {
-    /// Append one or more `WriteOperation`s to the log.
     fn write_many(&mut self, ops: Vec<WriteOperation>) -> anyhow::Result<()>;
 
-    /// Retrieve logs that fall between the current 'commit' index and target 'log' index.
-    /// This is NOT async as it is expected to be infallible and an in-memory operation.
     fn range(&self, start_exclusive: u64, end_inclusive: u64) -> Vec<WriteOperation>;
 
-    /// Replays all logged operations from the beginning of the WAL, calling the provided callback `f` for each operation.
-    /// The callback `f(WriteOperation)` receives each operation in the order it was appended.
-    fn replay<F>(&mut self, f: F) -> anyhow::Result<()>
-    where
-        F: FnMut(WriteOperation) + Send;
+    fn replay(&mut self, f: &mut dyn FnMut(WriteOperation)) -> anyhow::Result<()>;
 
-    /// Retrieves the log at a given index.
     fn read_at(&mut self, at: u64) -> Option<WriteOperation>;
 
-    /// Returns true if there are no logs. Otherwise, returns false.
     fn is_empty(&self) -> bool;
 
-    /// Truncate logs that are positioned after `log_index`.
     fn truncate_after(&mut self, log_index: u64);
 }

duva/src/adapters/loggers/op_logs.rs

@@ -17,60 +17,48 @@ impl OperationLogs {
     pub fn new_ondisk() -> Self {
         Self::OnDisk(FileOpLogs::new(ENV.dir.clone()).unwrap())
     }
-    /// Append one or more `WriteOperation`s to the log.
-    pub(crate) fn write_many(&mut self, ops: Vec<WriteOperation>) -> anyhow::Result<()> {
+    fn get_mut(&mut self) -> &mut dyn TWriteAheadLog {
+        match self {
+            OperationLogs::Memory(m) => m as &mut dyn TWriteAheadLog,
+            OperationLogs::OnDisk(f) => f as &mut dyn TWriteAheadLog,
+        }
+    }
+    fn get(&self) -> &dyn TWriteAheadLog {
         match self {
-            OperationLogs::Memory(memory_op_logs) => memory_op_logs.write_many(ops),
-            OperationLogs::OnDisk(file_op_logs) => file_op_logs.write_many(ops),
+            OperationLogs::Memory(m) => m as &dyn TWriteAheadLog,
+            OperationLogs::OnDisk(f) => f as &dyn TWriteAheadLog,
         }
     }
 
+    /// Append one or more `WriteOperation`s to the log.
+    pub(crate) fn write_many(&mut self, ops: Vec<WriteOperation>) -> anyhow::Result<()> {
+        self.get_mut().write_many(ops)
+    }
+
     /// Retrieve logs that fall between the current 'commit' index and target 'log' index.
     /// This is NOT async as it is expected to be infallible and an in-memory operation.
     pub(crate) fn range(&self, start_exclusive: u64, end_inclusive: u64) -> Vec<WriteOperation> {
-        match self {
-            OperationLogs::Memory(memory_op_logs) => {
-                memory_op_logs.range(start_exclusive, end_inclusive)
-            },
-            OperationLogs::OnDisk(file_op_logs) => {
-                file_op_logs.range(start_exclusive, end_inclusive)
-            },
-        }
+        self.get().range(start_exclusive, end_inclusive)
     }
 
     /// Replays all logged operations from the beginning of the WAL, calling the provided callback `f` for each operation.
     /// The callback `f(WriteOperation)` receives each operation in the order it was appended.
-    pub(crate) fn replay<F>(&mut self, f: F) -> anyhow::Result<()>
-    where
-        F: FnMut(WriteOperation) + Send,
-    {
-        match self {
-            OperationLogs::Memory(memory_op_logs) => memory_op_logs.replay(f),
-            OperationLogs::OnDisk(file_op_logs) => file_op_logs.replay(f),
-        }
+    pub(crate) fn replay(&mut self, f: &mut dyn FnMut(WriteOperation)) -> anyhow::Result<()> {
+        self.get_mut().replay(f)
     }
 
     /// Retrieves the log at a given index.
     pub(crate) fn read_at(&mut self, at: u64) -> Option<WriteOperation> {
-        match self {
-            OperationLogs::Memory(memory_op_logs) => memory_op_logs.read_at(at),
-            OperationLogs::OnDisk(file_op_logs) => file_op_logs.read_at(at),
-        }
+        self.get_mut().read_at(at)
     }
 
     /// Returns true if there are no logs. Otherwise, returns false.
     pub(crate) fn is_empty(&self) -> bool {
-        match self {
-            OperationLogs::Memory(memory_op_logs) => memory_op_logs.is_empty(),
-            OperationLogs::OnDisk(file_op_logs) => file_op_logs.is_empty(),
-        }
+        self.get().is_empty()
     }
 
     /// Truncate logs that are positioned after `log_index`.
     pub(crate) fn truncate_after(&mut self, log_index: u64) {
-        match self {
-            OperationLogs::Memory(memory_op_logs) => memory_op_logs.truncate_after(log_index),
-            OperationLogs::OnDisk(file_op_logs) => file_op_logs.truncate_after(log_index),
-        }
+        self.get_mut().truncate_after(log_index)
     }
 }

duva/src/main.rs

@@ -4,13 +4,9 @@ use duva::{ENV, Environment, StartUpFacade, adapters::loggers::op_logs::Operatio
 async fn main() -> anyhow::Result<()> {
     let topology_writer = Environment::open_topology_file(ENV.tpp.clone()).await;
 
-    if ENV.append_only {
-        let local_aof = OperationLogs::new_ondisk();
-        let start_up_runner = StartUpFacade::new(local_aof, topology_writer);
-        start_up_runner.run().await
-    } else {
-        let in_memory_aof = OperationLogs::new_inmemory();
-        let start_up_runner = StartUpFacade::new(in_memory_aof, topology_writer);
-        start_up_runner.run().await
-    }
+    let op_logs =
+        if ENV.append_only { OperationLogs::new_ondisk() } else { OperationLogs::new_inmemory() };
+
+    let start_up_runner = StartUpFacade::new(op_logs, topology_writer);
+    start_up_runner.run().await
 }