Move transaction lock to native

.oxfmtrc.json

@@ -0,0 +1,4 @@
+{
+  "$schema": "./node_modules/oxfmt/configuration_schema.json",
+  "ignorePatterns": []
+}

.oxlintrc.json

@@ -0,0 +1,15 @@
+{
+  "$schema": "./node_modules/oxlint/configuration_schema.json",
+  "plugins": [
+    "typescript",
+    "unicorn",
+    "oxc"
+  ],
+  "categories": {
+    "correctness": "warn"
+  },
+  "rules": {},
+  "env": {
+    "builtin": true
+  }
+}

cpp/DBHostObject.cpp

@@ -298,6 +298,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
   });
 
   function_map["close"] = HFN(this) {
+    reject_all_transaction_lock_waiters(rt,
+                                        "[op-sqlite] database is closed");
     invalidated = true;
     // Abort pending native SQLite work before waiting on the thread pool.
 #if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO)
@@ -346,6 +348,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
       throw std::runtime_error("[op-sqlite] Delete no longer takes arguments");
     }
 
+    reject_all_transaction_lock_waiters(rt,
+                      "[op-sqlite] database was deleted");
     invalidated = true;
     // Abort pending native SQLite work before waiting on the thread pool.
 #if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO)
@@ -414,6 +418,33 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     return create_js_rows(rt, status);
   });
 
+  function_map["beginTransaction"] = HFN(this) {
+#ifdef OP_SQLITE_USE_LIBSQL
+    auto status = opsqlite_libsql_execute(db, "BEGIN TRANSACTION;", nullptr);
+#else
+    auto status = opsqlite_execute(db, "BEGIN TRANSACTION;", nullptr);
+#endif
+    return create_js_rows(rt, status);
+  });
+
+  function_map["commitTransaction"] = HFN(this) {
+#ifdef OP_SQLITE_USE_LIBSQL
+    auto status = opsqlite_libsql_execute(db, "COMMIT;", nullptr);
+#else
+    auto status = opsqlite_execute(db, "COMMIT;", nullptr);
+#endif
+    return create_js_rows(rt, status);
+  });
+
+  function_map["rollbackTransaction"] = HFN(this) {
+#ifdef OP_SQLITE_USE_LIBSQL
+    auto status = opsqlite_libsql_execute(db, "ROLLBACK;", nullptr);
+#else
+    auto status = opsqlite_execute(db, "ROLLBACK;", nullptr);
+#endif
+    return create_js_rows(rt, status);
+  });
+
   function_map["executeRawSync"] = HFN(this) {
     const std::string query = jsi_string_to_utf8(rt, args[0].asString(rt));
     std::vector<JSVariant> params = count == 2 && args[1].isObject()
@@ -523,6 +554,39 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
         });
   });
 
+  function_map["acquireTransactionLock"] = HFN(this) {
+    if (invalidated) {
+      throw std::runtime_error(
+          "[op-sqlite][acquireTransactionLock] database is closed");
+    }
+
+    auto promiseCtr = rt.global().getPropertyAsFunction(rt, "Promise");
+    auto promise = promiseCtr.callAsConstructor(rt, HFN(this) {
+      auto resolve = std::make_shared<jsi::Value>(rt, args[0]);
+      auto reject = std::make_shared<jsi::Value>(rt, args[1]);
+
+      if (!transaction_lock_in_progress) {
+        transaction_lock_in_progress = true;
+        resolve->asObject(rt).asFunction(rt).call(rt, {});
+      } else {
+        transaction_lock_waiters.push_back({resolve, reject});
+      }
+
+      return {};
+    }));
+
+    return promise;
+  });
+
+  function_map["releaseTransactionLock"] = HFN(this) {
+    if (!transaction_lock_in_progress) {
+      return {};
+    }
+
+    resolve_next_transaction_lock_waiter(rt);
+    return {};
+  });
+
 #if defined(OP_SQLITE_USE_LIBSQL) || defined(OP_SQLITE_USE_TURSO)
   function_map["sync"] = HFN(this) {
 #ifdef OP_SQLITE_USE_LIBSQL
@@ -742,6 +806,29 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
   });
 }
 
