Repository: tursodatabase/libsql (c)

Issue:
Concurrent transactions on embedded replica connections fail
Facing an issue where multiple concurrent transactions started on separate embedded replica connections fail, with only one of them succeeding. Others fail with error `Err(RemoteSqliteFailure(3, 1, "cannot start a transaction within a transaction"))`

Tested on remote connections and the scenario works fine there.

Below is a reproducer for this:
```rust
use libsql::{Builder, Connection, Result};

#[tokio::main]
async fn main() {
    let db_url = "http://localhost:8080";
    let replica = Builder::new_remote_replica(
        "/tmp/embedded_transaction.db",
        db_url.to_string(),
        String::new(),
    )
    .build()
    .await
    .unwrap();
    let remote = Builder::new_remote(db_url.to_string(), String::new())
        .build()
        .await
        .unwrap();
    let replica_conn_1 = replica.connect().unwrap();
    let replica_conn_2 = replica.connect().unwrap();

    let remote_conn_1 = remote.connect().unwrap();
    let remote_conn_2 = remote.connect().unwrap();

    let remote_task_1 = tokio::task::spawn(async move { db_work(remote_conn_1).await });
    let remote_task_2 = tokio::task::spawn(async move { db_work(remote_conn_2).await });

    let (task_1_res, task_2_res) = tokio::join!(remote_task_1, remote_task_2);
    let remote_task_1_res = task_1_res.unwrap();
    let remote_task_2_res = task_2_res.unwrap();

    // Everything works as expected in case of remote connections.
    assert!(remote_task_1_res.is_ok());
    assert!(remote_task_2_res.is_ok());

    let replica_task_1 = tokio::task::spawn(async move { db_work(replica_conn_1).await });
    let replica_task_2 = tokio::task::spawn(async move { db_work(replica_conn_2).await });

    let (task_1_res, task_2_res) = tokio::join!(replica_task_1, replica_task_2);
    let replica_task_1_res = task_1_res.unwrap();
    let replica_task_2_res = task_2_res.unwrap();

    if replica_task_1_res.is_err() {
        eprintln!("Task 1 failed: {:?}", replica_task_1_res);
    }
    if replica_task_2_res.is_err() {
        eprintln!("Task 2 failed: {:?}", replica_task_2_res);
    }

    // One of these concurrent tasks fail currently. Both tasks should succeed.
    assert!(replica_task_1_res.is_ok());
    assert!(replica_task_2_res.is_ok());
}

async fn db_work(conn: Connection) -> Result<()> {
    let tx = conn.transaction().await?;
    // Some business logic here...
    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
    tx.execute("SELECT 1", ()).await?;
    tx.commit().await?;
    Ok(())
}

```

Pull request description:
libsql: fix embedded replica multiple txn
This commit fixes an issue where new connections for embedded replica's with write delegation would not create a new "connection" on the server causing weird issues with transactions and state. The fix here is to create a new `client_id` when creating a new writer for a new connection locally. This forces the server to use separate connections to execute the writes.

