Feature: Add Connection::snapshot() for point-in-time database copy

This addresses feedback from tursodatabase/agentfs#119 suggesting that snapshot functionality should be implemented in turso.git using direct fily copying rather than SQL-based row by row copying.
Comment: https:://github.com/tursodatabase/agentfs/pull/119#issuecomment-3681336678

- extract core logic from Pager::checkpoint function to a new Pager::checkpoing_internal function and add a flag to keep_lock during the Finalize phase
- create wrapper functions Pager::checkpoint, Pager::checkpoint_with_lock and Pager::block_checkpoint_keep_lock
- add Connection::snapshot API that checkpoints while keeping the lock and copies the database file

The lock is held to avoid a race condition after finishing checkpointing and before copying the file when concurrent writers can write to db.

Signed-off-by: Prateek Singh Rathore <[email protected]>

Author: psrvere

core/lib.rs

@@ -2017,6 +2017,67 @@ impl Connection {
         }
     }
 
+    /// Create a point-in-time snapshot of the database by copying the database file.
+    #[cfg(feature = "fs")]
+    pub fn snapshot(self: &Arc<Self>, output_path: &str) -> Result<()> {
+        self._snapshot_copy_file(output_path)
+    }
+
+    /// Create a point-in-time snapshot of the database by copying the database file.
+    ///
+    /// It checkpoints and keeps the lock after checkpointing to avoid race window between
+    /// checkpoint completion and file copy operation where concurrent writers could modify
+    /// the database.
+    #[cfg(feature = "fs")]
+    pub fn _snapshot_copy_file(self: &Arc<Self>, output_path: &str) -> Result<()> {
+        use crate::util::MEMORY_PATH;
+        use std::path::Path;
+
+        if self.is_closed() {
+            return Err(LimboError::InternalError("Connection closed".to_string()));
+        }
+
+        // FIXME: enable mvcc
+        if self.mvcc_enabled() {
+            return Err(LimboError::InternalError("Snapshot not yet supported with MVCC mode".to_string()))
+        }
+
+        // Cannot snapshot in-memory databases
+        if self.db.path == MEMORY_PATH {
+            return Err(LimboError::InvalidArgument(
+                "Cannot snapshot in-memory database".to_string(),
+            ));
+        }
+
+        let path = Path::new(output_path);
+        if path.exists() {
+            return Err(LimboError::InvalidArgument(format!(
+                "Output file already exists: {}",
+                output_path
+            )));
+        }
+        if let Some(parent) = path.parent() {
+            std::fs::create_dir_all(parent).map_err(|e| {
+                LimboError::InternalError(format!("Failed to create parent directory: {}", e))
+            })?;
+        }
+
+        let pager = self.pager.load();
+        let result = (|| -> Result<()> {
+            // Checkpoint and keep the lock
+            let _res = pager.blocking_checkpoint_keep_lock(CheckpointMode::Truncate { upper_bound_inclusive: None }, self.get_sync_mode())?;
+
+            let source_path = Path::new(&self.db.path);
+            std::fs::copy(source_path, path).map_err(|e| {
+                LimboError::InternalError(format!("Failed to copy database file: {}", e))
+            })?;
+
+            Ok(()) // lock is dropped here when _res is dropped here
+        })();
+
+        result
+    }
+
     /// Close a connection and checkpoint.
     pub fn close(&self) -> Result<()> {
         if self.is_closed() {

core/storage/pager.rs

@@ -2856,17 +2856,40 @@ impl Pager {
         }
     }
 
+    #[instrument(skip_all, level = Level::DEBUG)]
+    /// checkpoint_keep_lock checkpoints the WAL to the database file (if needed) and keeps the lock
+    pub fn checkpoint_keep_lock(
+        &self,
+        mode: CheckpointMode,
+        sync_mode: crate::SyncMode,
+        clear_page_cache: bool,
+    ) -> Result<IOResult<CheckpointResult>> {
+        self.checkpoint_inner(mode, sync_mode, clear_page_cache, true)
+    }
+
+    #[instrument(skip_all, level = Level::DEBUG)]
+    /// checkpoint checkpoints the WAL to the database file (if needed) and drops the lock
+    pub fn checkpoint(
+        &self,
+        mode: CheckpointMode,
+        sync_mode: crate::SyncMode,
+        clear_page_cache: bool,
+    ) -> Result<IOResult<CheckpointResult>> {
+        self.checkpoint_inner(mode, sync_mode, clear_page_cache, false)
+    }
+
     #[instrument(skip_all, level = Level::DEBUG, name = "pager_checkpoint",)]
     /// Checkpoint the WAL to the database file (if needed).
     /// Args:
     /// - mode: The checkpoint mode to use (PASSIVE, FULL, RESTART, TRUNCATE)
     /// - sync_mode: The fsync mode to use (OFF, NORMAL, FULL)
     /// - clear_page_cache: Whether to clear the page cache after checkpointing
-    pub fn checkpoint(
+    fn checkpoint_inner(
         &self,
         mode: CheckpointMode,
         sync_mode: crate::SyncMode,
         clear_page_cache: bool,
+        keep_lock: bool,
     ) -> Result<IOResult<CheckpointResult>> {
         let Some(wal) = self.wal.as_ref() else {
             return Err(LimboError::InternalError(
@@ -2994,8 +3017,10 @@ impl Pager {
                     let mut state = self.checkpoint_state.write();
                     let mut res = state.result.take().expect("result should be set");
 
-                    // Release checkpoint guard
-                    res.release_guard();
+                    // Release checkpoint guard if lock is not to be kept
+                    if !keep_lock {
+                        res.release_guard();
+                    }
 
                     // Clear page cache only if requested (explicit checkpoints do this, auto-checkpoint does not)
                     if clear_page_cache {
@@ -3092,6 +3117,15 @@ impl Pager {
         self.io.block(|| self.checkpoint(mode, sync_mode, true))
     }
 
+    #[instrument(skip_all, level = Level::DEBUG)]
+    pub fn blocking_checkpoint_keep_lock(
+        &self,
+        mode: CheckpointMode,
+        sync_mode: crate::SyncMode,
+    ) -> Result<CheckpointResult> {
+        self.io.block(|| self.checkpoint_keep_lock(mode, sync_mode, true))
+    }
+
     pub fn freepage_list(&self) -> u32 {
         self.io
             .block(|| HeaderRefMut::from_pager(self))