do not open wal session from the beginning - wait for pulled frames

- otherwise, if client is on the checkpoint boundary and immediately get
  checkpoint command - we will crash with assertion

Author: sivukhin

libsql/src/local/connection.rs

@@ -480,7 +480,7 @@ impl Connection {
                 let callback = authorizer_callback as unsafe extern "C" fn(_, _, _, _, _, _) -> _;
                 let user_data = self as *const Connection as *mut ::std::os::raw::c_void;
                 (Some(callback), user_data)
-            },
+            }
             None => (None, std::ptr::null_mut()),
         };
 
@@ -630,9 +630,11 @@ impl Connection {
         Ok(())
     }
 
-    pub(crate) fn wal_insert_handle(&self) -> Result<WalInsertHandle<'_>> {
-        self.wal_insert_begin()?;
-        Ok(WalInsertHandle { conn: self, in_session: RefCell::new(true) })
+    pub(crate) fn wal_insert_handle(&self) -> WalInsertHandle<'_> {
+        WalInsertHandle {
+            conn: self,
+            in_session: RefCell::new(false),
+        }
     }
 }
 
@@ -768,7 +770,9 @@ mod tests {
                 )
                 .unwrap();
         }
-        let handle = conn2.wal_insert_handle().unwrap();
+        let handle = conn2.wal_insert_handle();
+        handle.begin().unwrap();
+
         let frame_count = conn1.wal_frame_count();
         for frame_no in 0..frame_count {
             let frame = conn1.wal_get_frame(frame_no + 1, 4096).unwrap();

libsql/src/sync.rs

@@ -961,7 +961,7 @@ pub async fn try_pull(
 
     // libsql maintain consistent state about WAL sync session locally in the insert_handle
     // note, that insert_handle will always close the session on drop - so we never keep active WAL session after we exit from the method
-    let insert_handle = conn.wal_insert_handle()?;
+    let insert_handle = conn.wal_insert_handle();
 
     loop {
         // get current generation (it may be updated multiple times during execution)
@@ -990,8 +990,7 @@ pub async fn try_pull(
                     if !insert_handle.in_session() {
                         tracing::debug!(
                             "pull_frames: generation={}, frame={}, start wal transaction session",
-                            generation,
-                            next_frame_no
+                            generation, next_frame_no
                         );
                         insert_handle.begin()?;
                     }