Merge pull request #653 from Horusiath/libsql-transactions

libsql: remote hrana transactions

Author: LucioFranco

libsql-server/tests/hrana/mod.rs

@@ -5,6 +5,7 @@ use tempfile::tempdir;
 
 use crate::common::net::{init_tracing, SimServer, TestServer};
 mod batch;
+mod transaction;
 
 async fn make_standalone_server() -> Result<(), Box<dyn std::error::Error>> {
     init_tracing();

libsql-server/tests/hrana/transaction.rs

@@ -0,0 +1,98 @@
+use crate::common::net::TurmoilConnector;
+use libsql::{params, Database, TransactionBehavior};
+
+#[test]
+fn transaction_commit_and_rollback() {
+    let mut sim = turmoil::Builder::new().build();
+    sim.host("primary", super::make_standalone_server);
+    sim.client("client", async {
+        let db = Database::open_remote_with_connector("http://primary:8080", "", TurmoilConnector)?;
+        let conn = db.connect()?;
+
+        // initialize tables
+        let tx = conn.transaction().await?;
+        tx.execute_batch(r#"create table t(x text);"#).await?;
+        tx.commit().await?;
+
+        // transaction with temporary data
+        let tx = conn.transaction().await?;
+        tx.execute("insert into t(x) values('hello');", ()).await?;
+
+        let mut rows = tx
+            .query("select * from t where x = ?", params!["hello"])
+            .await?;
+
+        assert_eq!(rows.column_count(), 1);
+        assert_eq!(rows.column_name(0), Some("x"));
+        assert_eq!(rows.next()?.unwrap().get::<String>(0)?, "hello");
+        assert!(rows.next()?.is_none());
+        tx.rollback().await?;
+
+        // confirm that temporary that was not committed
+        let mut rows = conn
+            .query("select * from t where x = ?", params!["hello"])
+            .await?;
+
+        assert_eq!(rows.column_count(), 1);
+        assert_eq!(rows.column_name(0), Some("x"));
+        assert!(rows.next()?.is_none());
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}
+
+#[test]
+fn multiple_concurrent_transactions() {
+    let mut sim = turmoil::Builder::new().build();
+    sim.host("primary", super::make_standalone_server);
+    sim.client("client", async {
+        let db = Database::open_remote_with_connector("http://primary:8080", "", TurmoilConnector)?;
+        let conn = db.connect()?;
+        conn.execute_batch(r#"create table t(x text);"#).await?;
+
+        // open first transaction and alter data
+        let tx1 = conn
+            .transaction_with_behavior(TransactionBehavior::Deferred)
+            .await?;
+        tx1.execute("insert into t(x) values('hello');", ()).await?;
+
+        // while first transaction is still open open another read-only transaction and try to read
+        let tx2 = conn
+            .transaction_with_behavior(TransactionBehavior::ReadOnly)
+            .await?;
+        let mut rows = tx2
+            .query("select * from t where x = ?", params!["hello"])
+            .await?;
+        assert_eq!(rows.column_count(), 1);
+        assert_eq!(rows.column_name(0), Some("x"));
+        assert!(rows.next()?.is_none());
+
+        // commit first transaction - T2 should still read old data
+        tx1.commit().await?;
+
+        let mut rows = tx2
+            .query("select * from t where x = ?", params!["hello"])
+            .await?;
+        assert_eq!(rows.column_count(), 1);
+        assert_eq!(rows.column_name(0), Some("x"));
+        assert!(rows.next()?.is_none());
+        tx2.commit().await?;
+
+        // finally open new transaction - it now should read actual data
+        let tx3 = conn
+            .transaction_with_behavior(TransactionBehavior::ReadOnly)
+            .await?;
+        let mut rows = tx3
+            .query("select * from t where x = ?", params!["hello"])
+            .await?;
+        assert_eq!(rows.column_count(), 1);
+        assert_eq!(rows.column_name(0), Some("x"));
+        assert_eq!(rows.next()?.unwrap().get::<String>(0)?, "hello");
+        assert!(rows.next()?.is_none());
+
+        Ok(())
+    });
+    sim.run().unwrap();
+}

libsql/Cargo.toml

@@ -88,6 +88,7 @@ hrana = [
   "serde",
   "dep:base64",
   "dep:serde_json",
+  "dep:futures"
 ]
 serde = ["dep:serde"]
 remote = [

libsql/src/connection.rs

@@ -16,13 +16,11 @@ pub(crate) trait Conn {
 
     async fn transaction(&self, tx_behavior: TransactionBehavior) -> Result<Transaction>;
 
-    fn is_autocommit(&self) -> bool;
+    async fn is_autocommit(&self) -> Result<bool>;
 
     fn changes(&self) -> u64;
 
     fn last_insert_rowid(&self) -> i64;
-
-    fn close(&mut self);
 }
 
 #[derive(Clone)]
@@ -69,8 +67,8 @@ impl Connection {
         self.conn.transaction(tx_behavior).await
     }
 
-    pub fn is_autocommit(&self) -> bool {
-        self.conn.is_autocommit()
+    pub async fn is_autocommit(&self) -> Result<bool> {
+        self.conn.is_autocommit().await
     }
 
     pub fn changes(&self) -> u64 {
@@ -80,10 +78,4 @@ impl Connection {
     pub fn last_insert_rowid(&self) -> i64 {
         self.conn.last_insert_rowid()
     }
-
-    pub fn close(&mut self) {
-        if let Some(conn) = Arc::get_mut(&mut self.conn) {
-            conn.close()
-        }
-    }
 }