add storage replicator to SharedWal

Author: MarinPostma

libsql-wal/src/registry.rs

@@ -294,6 +294,10 @@ where
             checkpointed_frame_no: header.replication_index.get().into(),
             new_frame_notifier,
             durable_frame_no,
+            stored_segments: Box::new(StorageReplicator::new(
+                self.storage.clone(),
+                namespace.clone(),
+            )),
         });
 
         opened.with_upgraded(|opened| {

libsql-wal/src/shared_wal.rs

@@ -11,6 +11,7 @@ use parking_lot::{Mutex, MutexGuard};
 use crate::error::{Error, Result};
 use crate::io::file::FileExt;
 use crate::io::Io;
+use crate::replication::storage::ReplicateFromStorage;
 use crate::segment::current::CurrentSegment;
 use crate::transaction::{ReadTransaction, Savepoint, Transaction, TxGuard, WriteTransaction};
 use libsql_sys::name::NamespaceName;
@@ -43,8 +44,9 @@ pub struct SharedWal<IO: Io> {
     #[allow(dead_code)] // used by replication
     pub(crate) checkpointed_frame_no: AtomicU64,
     /// max frame_no acknoledged by the durable storage
-    pub(crate) durable_frame_no: AtomicU64,
+    pub(crate) durable_frame_no: Arc<Mutex<u64>>,
     pub(crate) new_frame_notifier: tokio::sync::watch::Sender<u64>,
+    pub(crate) stored_segments: Box<dyn ReplicateFromStorage>,
 }
 
 impl<IO: Io> SharedWal<IO> {
@@ -265,7 +267,7 @@ impl<IO: Io> SharedWal<IO> {
     }
 
     pub async fn checkpoint(&self) -> Result<Option<u64>> {
-        let durable_frame_no = self.durable_frame_no.load(Ordering::SeqCst);
+        let durable_frame_no = *self.durable_frame_no.lock();
         let checkpointed_frame_no = self
             .current
             .load()
@@ -312,7 +314,7 @@ mod test {
 
         seal_current_segment(&shared);
 
-        shared.durable_frame_no.store(99999, Ordering::Relaxed);
+        *shared.durable_frame_no.lock() = 999999;
 
         let frame_no = shared.checkpoint().await.unwrap().unwrap();
         assert_eq!(frame_no, 4);