expose bottomless backup savepoint through PrimaryDatabase API

bottomless/src/bottomless_wal.rs

@@ -1,6 +1,7 @@
 use std::ffi::c_int;
 use std::sync::{Arc, Mutex};
 
+use crate::completion_progress::SavepointTracker;
 use libsql_sys::ffi::{SQLITE_BUSY, SQLITE_IOERR_WRITE};
 use libsql_sys::wal::wrapper::{WalWrapper, WrapWal};
 use libsql_sys::wal::{
@@ -29,6 +30,14 @@ impl BottomlessWalWrapper {
         }
     }
 
+    pub fn backup_savepoint(&self) -> Option<SavepointTracker> {
+        let lock = self.replicator.lock().unwrap();
+        match &*lock {
+            None => None,
+            Some(replicator) => Some(replicator.savepoint()),
+        }
+    }
+
     pub fn shutdown(&self) -> Option<Replicator> {
         self.replicator.lock().unwrap().take()
     }

bottomless/src/completion_progress.rs

@@ -11,9 +11,9 @@ use uuid::Uuid;
 pub struct SavepointTracker {
     next_frame_no: Arc<AtomicU32>,
     receiver: Receiver<u32>,
-    pub generation: Arc<ArcSwapOption<Uuid>>,
-    pub generation_snapshot: Receiver<Result<Option<Uuid>>>,
-    pub db_path: String,
+    generation: Arc<ArcSwapOption<Uuid>>,
+    generation_snapshot: Receiver<Result<Option<Uuid>>>,
+    db_path: String,
 }
 
 impl SavepointTracker {
@@ -41,7 +41,7 @@ impl SavepointTracker {
                     .wait_for(|gen| match gen {
                         Ok(Some(gen)) => gen == &*generation,
                         Ok(None) => false,
-                        Err(e) => true,
+                        Err(_) => true,
                     })
                     .await?;
                 return match &*res {
@@ -98,8 +98,4 @@ impl CompletionProgress {
             self.detached_ranges.insert(start_frame, end_frame);
         }
     }
-
-    pub(crate) fn subscribe(&mut self) -> Receiver<u32> {
-        self.tx.subscribe()
-    }
 }

libsql-server/src/database.rs

@@ -21,6 +21,11 @@ pub trait Database: Sync + Send + 'static {
     fn destroy(self);
 
     async fn shutdown(self) -> Result<()>;
+
+    /// Snapshots a DB state at the moment of a call and awaits for a backup service
+    /// to complete saving that state. Any changes happening concurrently are not
+    /// guaranteed to be part of confirmation.
+    async fn backup_savepoint(&self) -> Result<()>;
 }
 
 pub struct ReplicaDatabase {
@@ -41,6 +46,10 @@ impl Database for ReplicaDatabase {
     async fn shutdown(self) -> Result<()> {
         Ok(())
     }
+
+    async fn backup_savepoint(&self) -> Result<()> {
+        Ok(()) // backup doesn't happen on replicas
+    }
 }
 
 pub struct PrimaryDatabase {
@@ -81,4 +90,13 @@ impl Database for PrimaryDatabase {
 
         Ok(())
     }
+
+    async fn backup_savepoint(&self) -> Result<()> {
+        if let Some(wal) = self.wal_manager.wrapper() {
+            if let Some(mut tracker) = wal.backup_savepoint() {
+                tracker.confirmed().await?;
+            }
+        }
+        Ok(())
+    }
 }