Merge pull request tursodatabase#1885 from tursodatabase/interrupt

libsql: Add support for Connection::interrupt()

Author: penberg

libsql/src/connection.rs

@@ -21,6 +21,8 @@ pub(crate) trait Conn {
 
     async fn transaction(&self, tx_behavior: TransactionBehavior) -> Result<Transaction>;
 
+    fn interrupt(&self) -> Result<()>;
+
     fn is_autocommit(&self) -> bool;
 
     fn changes(&self) -> u64;
@@ -185,6 +187,11 @@ impl Connection {
         self.conn.transaction(tx_behavior).await
     }
 
+    /// Cancel ongoing operations and return at earliest opportunity.
+    pub fn interrupt(&self) -> Result<()> {
+        self.conn.interrupt()
+    }
+
     /// Check weather libsql is in `autocommit` or not.
     pub fn is_autocommit(&self) -> bool {
         self.conn.is_autocommit()

libsql/src/hrana/hyper.rs

@@ -163,6 +163,11 @@ impl Conn for HttpConnection<HttpSender> {
         })
     }
 
+    fn interrupt(&self) -> crate::Result<()> {
+        // Interrupt is a no-op for remote connections.
+        Ok(())
+    }
+
     fn is_autocommit(&self) -> bool {
         self.is_autocommit()
     }
@@ -343,6 +348,11 @@ impl Conn for HranaStream<HttpSender> {
         todo!("sounds like nested transactions innit?")
     }
 
+    fn interrupt(&self) -> crate::Result<()> {
+        // Interrupt is a no-op for remote connections.
+        Ok(())
+    }
+
     fn is_autocommit(&self) -> bool {
         false // for streams this method is callable only when we're within explicit transaction
     }

libsql/src/local/connection.rs

@@ -38,14 +38,20 @@ impl Connection {
     pub(crate) fn connect(db: &Database) -> Result<Connection> {
         let mut raw = std::ptr::null_mut();
         let db_path = db.db_path.clone();
+        let flags: c_int = ffi::SQLITE_OPEN_FULLMUTEX | db.flags.bits();
+
         let err = unsafe {
+            if ffi::sqlite3_threadsafe() == 0 {
+                return Err(Error::ConnectionFailed(String::from("sqlite3 was compiled with SQLITE_THREADSAFE=0 and must be compiled with SQLITE_THREADSAFE=1 or SQLITE_THREADSAFE=2.")));
+            }
+
             ffi::sqlite3_open_v2(
                 std::ffi::CString::new(db_path.as_str())
                     .unwrap()
                     .as_c_str()
                     .as_ptr() as *const _,
                 &mut raw,
-                db.flags.bits() as c_int,
+                flags,
                 std::ptr::null(),
             )
         };
@@ -355,6 +361,11 @@ impl Connection {
         Transaction::begin(self.clone(), tx_behavior)
     }
 
+    pub fn interrupt(&self) -> Result<()> {
+        unsafe { ffi::sqlite3_interrupt(self.raw) };
+        Ok(())
+    }
+
     pub fn is_autocommit(&self) -> bool {
         unsafe { ffi::sqlite3_get_autocommit(self.raw) != 0 }
     }

libsql/src/local/database.rs

@@ -1,5 +1,3 @@
-use std::sync::Once;
-
 cfg_replication!(
     use http::uri::InvalidUri;
     use crate::database::{EncryptionConfig, FrameNo};
@@ -24,7 +22,6 @@ cfg_sync! {
 
 use crate::{database::OpenFlags, local::connection::Connection};
 use crate::{Error::ConnectionFailed, Result};
-use libsql_sys::ffi;
 
 // A libSQL database.
 pub struct Database {
@@ -230,29 +227,6 @@ impl Database {
     }
 
     pub fn new(db_path: String, flags: OpenFlags) -> Database {
-        static LIBSQL_INIT: Once = Once::new();
-
-        LIBSQL_INIT.call_once(|| {
-            // Ensure that we are configured with the correct threading model
-            // if this config is not set correctly the entire api is unsafe.
-            unsafe {
-                assert_eq!(
-                    ffi::sqlite3_config(ffi::SQLITE_CONFIG_SERIALIZED),
-                    ffi::SQLITE_OK,
-                    "libsql was configured with an incorrect threading configuration and
-                    the api is not safe to use. Please check that no multi-thread options have
-                    been set. If nothing was configured then please open an issue at:
-                    https://github.com/libsql/libsql"
-                );
-
-                assert_eq!(
-                    ffi::sqlite3_initialize(),
-                    ffi::SQLITE_OK,
-                    "libsql failed to initialize"
-                );
-            }
-        });
-
         Database {
             db_path,
             flags,
@@ -420,7 +394,11 @@ impl Database {
     }
 
     #[cfg(feature = "sync")]
-    async fn try_push(&self, sync_ctx: &mut SyncContext, conn: &Connection) -> Result<crate::database::Replicated> {
+    async fn try_push(
+        &self,
+        sync_ctx: &mut SyncContext,
+        conn: &Connection,
+    ) -> Result<crate::database::Replicated> {
         let page_size = {
             let rows = conn
                 .query("PRAGMA page_size", crate::params::Params::None)?
@@ -471,7 +449,11 @@ impl Database {
     }
 
     #[cfg(feature = "sync")]
-    async fn try_pull(&self, sync_ctx: &mut SyncContext, conn: &Connection) -> Result<crate::database::Replicated> {
+    async fn try_pull(
+        &self,
+        sync_ctx: &mut SyncContext,
+        conn: &Connection,
+    ) -> Result<crate::database::Replicated> {
         let generation = sync_ctx.generation();
         let mut frame_no = sync_ctx.durable_frame_num() + 1;
         conn.wal_insert_begin()?;

libsql/src/local/impls.rs

@@ -54,6 +54,10 @@ impl Conn for LibsqlConnection {
         })
     }
 
+    fn interrupt(&self) -> Result<()> {
+        self.conn.interrupt()
+    }
+
     fn is_autocommit(&self) -> bool {
         self.conn.is_autocommit()
     }

libsql/src/replication/connection.rs

@@ -503,6 +503,11 @@ impl Conn for RemoteConnection {
         })
     }
 
+    fn interrupt(&self) -> Result<()> {
+        // Interrupt is a no-op for remote connections.
+        Ok(())
+    }
+
     fn is_autocommit(&self) -> bool {
         self.is_state_init()
     }