+void DBHostObject::resolve_next_transaction_lock_waiter(jsi::Runtime &rt) {
+  if (!transaction_lock_waiters.empty()) {
+    auto waiter = transaction_lock_waiters.front();
+    transaction_lock_waiters.pop_front();
+    waiter.resolve->asObject(rt).asFunction(rt).call(rt, {});
+    return;
+  }
+
+  transaction_lock_in_progress = false;
+}
+
+void DBHostObject::reject_all_transaction_lock_waiters(
+    jsi::Runtime &rt, const std::string &message) {
+  transaction_lock_in_progress = false;
+  while (!transaction_lock_waiters.empty()) {
+    auto waiter = transaction_lock_waiters.front();
+    transaction_lock_waiters.pop_front();
+    jsi::JSError js_error(rt, message);
+    const auto &error_value = js_error.value();
+    waiter.reject->asObject(rt).asFunction(rt).call(rt, error_value);
+  }
+}
+
 std::vector<jsi::PropNameID> DBHostObject::getPropertyNames(jsi::Runtime &_rt) {
   std::vector<jsi::PropNameID> keys;
   keys.reserve(function_map.size());

cpp/DBHostObject.h

@@ -3,6 +3,7 @@
 #include "OPThreadPool.h"
 #include "types.hpp"
 #include <ReactCommon/CallInvoker.h>
+#include <deque>
 #include <jsi/jsi.h>
 #include <set>
 #ifdef OP_SQLITE_USE_LIBSQL
@@ -41,6 +42,11 @@ struct ReactiveQuery {
   std::shared_ptr<jsi::Value> callback;
 };
 
+struct PendingTransactionLockWaiter {
+  std::shared_ptr<jsi::Value> resolve;
+  std::shared_ptr<jsi::Value> reject;
+};
+
 class JSI_EXPORT DBHostObject : public jsi::HostObject {
 public:
   // Normal constructor shared between all backends
@@ -84,6 +90,9 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject {
   void auto_register_update_hook();
   void create_jsi_functions(jsi::Runtime &rt);
   void flush_pending_reactive_queries(const std::shared_ptr<jsi::Value> &resolve);
+  void resolve_next_transaction_lock_waiter(jsi::Runtime &rt);
+  void reject_all_transaction_lock_waiters(jsi::Runtime &rt,
+                                           const std::string &message);
 
   std::unordered_map<std::string, jsi::Value> function_map;
   std::string base_path;
@@ -95,6 +104,8 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject {
   std::shared_ptr<jsi::Value> rollback_hook_callback;
   std::vector<std::shared_ptr<ReactiveQuery>> reactive_queries;
   std::vector<PendingReactiveInvocation> pending_reactive_invocations;
+  std::deque<PendingTransactionLockWaiter> transaction_lock_waiters;
+  bool transaction_lock_in_progress = false;
   bool is_update_hook_registered = false;
   bool invalidated = false;
 #ifdef OP_SQLITE_USE_LIBSQL

cpp/bridge.cpp

@@ -328,7 +328,7 @@
   if (isFailed) {
     throw std::runtime_error(
         "[op-sqlite] SQLite code: " + std::to_string(result) +
         " execution error: " + std::string(errorMessage));
   }
 
   int changedRowCount = sqlite3_changes(db);
@@ -642,7 +642,7 @@
   if (isFailed) {
     throw std::runtime_error(
         "[op-sqlite] SQLite error code: " + std::to_string(result) +
         ", description: " + std::string(errorMessage));
   }
 
   int changedRowCount = sqlite3_changes(db);
@@ -774,7 +774,7 @@
   if (isFailed) {
     throw std::runtime_error(
         "[op-sqlite] SQLite error code: " + std::to_string(step) +
         ", description: " + std::string(errorMessage));
   }
 
   int changedRowCount = sqlite3_changes(db);
@@ -880,15 +880,31 @@
   }
 
   int affectedRows = 0;
-  // opsqlite_execute(db, "BEGIN EXCLUSIVE TRANSACTION", nullptr);
-  for (int i = 0; i < commandCount; i++) {
-    const auto &command = commands->at(i);
-    // We do not provide a datastructure to receive query data because we
-    // don't need/want to handle this results in a batch execution
-    // There is also no need to commit/catch this transaction, this is done
-    // in the JS code
-    auto result = opsqlite_execute(db, command.sql, &command.params);
-    affectedRows += result.affectedRows;
+  const bool should_manage_transaction = sqlite3_get_autocommit(db) != 0;
+  if (should_manage_transaction) {
+    opsqlite_execute(db, "BEGIN TRANSACTION;", nullptr);
+  }
+
+  try {
+    for (int i = 0; i < commandCount; i++) {
+      const auto &command = commands->at(i);
+      // We do not provide a datastructure to receive query data because we
+      // don't need/want to handle this results in a batch execution
+      auto result = opsqlite_execute(db, command.sql, &command.params);
+      affectedRows += result.affectedRows;
+    }
+
+    if (should_manage_transaction) {
+      opsqlite_execute(db, "COMMIT;", nullptr);
+    }
+  } catch (...) {
+    if (should_manage_transaction) {
+      try {
+        opsqlite_execute(db, "ROLLBACK;", nullptr);
+      } catch (...) {
+      }
+    }
+    throw;
   }
 
   return BatchResult{

cpp/libsql/bridge.cpp

@@ -744,9 +744,10 @@ opsqlite_libsql_execute_batch(DB const &db,
         throw std::runtime_error("No SQL commands provided");
     }
 
+    int affectedRows = 0;
+    opsqlite_libsql_execute(db, "BEGIN TRANSACTION;", nullptr);
+
     try {
-        int affectedRows = 0;
-        // opsqlite_libsql_execute(db, "BEGIN EXCLUSIVE TRANSACTION", nullptr);
         for (int i = 0; i < commandCount; i++) {
             auto command = commands->at(i);
             // We do not provide a datastructure to receive query data because
@@ -755,16 +756,19 @@ opsqlite_libsql_execute_batch(DB const &db,
                 opsqlite_libsql_execute(db, command.sql, &command.params);
             affectedRows += result.affectedRows;
         }
-        // opsqlite_libsql_execute(db, "COMMIT", nullptr);
+
+        opsqlite_libsql_execute(db, "COMMIT;", nullptr);
+
         return BatchResult{
             .affectedRows = affectedRows,
             .commands = static_cast<int>(commandCount),
         };
-    } catch (std::exception &exc) {
-        // opsqlite_libsql_execute(db, "ROLLBACK", nullptr);
-        return BatchResult{
-            .message = exc.what(),
-        };
+    } catch (...) {
+        try {
+            opsqlite_libsql_execute(db, "ROLLBACK;", nullptr);
+        } catch (...) {
+        }
+        throw;
     }
 }
 

example/src/App.tsx

@@ -36,10 +36,10 @@ export default function App() {
 				console.log("OPSQLITE_TEST_RESULT:FAIL");
 			}
 
-			setTimeout(() => {
+			setTimeout(async () => {
 			  try {
 			    global?.gc?.();
-			    let perfRes = performanceTest();
+			    let perfRes = await performanceTest();
 			    setPerfResult(perfRes);
 			  } catch (e) {
 			    // intentionally left blank

example/src/performance_test.ts

@@ -1,37 +1,83 @@
-import {open} from '@op-engineering/op-sqlite';
+import { open } from "@op-engineering/op-sqlite";
 
-export function performanceTest() {
+export async function performanceTest() {
   const db = open({
-    name: 'perfTest.sqlite',
+    name: "perfTest.sqlite",
   });
 
-  // Create table with 14 columns
   db.executeSync(
     `CREATE TABLE IF NOT EXISTS perf_table (
       id INTEGER PRIMARY KEY,
       col1 TEXT, col2 TEXT, col3 TEXT, col4 TEXT, col5 TEXT, col6 TEXT, col7 TEXT,
       col8 TEXT, col9 TEXT, col10 TEXT, col11 TEXT, col12 TEXT, col13 TEXT, col14 TEXT
     )`,
   );
-  // Clear table
-  db.executeSync('DELETE FROM perf_table');
-  const testRow =Array(14).fill('test') ;
+
+  db.executeSync("DELETE FROM perf_table");
+  const testRow = Array(14).fill("test");
 
   let start = performance.now();
 
-  for (let i = 0; i < 1_000; i++) {
-    // Insert a single row for querying
-    db.executeSync(
-      `INSERT INTO perf_table (
+  for (let i = 0; i < 200; i++) {
+    let firstPromise = db.transaction(async (tx) => {
+      await tx.execute(
+        `INSERT INTO perf_table (
+        col1, col2, col3, col4, col5, col6, col7,
+        col8, col9, col10, col11, col12, col13, col14
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        testRow,
+      );
+
+      await tx.execute(
+        `INSERT INTO perf_table (
+        col1, col2, col3, col4, col5, col6, col7,
+        col8, col9, col10, col11, col12, col13, col14
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        testRow,
+      );
+    });
+
+    let secondPromise = db.transaction(async (tx) => {
+      await tx.execute(
+        `INSERT INTO perf_table (
+        col1, col2, col3, col4, col5, col6, col7,
+        col8, col9, col10, col11, col12, col13, col14
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        testRow,
+      );
+
+      await tx.execute(
+        `INSERT INTO perf_table (
+        col1, col2, col3, col4, col5, col6, col7,
+        col8, col9, col10, col11, col12, col13, col14
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        testRow,
+      );
+    });
+
+    let thirdPromise = db.transaction(async (tx) => {
+      await tx.execute(
+        `INSERT INTO perf_table (
         col1, col2, col3, col4, col5, col6, col7,
         col8, col9, col10, col11, col12, col13, col14
       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
-      testRow,
-    );
+        testRow,
+      );
+
+      await tx.execute(
+        `INSERT INTO perf_table (
+        col1, col2, col3, col4, col5, col6, col7,
+        col8, col9, col10, col11, col12, col13, col14
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        testRow,
+      );
+    });
+
+    await Promise.all([firstPromise, secondPromise, thirdPromise]);
   }
 
   for (let i = 0; i < 100000; i++) {
-    db.executeSync('SELECT * FROM perf_table WHERE id = 1');
+    await db.execute("SELECT * FROM perf_table WHERE id = 1");
   }
   const end = performance.now();
   // console.log(`Queried 100000 times in ${end - start} ms`);