Closes #1283

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index ad6ea1d1a0..0019e6a7a0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2791,7 +2791,7 @@ dependencies = [
 
 [[package]]
 name = "libsql-server"
-version = "0.24.4"
+version = "0.24.5"
 dependencies = [
  "aes",
  "anyhow",
diff --git a/libsql-server/Cargo.toml b/libsql-server/Cargo.toml
index 7de7d65fe6..afedc154fd 100644
--- a/libsql-server/Cargo.toml
+++ b/libsql-server/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "libsql-server"
-version = "0.24.4"
+version = "0.24.5"
 edition = "2021"
 default-run = "sqld"
 
diff --git a/libsql/src/database.rs b/libsql/src/database.rs
index dd5ebc671f..f1fe3e755b 100644
--- a/libsql/src/database.rs
+++ b/libsql/src/database.rs
@@ -549,7 +549,7 @@ impl Database {
                 }
 
                 let local = LibsqlConnection { conn };
-                let writer = local.conn.writer().cloned();
+                let writer = local.conn.new_connection_writer();
                 let remote = crate::replication::RemoteConnection::new(local, writer);
                 let conn = std::sync::Arc::new(remote);
 
diff --git a/libsql/src/local/connection.rs b/libsql/src/local/connection.rs
index e2e72c2af6..9d568ca736 100644
--- a/libsql/src/local/connection.rs
+++ b/libsql/src/local/connection.rs
@@ -226,6 +226,14 @@ impl Connection {
     pub(crate) fn writer(&self) -> Option<&crate::replication::Writer> {
         self.writer.as_ref()
     }
+
+    #[cfg(feature = "replication")]
+    pub(crate) fn new_connection_writer(&self) -> Option<crate::replication::Writer> {
+        self.writer.as_ref().cloned().map(|mut w| {
+            w.new_client_id();
+            w
+        })
+    }
 }
 
 impl fmt::Debug for Connection {
diff --git a/libsql/src/replication/client.rs b/libsql/src/replication/client.rs
index 1357ba76e8..6e86413ed8 100644
--- a/libsql/src/replication/client.rs
+++ b/libsql/src/replication/client.rs
@@ -90,6 +90,10 @@ impl Client {
         })
     }
 
+    pub fn new_client_id(&mut self) {
+        self.client_id = Uuid::new_v4();
+    }
+
     pub fn client_id(&self) -> String {
         self.client_id.to_string()
     }
@@ -124,7 +128,10 @@ impl GrpcChannel {
         connector: ConnectorService,
         http_request_callback: Option<HttpRequestCallback>,
     ) -> Self {
-        let client = hyper::Client::builder().pool_idle_timeout(None).pool_max_idle_per_host(3).build(connector);
+        let client = hyper::Client::builder()
+            .pool_idle_timeout(None)
+            .pool_max_idle_per_host(3)
+            .build(connector);
         let client = GrpcWebClientService::new(client);
 
         let classifier = GrpcErrorsAsFailures::new().with_success(GrpcCode::FailedPrecondition);
diff --git a/libsql/src/replication/mod.rs b/libsql/src/replication/mod.rs
index 4c25a1acb2..32561a3788 100644
--- a/libsql/src/replication/mod.rs
+++ b/libsql/src/replication/mod.rs
@@ -94,6 +94,10 @@ impl Writer {
     pub(crate) fn replicator(&self) -> Option<&EmbeddedReplicator> {
         self.replicator.as_ref()
     }
+
+    pub(crate) fn new_client_id(&mut self) {
+        self.client.new_client_id()
+    }
 }
 
 #[derive(Clone)]


Test patch:
diff --git a/libsql-server/tests/embedded_replica/mod.rs b/libsql-server/tests/embedded_replica/mod.rs
index 6fe50c9ef7..b79ff3e483 100644
--- a/libsql-server/tests/embedded_replica/mod.rs
+++ b/libsql-server/tests/embedded_replica/mod.rs
@@ -1035,3 +1035,92 @@ fn malformed_database() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn txn_bug_issue_1283() {
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(u64::MAX))
+        .build();
+
+    let tmp_embedded = tempdir().unwrap();
+    let tmp_host = tempdir().unwrap();
+    let tmp_embedded_path = tmp_embedded.path().to_owned();
+    let tmp_host_path = tmp_host.path().to_owned();
+
+    make_primary(&mut sim, tmp_host_path.clone());
+
+    sim.client("client", async move {
+        let path = tmp_embedded_path.join("embedded");
+
+        let client = Client::new();
+        client
+            .post("http://primary:9090/v1/namespaces/foo/create", json!({}))
+            .await?;
+
+        let db_url = "http://foo.primary:8080";
+        let replica = libsql::Builder::new_remote_replica(
+            path.to_str().unwrap(),
+            db_url.to_string(),
+            String::new(),
+        )
+        .connector(TurmoilConnector)
+        .build()
+        .await
+        .unwrap();
+
+        let remote = libsql::Builder::new_remote(db_url.to_string(), String::new())
+            .connector(TurmoilConnector)
+            .build()
+            .await
+            .unwrap();
+
+        let replica_conn_1 = replica.connect().unwrap();
+        let replica_conn_2 = replica.connect().unwrap();
+
+        // Not really an embedded replica test but good to check this for remote connections too
+        let remote_conn_1 = remote.connect().unwrap();
+        let remote_conn_2 = remote.connect().unwrap();
+
+        let remote_task_1 = tokio::task::spawn(async move { db_work(remote_conn_1).await });
+        let remote_task_2 = tokio::task::spawn(async move { db_work(remote_conn_2).await });
+
+        let (task_1_res, task_2_res) = tokio::join!(remote_task_1, remote_task_2);
+        let remote_task_1_res = task_1_res.unwrap();
+        let remote_task_2_res = task_2_res.unwrap();
+
+        // Everything works as expected in case of remote connections.
+        assert!(remote_task_1_res.is_ok());
+        assert!(remote_task_2_res.is_ok());
+
+        let replica_task_1 = tokio::task::spawn(async move { db_work(replica_conn_1).await });
+        let replica_task_2 = tokio::task::spawn(async move { db_work(replica_conn_2).await });
+
+        let (task_1_res, task_2_res) = tokio::join!(replica_task_1, replica_task_2);
+        let replica_task_1_res = task_1_res.unwrap();
+        let replica_task_2_res = task_2_res.unwrap();
+
+        if replica_task_1_res.is_err() {
+            panic!("Task 1 failed: {:?}", replica_task_1_res);
+        }
+        if replica_task_2_res.is_err() {
+            panic!("Task 2 failed: {:?}", replica_task_2_res);
+        }
+
+        // One of these concurrent tasks fail currently. Both tasks should succeed.
+        assert!(replica_task_1_res.is_ok());
+        assert!(replica_task_2_res.is_ok());
+
+        Ok(())
+    });
+
+    async fn db_work(conn: libsql::Connection) -> Result<(), anyhow::Error> {
+        let tx = conn.transaction().await?;
+        // Some business logic here...
+        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+        tx.execute("SELECT 1", ()).await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    sim.run().unwrap();
+}