fix bad autocommit (#1542)

* add test

* fix bug when updating autocommit status in write proxy

* fmt

Author: MarinPostma

libsql-server/src/connection/write_proxy.rs

@@ -370,27 +370,30 @@ where
         program: Program,
         builder: B,
     ) -> crate::Result<(B, TxnStatus, Option<FrameNo>)> {
-        let mut txn_status = TxnStatus::Invalid;
+        let txn_status = Arc::new(parking_lot::Mutex::new(TxnStatus::Invalid));
         let mut new_frame_no = None;
         let builder_config = self.builder_config.clone();
-        let cb = move |response: exec_resp::Response, builder: &mut B| match response {
-            exec_resp::Response::ProgramResp(resp) => {
-                crate::rpc::streaming_exec::apply_program_resp_to_builder(
-                    &builder_config,
-                    builder,
-                    resp,
-                    |last_frame_no, is_autocommit| {
-                        txn_status = if is_autocommit {
-                            TxnStatus::Init
-                        } else {
-                            TxnStatus::Txn
-                        };
-                        new_frame_no = last_frame_no;
-                    },
-                )
+        let cb = {
+            let txn_status = txn_status.clone();
+            move |response: exec_resp::Response, builder: &mut B| match response {
+                exec_resp::Response::ProgramResp(resp) => {
+                    crate::rpc::streaming_exec::apply_program_resp_to_builder(
+                        &builder_config,
+                        builder,
+                        resp,
+                        |last_frame_no, is_autocommit| {
+                            *txn_status.lock() = if is_autocommit {
+                                TxnStatus::Init
+                            } else {
+                                TxnStatus::Txn
+                            };
+                            new_frame_no = last_frame_no;
+                        },
+                    )
+                }
+                exec_resp::Response::DescribeResp(_) => Err(Error::PrimaryStreamMisuse),
+                exec_resp::Response::Error(e) => Err(Error::RpcQueryError(e)),
             }
-            exec_resp::Response::DescribeResp(_) => Err(Error::PrimaryStreamMisuse),
-            exec_resp::Response::Error(e) => Err(Error::RpcQueryError(e)),
         };
 
         let builder = self
@@ -403,6 +406,7 @@ where
             )
             .await?;
 
+        let txn_status = *txn_status.lock();
         Ok((builder, txn_status, new_frame_no))
     }
 

libsql-server/tests/cluster/replication.rs

@@ -237,3 +237,97 @@ fn replica_lazy_creation() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn replica_interactive_transaction() {
+    let mut sim = turmoil::Builder::new()
+        .simulation_duration(Duration::from_secs(1000))
+        .build();
+
+    let prim_tmp = tempfile::tempdir().unwrap();
+
+    sim.host("primary", {
+        let prim_path = prim_tmp.path().to_path_buf();
+        move || {
+            let prim_path = prim_path.clone();
+            async move {
+                let primary = TestServer {
+                    path: prim_path.into(),
+                    db_config: DbConfig {
+                        max_log_size: 1,
+                        ..Default::default()
+                    },
+                    rpc_server_config: Some(RpcServerConfig {
+                        acceptor: TurmoilAcceptor::bind(([0, 0, 0, 0], 5050)).await.unwrap(),
+                        tls_config: None,
+                    }),
+                    ..Default::default()
+                };
+
+                primary.start_sim(8080).await.unwrap();
+
+                Ok(())
+            }
+        }
+    });
+
+    sim.host("replica", {
+        move || async move {
+            let tmp = tempfile::tempdir().unwrap();
+            let replica = TestServer {
+                path: tmp.path().to_path_buf().into(),
+                db_config: DbConfig {
+                    max_log_size: 1,
+                    ..Default::default()
+                },
+                rpc_client_config: Some(RpcClientConfig {
+                    remote_url: "http://primary:5050".into(),
+                    tls_config: None,
+                    connector: TurmoilConnector,
+                }),
+                ..Default::default()
+            };
+
+            replica.start_sim(8080).await.unwrap();
+
+            Ok(())
+        }
+    });
+
+    sim.client("client", async move {
+        let db = Database::open_remote_with_connector("http://replica:8080", "", TurmoilConnector)
+            .unwrap();
+        let conn = db.connect().unwrap();
+
+        let tx = conn
+            .transaction_with_behavior(libsql::TransactionBehavior::Immediate)
+            .await
+            .unwrap();
+
+        tx.execute("create table test (x)", ()).await.unwrap();
+        tx.execute("insert into test values (12)", ())
+            .await
+            .unwrap();
+        tx.execute("insert into test values (12)", ())
+            .await
+            .unwrap();
+        tx.commit().await.unwrap();
+
+        let count = conn
+            .query("select count(0) from test", ())
+            .await
+            .unwrap()
+            .next()
+            .await
+            .unwrap()
+            .unwrap()
+            .get::<u32>(0)
+            .unwrap();
+
+        assert_eq!(count, 2);
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}