bug: fix close & transaction handling

Author: codekeyz

src/lib.rs

@@ -67,6 +67,7 @@ pub unsafe extern "C" fn sqlite3_open_v2(
 
     let reqwest_client = reqwest::Client::builder()
         .user_agent("libsqlite3_turso/1.0.0")
+        .timeout(std::time::Duration::from_secs(30))
         .build()
         .unwrap();
 
@@ -81,6 +82,7 @@ pub unsafe extern "C" fn sqlite3_open_v2(
         error_stack: Mutex::new(vec![]),
         transaction_baton: Mutex::new(None),
         last_insert_rowid: Mutex::new(None),
+        transaction_has_began: Mutex::new(false),
         delete_hook: Mutex::new(None),
         insert_hook: Mutex::new(None),
         update_hook: Mutex::new(None),
@@ -383,7 +385,7 @@ pub extern "C" fn sqlite3_errmsg(db: *mut SQLite3) -> *const c_char {
 
     if let Some(error_entry) = sqlite::get_latest_error(db) {
         match CString::new(error_entry.0) {
-            Ok(c_string) => c_string.as_ptr(),
+            Ok(c_string) => c_string.into_raw(),
             Err(_) => std::ptr::null(),
         }
     } else {
@@ -669,7 +671,7 @@ pub extern "C" fn sqlite3_get_autocommit(db: *mut SQLite3) -> c_int {
 
     let db = unsafe { &*db };
 
-    if db.transaction_active() {
+    if db.has_began_transaction() {
         0 // Transaction is active
     } else {
         1 // Autocommit mode

src/proxy.rs

@@ -1,6 +1,6 @@
 use reqwest::Client;
 use serde::Deserialize;
-use std::{collections::HashMap, error::Error};
+use std::{collections::HashMap, error::Error, time::Duration};
 
 use crate::{
     sqlite::{SQLite3, Value},
@@ -51,12 +51,6 @@ pub async fn execute_sql_and_params(
 ) -> Result<RemoteSqliteResponse, Box<dyn Error>> {
     let mut query_request = serde_json::Map::new();
 
-    if let Some(b) = baton {
-        query_request.insert("baton".to_string(), serde_json::json!(b));
-    }
-
-    let can_keep_open = !(baton.is_some() && sql.contains("COMMIT"));
-
     let mut json_array: Vec<serde_json::Value> = Vec::new();
 
     json_array.push(serde_json::json!({
@@ -67,7 +61,9 @@ pub async fn execute_sql_and_params(
             }
         }));
 
-    if !can_keep_open {
+    if db.has_began_transaction() {
+        query_request.insert("baton".to_string(), serde_json::json!(baton));
+    } else {
         json_array.push(serde_json::json!({
             "type": "close"
         }));
@@ -122,55 +118,107 @@ async fn send_sql_request(
     Ok(parsed)
 }
 
-async fn send_remote_request(
+pub async fn send_remote_request(
     client: &Client,
     turso_config: &TursoConfig,
     path: &str,
     request: serde_json::Value,
 ) -> Result<serde_json::Value, Box<dyn Error>> {
-    let response = client
-        .post(format!("https://{}/{}", turso_config.db_url, path))
-        .header("Content-Type", "application/json")
-        .header("Authorization", format!("Bearer {}", turso_config.db_token))
-        .json(&request)
-        .send()
-        .await?;
+    const MAX_ATTEMPTS: usize = 5;
+    let mut last_error = String::new();
+
+    for attempt in 1..=MAX_ATTEMPTS {
+        if cfg!(debug_assertions) {
+            println!(
+                "Attempt {}: Sending request to {}",
+                attempt, turso_config.db_url
+            );
+        }
 
-    let status = response.status();
-    let response_text = response.text().await?;
+        let resp = client
+            .post(format!("https://{}/{}", turso_config.db_url, path))
+            .header("Content-Type", "application/json")
+            .header("Authorization", format!("Bearer {}", turso_config.db_token))
+            .json(&request)
+            .send()
+            .await;
+
+        let resp = match resp {
+            Ok(r) => r,
+            Err(e) => {
+                last_error = format!("Request failed: {}", e);
+                if attempt < MAX_ATTEMPTS {
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                    continue;
+                } else {
+                    return Err(last_error.into());
+                }
+            }
+        };
+
+        let status = resp.status();
+        let text = match resp.text().await {
+            Ok(t) => t,
+            Err(e) => {
+                last_error = format!("Failed to read response body: {}", e);
+                if attempt < MAX_ATTEMPTS {
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                    continue;
+                } else {
+                    return Err(last_error.into());
+                }
+            }
+        };
 
-    if cfg!(debug_assertions) {
-        println!("Received Response: {}\n", &response_text);
-    }
+        if cfg!(debug_assertions) {
+            println!("Response received: {}", text);
+        }
 
-    if !status.is_success() {
-        if let Ok(error_body) = serde_json::from_str::<serde_json::Value>(&response_text) {
-            if let Some(error_message) = error_body.get("error").and_then(|e| e.as_str()) {
-                return Err(error_message.into());
+        if !status.is_success() {
+            if let Ok(err_json) = serde_json::from_str::<serde_json::Value>(&text) {
+                if let Some(msg) = err_json.get("error").and_then(|v| v.as_str()) {
+                    last_error = format!("API error: {}", msg);
+                } else {
+                    last_error = format!("HTTP error {}: {}", status, text);
+                }
+            } else {
+                last_error = format!("HTTP error {} with invalid JSON: {}", status, text);
             }
-        }
-        return Err(format!("LibSqlite3_Turso Error: {}", response_text).into());
-    }
 
-    let parsed_response = serde_json::from_str(&response_text);
-    if parsed_response.is_err() {
-        return Err(format!("Failed to parse response: {}", parsed_response.unwrap_err()).into());
-    }
+            if attempt < MAX_ATTEMPTS {
+                tokio::time::sleep(Duration::from_millis(100)).await;
+                continue;
+            } else {
+                return Err(last_error.into());
+            }
+        }
 
-    let parsed_response: serde_json::Value = parsed_response.unwrap();
-    if let Some(results) = parsed_response.get("results").and_then(|r| r.as_array()) {
-        for result in results {
-            if let Some(error) = result
-                .get("error")
-                .and_then(|e| e.get("message"))
-                .and_then(|m| m.as_str())
-            {
-                return Err(error.into());
+        let parsed: serde_json::Value = match serde_json::from_str(&text) {
+            Ok(v) => v,
+            Err(e) => return Err(format!("Failed to parse JSON: {}", e).into()),
+        };
+
+        // Check for embedded DB errors
+        if let Some(results) = parsed.get("results").and_then(|r| r.as_array()) {
+            for result in results {
+                if let Some(msg) = result
+                    .get("error")
+                    .and_then(|e| e.get("message"))
+                    .and_then(|m| m.as_str())
+                {
+                    return Err(msg.to_string().into());
+                }
             }
         }
+
+        return Ok(parsed);
     }
 
-    Ok(parsed_response)
+    Err(format!(
+        "Failed to get successful response after {} attempts: {}",
+        MAX_ATTEMPTS, last_error
+    )
+    .into())
 }
 
 pub fn convert_params_to_json(params: &HashMap<i32, Value>) -> Vec<serde_json::Value> {

src/sqlite.rs

@@ -64,6 +64,7 @@ pub struct SQLite3 {
     pub last_insert_rowid: Mutex<Option<i64>>, // Last inserted row ID
     pub error_stack: Mutex<Vec<(String, c_int)>>, // Stack to store error messages
     pub transaction_baton: Mutex<Option<String>>, // Baton for transaction management
+    pub transaction_has_began: Mutex<bool>, // Flag to check if a transaction has started
     pub update_hook: Mutex<Option<(SqliteHook, *mut c_void)>>, // Update hook callback
     pub insert_hook: Mutex<Option<(SqliteHook, *mut c_void)>>, // Insert hook callback
     pub delete_hook: Mutex<Option<(SqliteHook, *mut c_void)>>, // Delete hook callback
@@ -114,8 +115,8 @@ impl SQLite3 {
         SQLITE_OK
     }
 
-    pub fn transaction_active(&self) -> bool {
-        self.transaction_baton.lock().unwrap().is_some()
+    pub fn has_began_transaction(&self) -> bool {
+        *self.transaction_has_began.lock().unwrap()
     }
 }
 
@@ -172,10 +173,11 @@ pub fn reset_txn_on_db(db: *mut SQLite3) -> c_int {
 
     let db = unsafe { &mut *db };
 
-    if !db.transaction_active() {
+    if !db.has_began_transaction() {
         return SQLITE_OK;
     }
 
+    *db.transaction_has_began.lock().unwrap() = false;
     db.transaction_baton.lock().unwrap().take();
 
     SQLITE_OK
@@ -284,7 +286,7 @@ pub async fn begin_tnx_on_db(db: *mut SQLite3) -> Result<c_int, Box<dyn Error>>
 
     let db = unsafe { &mut *db };
 
-    if db.transaction_active() {
+    if db.has_began_transaction() {
         push_error(
             db,
             (
@@ -297,6 +299,7 @@ pub async fn begin_tnx_on_db(db: *mut SQLite3) -> Result<c_int, Box<dyn Error>>
 
     let baton_value = get_transaction_baton(&db.client, &db.turso_config).await?;
     db.transaction_baton.lock().unwrap().replace(baton_value);
+    *db.transaction_has_began.lock().unwrap() = true;
 
     Ok(SQLITE_OK)
 }
@@ -308,7 +311,7 @@ pub async fn commit_tnx_on_db(db: *mut SQLite3) -> Result<c_int, Box<dyn Error>>
 
     let db = unsafe { &mut *db };
 
-    if !db.transaction_active() {
+    if !db.has_began_transaction() {
         push_error(
             db,
             (