libsql: Add Database::sync_until

Signed-off-by: Piotr Jastrzebski <piotr@chiselstrike.com>

Author: haaawk

libsql/src/database.rs

@@ -340,6 +340,16 @@ cfg_replication! {
             }
         }
 
+        /// Sync database from remote until it gets to a given replication_index or further,
+        /// and returns the committed frame_no after syncing, if applicable.
+        pub async fn sync_until(&self, replication_index: FrameNo) -> Result<crate::replication::Replicated> {
+            if let DbType::Sync { db, encryption_config: _ } = &self.db_type {
+                db.sync_until(replication_index).await
+            } else {
+                Err(Error::SyncNotSupported(format!("{:?}", self.db_type)))
+            }
+        }
+
         /// Apply a set of frames to the database and returns the committed frame_no after syncing, if
         /// applicable.
         pub async fn sync_frames(&self, frames: crate::replication::Frames) -> Result<Option<FrameNo>> {

libsql/src/local/database.rs

@@ -277,6 +277,29 @@ impl Database {
         Ok(self.sync_oneshot().await?)
     }
 
+    #[cfg(feature = "replication")]
+    /// Sync with primary at least to a given replication index
+    pub async fn sync_until(&self, replication_index: FrameNo) -> Result<crate::replication::Replicated> {
+        if let Some(ctx) = &self.replication_ctx {
+            let mut frame_no: Option<FrameNo> = ctx.replicator.committed_frame_no().await;
+            let mut frames_synced: usize = 0;
+            while frame_no.unwrap_or(0) < replication_index {
+                let res = ctx.replicator.sync_oneshot().await?;
+                frame_no = res.frame_no();
+                frames_synced += res.frames_synced();
+            }
+            Ok(crate::replication::Replicated {
+                frame_no,
+                frames_synced,
+            })
+        } else {
+            Err(crate::errors::Error::Misuse(
+                "No replicator available. Use Database::with_replicator() to enable replication"
+                    .to_string(),
+            ))
+        }
+    }
+
     #[cfg(feature = "replication")]
     pub async fn sync_frames(&self, frames: Frames) -> Result<Option<FrameNo>> {
         if let Some(ref ctx) = self.replication_ctx {

libsql/src/replication/mod.rs

@@ -36,8 +36,8 @@ pub(crate) mod remote_client;
 
 #[derive(Debug)]
 pub struct Replicated {
-    frame_no: Option<FrameNo>,
-    frames_synced: usize,
+    pub(crate) frame_no: Option<FrameNo>,
+    pub(crate) frames_synced: usize,
 }
 
 impl Replicated {