libsql: add local offline writes metadata file

This adds a new `<database>-info` file that is a json file containing
metadata for syncing databases. At this point it just contains the
`max_frame_no` from the server to optimize sending wal frames that
already exist on the server.

Closes #1818

Author: LucioFranco

libsql/src/local/database.rs

@@ -144,11 +144,11 @@ impl Database {
             endpoint
         };
         let mut db = Database::open(&db_path, flags)?;
-        db.sync_ctx = Some(tokio::sync::Mutex::new(SyncContext::new(
-            connector,
-            endpoint,
-            Some(auth_token),
-        )));
+
+        let sync_ctx =
+            SyncContext::new(connector, db_path.into(), endpoint, Some(auth_token)).await?;
+        db.sync_ctx = Some(tokio::sync::Mutex::new(sync_ctx));
+
         Ok(db)
     }
 
@@ -388,7 +388,7 @@ impl Database {
     #[cfg(feature = "sync")]
     /// Push WAL frames to remote.
     pub async fn push(&self) -> Result<crate::database::Replicated> {
-        let sync_ctx = self.sync_ctx.as_ref().unwrap().lock().await;
+        let mut sync_ctx = self.sync_ctx.as_ref().unwrap().lock().await;
         let conn = self.connect()?;
 
         let page_size = {

libsql/src/sync.rs

@@ -2,32 +2,45 @@ use crate::{util::ConnectorService, Result};
 use bytes::Bytes;
 use hyper::Body;
 
+const METADATA_VERSION: u32 = 0;
+
 const DEFAULT_MAX_RETRIES: usize = 5;
 
 pub struct SyncContext {
+    db_path: String,
     sync_url: String,
     auth_token: Option<String>,
     max_retries: usize,
+    /// Represents the max_frame_no from the server.
     durable_frame_num: u32,
     client: hyper::Client<ConnectorService, Body>,
 }
 
 impl SyncContext {
-    pub fn new(connector: ConnectorService, sync_url: String, auth_token: Option<String>) -> Self {
-        // TODO(lucio): add custom connector + tls support here
+    pub async fn new(
+        connector: ConnectorService,
+        db_path: String,
+        sync_url: String,
+        auth_token: Option<String>,
+    ) -> Result<Self> {
         let client = hyper::client::Client::builder().build::<_, hyper::Body>(connector);
 
-        Self {
+        let mut me = Self {
+            db_path,
             sync_url,
             auth_token,
             durable_frame_num: 0,
             max_retries: DEFAULT_MAX_RETRIES,
             client,
-        }
+        };
+
+        me.read_metadata().await?;
+
+        Ok(me)
     }
 
     pub(crate) async fn push_one_frame(
-        &self,
+        &mut self,
         frame: Bytes,
         generation: u32,
         frame_no: u32,
@@ -41,6 +54,11 @@ impl SyncContext {
         );
         let max_frame_no = self.push_with_retry(uri, frame, self.max_retries).await?;
 
+        // Update our last known max_frame_no from the server.
+        self.durable_frame_num = max_frame_no;
+
+        self.write_metadata().await?;
+
         Ok(max_frame_no)
     }
 
@@ -93,4 +111,41 @@ impl SyncContext {
     pub(crate) fn durable_frame_num(&self) -> u32 {
         self.durable_frame_num
     }
+
+    async fn write_metadata(&mut self) -> Result<()> {
+        let path = format!("{}-info", self.db_path);
+
+        let contents = serde_json::to_vec(&MetadataJson {
+            version: METADATA_VERSION,
+            durable_frame_num: self.durable_frame_num,
+        })
+        .unwrap();
+
+        tokio::fs::write(path, contents).await.unwrap();
+
+        Ok(())
+    }
+
+    async fn read_metadata(&mut self) -> Result<()> {
+        let path = format!("{}-info", self.db_path);
+
+        let contents = tokio::fs::read(&path).await.unwrap();
+
+        let metadata = serde_json::from_slice::<MetadataJson>(&contents[..]).unwrap();
+
+        assert_eq!(
+            metadata.version, METADATA_VERSION,
+            "Reading metadata from a different version than expected"
+        );
+
+        self.durable_frame_num = metadata.durable_frame_num;
+
+        Ok(())
+    }
+}
+
+#[derive(serde::Serialize, serde::Deserialize)]
+struct MetadataJson {
+    version: u32,
+    durable_frame_num: u32,
 }