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,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     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 +528,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_waiters.push_back({resolve, reject});
+      } else {
+        transaction_lock_in_progress = true;
+        resolve->asObject(rt).asFunction(rt).call(rt, {});
+      }
+
+      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 +780,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

@@ -8,21 +8,14 @@ import "./tests"; // import all tests to register them
 import {performanceTest} from './performance_test';
 import { StyleSheet, Text, View } from "react-native";
 import { SafeAreaProvider, SafeAreaView } from "react-native-safe-area-context";
-// import {open} from '@op-engineering/op-sqlite';
 
 export default function App() {
 	const [results, setResults] = useState<any>(null);
 	const [perfResult, setPerfResult] = useState<number>(0);
-	const [openTime, setOpenTime] = useState(0);
 
 	useEffect(() => {
 		console.log("App has started 🟢");
 		const work = async () => {
-			// let start = performance.now();
-			// open({
-			//   name: 'dummyDb.sqlite',
-			// });
-			// setOpenTime(performance.now() - start);
 
 			try {
 				console.log("TESTS STARTED 🟠");
@@ -36,15 +29,15 @@ 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
 			  }
-			}, 4000);
+			}, 1000);
 		};
 
 		work();
@@ -79,10 +72,7 @@ export default function App() {
 			<SafeAreaView style={styles.container}>
 				<View>
 					<Text style={styles.performanceText}>
-						Open DB time: {openTime.toFixed(0)} ms
-					</Text>
-					<Text style={styles.performanceText}>
-						100_000 query time: {perfResult.toFixed(0)} ms
+						Performance Test: {perfResult.toFixed(0)} ms
 					</Text>
 				</View>
 				<View style={styles.results}>{displayResults(results)}</View>

example/src/performance_test.ts

@@ -1,41 +1,53 @@
-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') ;
 
-  let start = performance.now();
+  const testRow = Array(14).fill("test");
+  const runDurations: number[] = [];
 
-  for (let i = 0; i < 1_000; i++) {
-    // Insert a single row for querying
-    db.executeSync(
-      `INSERT INTO perf_table (
+  for (let run = 0; run < 10; run++) {
+    db.executeSync("DELETE FROM perf_table");
+
+    const start = performance.now();
+
+    for (let i = 0; i < 200; i++) {
+      const txPromises = Array.from({ length: 10 }, () =>
+        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,
+          );
+        }),
+      );
 
-  for (let i = 0; i < 100000; i++) {
-    db.executeSync('SELECT * FROM perf_table WHERE id = 1');
+      await Promise.all(txPromises);
+    }
+
+    for (let i = 0; i < 100000; i++) {
+      db.executeSync("SELECT * FROM perf_table WHERE id = 1");
+    }
+
+    const end = performance.now();
+    runDurations.push(end - start);
   }
-  const end = performance.now();
-  // console.log(`Queried 100000 times in ${end - start} ms`);
+
+  const total = runDurations.reduce((sum, duration) => sum + duration, 0);
+  const average = total / runDurations.length;
 
   // await db.close();
-  return end - start;
+  return average;
 }