pass durable frame_no to libsql-injector

Author: MarinPostma

libsql-replication/src/injector/libsql_injector.rs

@@ -49,4 +49,8 @@ impl super::Injector for LibsqlInjector {
             .map_err(|e| Error::FatalInjectError(e.into()))?;
         Ok(None)
     }
+
+    fn durable_frame_no(&mut self, frame_no: u64) {
+        self.injector.set_durable(frame_no);
+    }
 }

libsql-replication/src/injector/mod.rs

@@ -29,4 +29,6 @@ pub trait Injector {
     /// Trigger a dummy write, and flush the cache to trigger a call to xFrame. The buffer's frame
     /// are then injected into the wal.
     fn flush(&mut self) -> impl Future<Output = Result<Option<FrameNo>>> + Send;
+
+    fn durable_frame_no(&mut self, frame_no: u64);
 }

libsql-replication/src/injector/sqlite_injector/mod.rs

@@ -46,6 +46,9 @@ impl Injector for SqliteInjector {
         let inner = self.inner.clone();
         spawn_blocking(move || inner.lock().flush()).await.unwrap()
     }
+
+    #[inline]
+    fn durable_frame_no(&mut self, _frame_no: u64) { }
 }
 
 impl SqliteInjector {

libsql-wal/src/replication/injector.rs

@@ -1,5 +1,6 @@
 //! 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;
@@ -36,6 +37,14 @@ impl<IO: Io> Injector<IO> {
         })
     }
 
+    pub fn set_durable(&self, durable_frame_no: u64) {
+        let mut old = self.wal.durable_frame_no.lock();
+        if *old < durable_frame_no {
+            *old = durable_frame_no
+        } {
+            todo!("primary reported older frameno than current");
+        }
+    }
     pub async fn insert_frame(&mut self, frame: Box<Frame>) -> Result<Option<u64>> {
         let size_after = frame.size_after();
         self.max_tx_frame_no = self.max_tx_frame_no.max(frame.header().frame_no());