swap segments in injector

Author: MarinPostma

libsql-wal/src/replication/injector.rs

@@ -1,6 +1,5 @@
 //! The injector is the module in charge of injecting frames into a replica database.
 
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use crate::error::Result;
@@ -46,6 +45,7 @@ impl<IO: Io> Injector<IO> {
     pub fn current_durable(&self) -> u64 {
         *self.wal.durable_frame_no.lock()
     }
+
     pub fn maybe_begin_txn(&mut self) -> Result<()> {
         if self.tx.is_none() {
             let mut tx = Transaction::Read(self.wal.begin_read(u64::MAX));
@@ -94,6 +94,15 @@ impl<IO: Io> Injector<IO> {
             if size_after.is_some() {
                 let mut tx = self.tx.take().unwrap();
                 self.wal.new_frame_notifier.send_replace(last_committed_frame_no);
+                // the strategy to swap the current log is to do it on change of durable boundary,
+                // when we have caught up with the current durable frame_no
+                if self.current_durable() != self.previous_durable_frame_no && self.current_durable() >= self.max_tx_frame_no {
+                    let wal = self.wal.clone();
+                    // FIXME: tokio dependency here is annoying, we need an async version of swap_current.
+                    tokio::task::spawn_blocking(move || {
+                        tx.commit();
+                        wal.swap_current(&tx)
+                    }).await.unwrap()?
                 }
             }
         }