Go: Reset connection when it's returned to a pool

bindings/c/include/libsql.h

@@ -56,6 +56,8 @@ void libsql_close(libsql_database_t db);
 
 int libsql_connect(libsql_database_t db, libsql_connection_t *out_conn, const char **out_err_msg);
 
+int libsql_reset(libsql_connection_t conn, const char **out_err_msg);
+
 void libsql_disconnect(libsql_connection_t conn);
 
 int libsql_prepare(libsql_connection_t conn, const char *sql, libsql_stmt_t *out_stmt, const char **out_err_msg);

bindings/c/src/lib.rs

@@ -216,6 +216,20 @@ pub unsafe extern "C" fn libsql_connect(
     0
 }
 
+#[no_mangle]
+pub unsafe extern "C" fn libsql_reset(
+    conn: libsql_connection_t,
+    out_err_msg: *mut *const std::ffi::c_char,
+) -> std::ffi::c_int {
+    if conn.is_null() {
+        set_err_msg("Null connection".to_string(), out_err_msg);
+        return 1;
+    }
+    let conn = conn.get_ref();
+    RT.block_on(conn.reset());
+    0
+}
+
 #[no_mangle]
 pub unsafe extern "C" fn libsql_disconnect(conn: libsql_connection_t) {
     if conn.is_null() {
@@ -564,13 +578,13 @@ pub unsafe extern "C" fn libsql_column_type(
 #[no_mangle]
 pub unsafe extern "C" fn libsql_changes(conn: libsql_connection_t) -> u64 {
     let conn = conn.get_ref();
-    conn.changes()
+    RT.block_on(conn.changes())
 }
 
 #[no_mangle]
 pub unsafe extern "C" fn libsql_last_insert_rowid(conn: libsql_connection_t) -> i64 {
     let conn = conn.get_ref();
-    conn.last_insert_rowid()
+    RT.block_on(conn.last_insert_rowid())
 }
 
 #[no_mangle]

libsql-server/tests/hrana/batch.rs

@@ -134,17 +134,17 @@ fn affected_rows_and_last_rowid() {
 
         let r = conn.execute("insert into t(x) values('a');", ()).await?;
         assert_eq!(r, 1, "1st row inserted");
-        assert_eq!(conn.last_insert_rowid(), 1, "1st row id");
+        assert_eq!(conn.last_insert_rowid().await, 1, "1st row id");
 
         let r = conn
             .execute("insert into t(x) values('b'),('c');", ())
             .await?;
         assert_eq!(r, 2, "2nd and 3rd rows inserted");
-        assert_eq!(conn.last_insert_rowid(), 3, "3rd row id");
+        assert_eq!(conn.last_insert_rowid().await, 3, "3rd row id");
 
         let r = conn.execute("update t set x = 'd';", ()).await?;
         assert_eq!(r, 3, "all three rows updated");
-        assert_eq!(conn.last_insert_rowid(), 3, "last row id unchanged");
+        assert_eq!(conn.last_insert_rowid().await, 3, "last row id unchanged");
 
         Ok(())
     });

libsql-server/tests/standalone/mod.rs

@@ -336,26 +336,26 @@ fn is_autocommit() {
         let db = Database::open_remote_with_connector("http://primary:8080", "", TurmoilConnector)?;
         let conn = db.connect()?;
 
-        assert!(conn.is_autocommit());
+        assert!(conn.is_autocommit().await);
         conn.execute("create table test (x)", ()).await?;
 
         conn.execute("begin;", ()).await?;
-        assert!(!conn.is_autocommit());
+        assert!(!conn.is_autocommit().await);
         conn.execute("insert into test values (12);", ()).await?;
         conn.execute("commit;", ()).await?;
-        assert!(conn.is_autocommit());
+        assert!(conn.is_autocommit().await);
 
         // make an explicit transaction
         {
             let tx = conn.transaction().await?;
-            assert!(!tx.is_autocommit());
-            assert!(conn.is_autocommit()); // connection is still autocommit
+            assert!(!tx.is_autocommit().await);
+            assert!(conn.is_autocommit().await); // connection is still autocommit
 
             tx.execute("insert into test values (12);", ()).await?;
             // transaction rolls back
         }
 
-        assert!(conn.is_autocommit());
+        assert!(conn.is_autocommit().await);
 
         let mut rows = conn.query("select count(*) from test", ()).await?;
         assert_eq!(

libsql/src/connection.rs

@@ -16,11 +16,13 @@ pub(crate) trait Conn {
 
     async fn transaction(&self, tx_behavior: TransactionBehavior) -> Result<Transaction>;
 
-    fn is_autocommit(&self) -> bool;
+    async fn is_autocommit(&self) -> bool;
 
-    fn changes(&self) -> u64;
+    async fn changes(&self) -> u64;
 
-    fn last_insert_rowid(&self) -> i64;
+    async fn last_insert_rowid(&self) -> i64;
+
+    async fn reset(&self);
 }
 
 /// A connection to some libsql database, this can be a remote one or a local one.
@@ -98,17 +100,21 @@ impl Connection {
     }
 
     /// Check weather libsql is in `autocommit` or not.
-    pub fn is_autocommit(&self) -> bool {
-        self.conn.is_autocommit()
+    pub async fn is_autocommit(&self) -> bool {
+        self.conn.is_autocommit().await
     }
 
     /// Check the amount of changes the last query created.
-    pub fn changes(&self) -> u64 {
-        self.conn.changes()
+    pub async fn changes(&self) -> u64 {
+        self.conn.changes().await
     }
 
     /// Check the last inserted row id.
-    pub fn last_insert_rowid(&self) -> i64 {
-        self.conn.last_insert_rowid()
+    pub async fn last_insert_rowid(&self) -> i64 {
+        self.conn.last_insert_rowid().await
+    }
+
+    pub async fn reset(&self) {
+        self.conn.reset().await
     }
 }

libsql/src/hrana/hyper.rs

@@ -146,17 +146,21 @@ impl Conn for HttpConnection<HttpSender> {
         })
     }
 
-    fn is_autocommit(&self) -> bool {
+    async fn is_autocommit(&self) -> bool {
         self.is_autocommit()
     }
 
-    fn changes(&self) -> u64 {
+    async fn changes(&self) -> u64 {
         self.affected_row_count()
     }
 
-    fn last_insert_rowid(&self) -> i64 {
+    async fn last_insert_rowid(&self) -> i64 {
         self.last_insert_rowid()
     }
+
+    async fn reset(&self) {
+        self.current_stream().reset().await;
+    }
 }
 
 #[async_trait::async_trait]
@@ -273,15 +277,19 @@ impl Conn for HranaStream<HttpSender> {
         todo!("sounds like nested transactions innit?")
     }
 
-    fn is_autocommit(&self) -> bool {
+    async fn is_autocommit(&self) -> bool {
         false // for streams this method is callable only when we're within explicit transaction
     }
 
-    fn changes(&self) -> u64 {
+    async fn changes(&self) -> u64 {
         self.affected_row_count()
     }
 
-    fn last_insert_rowid(&self) -> i64 {
+    async fn last_insert_rowid(&self) -> i64 {
         self.last_insert_rowid()
     }
+
+    async fn reset(&self) {
+        self.reset().await;
+    }
 }

libsql/src/hrana/stream.rs

@@ -266,6 +266,10 @@ where
     pub fn is_autocommit(&self) -> bool {
         self.inner.is_autocommit.load(Ordering::SeqCst)
     }
+
+    pub async fn reset(&self) {
+        (*self.inner).stream.lock().await.reset();
+    }
 }
 
 #[derive(Debug)]

libsql/src/local/impls.rs

@@ -48,17 +48,19 @@ impl Conn for LibsqlConnection {
         })
     }
 
-    fn is_autocommit(&self) -> bool {
+    async fn is_autocommit(&self) -> bool {
         self.conn.is_autocommit()
     }
 
-    fn changes(&self) -> u64 {
+    async fn changes(&self) -> u64 {
         self.conn.changes()
     }
 
-    fn last_insert_rowid(&self) -> i64 {
+    async fn last_insert_rowid(&self) -> i64 {
         self.conn.last_insert_rowid()
     }
+
+    async fn reset(&self) {}
 }
 
 impl Drop for LibsqlConnection {

libsql/src/replication/connection.rs

@@ -232,7 +232,7 @@ impl RemoteConnection {
     // and will return false if no rollback happened and the
     // execute was valid.
     pub(self) async fn maybe_execute_rollback(&self) -> Result<bool> {
-        if self.inner.lock().state != State::TxnReadOnly && !self.local.is_autocommit() {
+        if self.inner.lock().state != State::TxnReadOnly && !self.local.is_autocommit().await {
             self.local.execute("ROLLBACK", Params::None).await?;
             Ok(true)
         } else {
@@ -332,17 +332,19 @@ impl Conn for RemoteConnection {
         })
     }
 
-    fn is_autocommit(&self) -> bool {
+    async fn is_autocommit(&self) -> bool {
         self.is_state_init()
     }
 
-    fn changes(&self) -> u64 {
+    async fn changes(&self) -> u64 {
         self.inner.lock().changes
     }
 
-    fn last_insert_rowid(&self) -> i64 {
+    async fn last_insert_rowid(&self) -> i64 {
         self.inner.lock().last_insert_rowid
     }
+
+    async fn reset(&self) {}
 }
 
 pub struct ColumnMeta {