moved to frames_synced

Author: LucioFranco

libsql-replication/src/replicator.rs

@@ -141,6 +141,7 @@ pub struct Replicator<C> {
     client: C,
     injector: Arc<Mutex<Injector>>,
     state: ReplicatorState,
+    frames_synced: usize,
 }
 
 const INJECTOR_BUFFER_CAPACITY: usize = 10;
@@ -178,6 +179,7 @@ impl<C: ReplicatorClient> Replicator<C> {
             client,
             injector: Arc::new(Mutex::new(injector)),
             state: ReplicatorState::NeedHandshake,
+            frames_synced: 0,
         })
     }
 
@@ -311,6 +313,8 @@ impl<C: ReplicatorClient> Replicator<C> {
     }
 
     async fn inject_frame(&mut self, frame: Frame) -> Result<(), Error> {
+        self.frames_synced += 1;
+
         let injector = self.injector.clone();
         match spawn_blocking(move || injector.lock().inject_frame(frame)).await? {
             Ok(Some(commit_fno)) => {
@@ -335,6 +339,10 @@ impl<C: ReplicatorClient> Replicator<C> {
 
         Ok(())
     }
+
+    pub fn frames_synced(&self) -> usize {
+        self.frames_synced
+    }
 }
 
 /// Helper function to convert rpc frames results to replicator frames

libsql-server/tests/embedded_replica/mod.rs

@@ -725,7 +725,8 @@ fn replicate_with_snapshots() {
         .await
         .unwrap();
 
-        db.sync().await.unwrap();
+        let rep = db.sync().await.unwrap();
+        assert_eq!(rep.frames_synced(), 427);
 
         let conn = db.connect().unwrap();
 
@@ -757,7 +758,8 @@ fn replicate_with_snapshots() {
 
         assert_eq!(stat, 427);
 
-        db.sync().await.unwrap();
+        let rep = db.sync().await.unwrap();
+        assert_eq!(rep.frames_synced(), 427);
 
         let conn = db.connect().unwrap();
 
@@ -1230,15 +1232,7 @@ fn txn_bug_issue_1283() {
 #[test]
 fn replicated_return() {
     let tmp_embedded = tempdir().unwrap();
-    let tmp_host = tempdir().unwrap();
     let tmp_embedded_path = tmp_embedded.path().to_owned();
-    let tmp_host_path = tmp_host.path().to_owned();
-
-    let mut sim = Builder::new()
-        .simulation_duration(Duration::from_secs(1000))
-        .build();
-
-    // make_primary(&mut sim, tmp_host_path.clone());
 
     let mut sim = Builder::new()
         .simulation_duration(Duration::from_secs(1000))
@@ -1318,7 +1312,7 @@ fn replicated_return() {
 
         let rep = db.sync().await.unwrap();
         assert_eq!(rep.frame_no(), None);
-        assert_eq!(rep.start_frame_no(), None);
+        assert_eq!(rep.frames_synced(), 0);
 
         let conn = db.connect()?;
 
@@ -1328,7 +1322,7 @@ fn replicated_return() {
 
         let rep = db.sync().await.unwrap();
         assert_eq!(rep.frame_no(), Some(1));
-        assert_eq!(rep.start_frame_no(), None);
+        assert_eq!(rep.frames_synced(), 2);
 
         conn.execute_batch(
             "
@@ -1342,16 +1336,17 @@ fn replicated_return() {
 
         let rep = db.sync().await.unwrap();
         assert_eq!(rep.frame_no(), Some(10));
-        assert_eq!(rep.start_frame_no(), Some(1));
+        assert_eq!(rep.frames_synced(), 11);
 
+        // Regenerate log
         notify.notify_waiters();
         notify.notified().await;
 
         tokio::time::sleep(std::time::Duration::from_secs(5)).await;
 
         let rep = db.sync().await.unwrap();
         assert_eq!(rep.frame_no(), Some(4));
-        assert_eq!(rep.start_frame_no(), Some(10));
+        assert_eq!(rep.frames_synced(), 16);
 
         let mut row = conn.query("select count(*) from user", ()).await.unwrap();
         let count = row.next().await.unwrap().unwrap().get::<u64>(0).unwrap();

libsql/src/replication/mod.rs

@@ -35,16 +35,16 @@ pub(crate) mod remote_client;
 #[derive(Debug)]
 pub struct Replicated {
     frame_no: Option<FrameNo>,
-    start_frame_no: Option<FrameNo>,
+    frames_synced: usize,
 }
 
 impl Replicated {
     pub fn frame_no(&self) -> Option<FrameNo> {
         self.frame_no
     }
 
-    pub fn start_frame_no(&self) -> Option<FrameNo> {
-        self.start_frame_no
+    pub fn frames_synced(&self) -> usize {
+        self.frames_synced
     }
 }
 
@@ -209,8 +209,6 @@ impl EmbeddedReplicator {
             ));
         }
 
-        let start_frame_no = replicator.client_mut().committed_frame_no();
-
         // we force a handshake to get the most up to date replication index from the primary.
         replicator.force_handshake();
 
@@ -235,7 +233,7 @@ impl EmbeddedReplicator {
                     let Some(primary_index) = client.last_handshake_replication_index() else {
                         return Ok(Replicated {
                             frame_no: None,
-                            start_frame_no: None,
+                            frames_synced: 0,
                         });
                     };
                     if let Some(replica_index) = replicator.client_mut().committed_frame_no() {
@@ -249,7 +247,7 @@ impl EmbeddedReplicator {
 
         let replicated = Replicated {
             frame_no: replicator.client_mut().committed_frame_no(),
-            start_frame_no,
+            frames_synced: replicator.frames_synced(),
         };
 
         Ok(replicated)

libsql/src/replication/remote_client.rs

@@ -8,7 +8,9 @@ use futures::StreamExt as _;
 use libsql_replication::frame::{Frame, FrameHeader, FrameNo};
 use libsql_replication::meta::WalIndexMeta;
 use libsql_replication::replicator::{map_frame_err, Error, ReplicatorClient};
-use libsql_replication::rpc::replication::{verify_session_token, Frames, HelloRequest, LogOffset, SESSION_TOKEN_KEY, HelloResponse};
+use libsql_replication::rpc::replication::{
+    verify_session_token, Frames, HelloRequest, HelloResponse, LogOffset, SESSION_TOKEN_KEY,
+};
 use tokio_stream::Stream;
 use tonic::metadata::AsciiMetadataValue;
 use tonic::{Response, Status};
@@ -81,7 +83,10 @@ impl RemoteClient {
         self.last_handshake_replication_index
     }
 
-    async fn handle_handshake_response(&mut self, hello:Result<Response<HelloResponse>, Status>) -> Result<bool, Error> {
+    async fn handle_handshake_response(
+        &mut self,
+        hello: Result<Response<HelloResponse>, Status>,
+    ) -> Result<bool, Error> {
         let hello = hello?.into_inner();
         verify_session_token(&hello.session_token).map_err(Error::Client)?;
         let new_session = self.session_token != Some(hello.session_token.clone());
@@ -130,7 +135,9 @@ impl RemoteClient {
             (hello_fut.await, None)
         };
         self.prefetched_batch_log_entries = if let Ok(true) = hello.0 {
-            tracing::warn!("Frames prefetching failed because of new session token returned by handshake");
+            tracing::warn!(
+                "Frames prefetching failed because of new session token returned by handshake"
+            );
             None
         } else {
             frames
@@ -139,7 +146,10 @@ impl RemoteClient {
         hello
     }
 
-    async fn handle_next_frames_response(&mut self, frames: Result<Response<Frames>, Status>) -> Result<<Self as ReplicatorClient>::FrameStream, Error> {
+    async fn handle_next_frames_response(
+        &mut self,
+        frames: Result<Response<Frames>, Status>,
+    ) -> Result<<Self as ReplicatorClient>::FrameStream, Error> {
         let frames = frames?.into_inner().frames;
 
         if let Some(f) = frames.last() {
@@ -157,7 +167,12 @@ impl RemoteClient {
         Ok(Box::pin(stream))
     }
 
-    async fn do_next_frames(&mut self) -> (Result<<Self as ReplicatorClient>::FrameStream, Error>, Duration) {
+    async fn do_next_frames(
+        &mut self,
+    ) -> (
+        Result<<Self as ReplicatorClient>::FrameStream, Error>,
+        Duration,
+    ) {
         let (frames, time) = match self.prefetched_batch_log_entries.take() {
             Some((result, time)) => (result, time),
             None => {
@@ -197,7 +212,13 @@ impl RemoteClient {
     }
 }
 
-fn maybe_log<T>(time: Duration, sum: &mut Duration, count: &mut u128, result: &Result<T, Error>, op_name: &str) {
+fn maybe_log<T>(
+    time: Duration,
+    sum: &mut Duration,
+    count: &mut u128,
+    result: &Result<T, Error>,
+    op_name: &str,
+) {
     if let Err(e) = &result {
         tracing::warn!("Failed {} in {} ms: {:?}", op_name, time.as_millis(), e);
     } else {
@@ -206,7 +227,12 @@ fn maybe_log<T>(time: Duration, sum: &mut Duration, count: &mut u128, result: &R
         let avg = (*sum).as_millis() / *count;
         let time = time.as_millis();
         if *count > 10 && time > 2 * avg {
-            tracing::warn!("Unusually long {}. Took {} ms, average {} ms", op_name, time, avg);
+            tracing::warn!(
+                "Unusually long {}. Took {} ms, average {} ms",
+                op_name,
+                time,
+                avg
+            );
         }
     }
 }
@@ -218,22 +244,40 @@ impl ReplicatorClient for RemoteClient {
     /// Perform handshake with remote
     async fn handshake(&mut self) -> Result<(), Error> {
         let (result, time) = self.do_handshake_with_prefetch().await;
-        maybe_log(time, &mut self.handshake_latency_sum, &mut self.handshake_latency_count, &result, "handshake");
+        maybe_log(
+            time,
+            &mut self.handshake_latency_sum,
+            &mut self.handshake_latency_count,
+            &result,
+            "handshake",
+        );
         result.map(|_| ())
     }
 
     /// Return a stream of frames to apply to the database
     async fn next_frames(&mut self) -> Result<Self::FrameStream, Error> {
         let (result, time) = self.do_next_frames().await;
-        maybe_log(time, &mut self.frames_latency_sum, &mut self.frames_latency_count, &result, "frames fetch");
+        maybe_log(
+            time,
+            &mut self.frames_latency_sum,
+            &mut self.frames_latency_count,
+            &result,
+            "frames fetch",
+        );
         result
     }
 
     /// Return a snapshot for the current replication index. Called after next_frame has returned a
     /// NeedSnapshot error
     async fn snapshot(&mut self) -> Result<Self::FrameStream, Error> {
         let (snapshot, time) = time(self.do_snapshot()).await;
-        maybe_log(time, &mut self.snapshot_latency_sum, &mut self.snapshot_latency_count, &snapshot, "snapshot fetch");
+        maybe_log(
+            time,
+            &mut self.snapshot_latency_sum,
+            &mut self.snapshot_latency_count,
+            &snapshot,
+            "snapshot fetch",
+        );
         snapshot
     }