perf: drain SQLite operations through a native FIFO

README.md

@@ -90,6 +90,8 @@ const db = open({ name: 'myDb.sqlite' })
 
 Async operations submitted on the opened `db` connection outside a transaction callback run in call order. Async work waits for an active transaction to finish, while a conflicting sync operation or `close()` throws a busy error.
 
+You can submit several `executeAsync` calls together with `Promise.all`. NitroSQLite sends them to a native FIFO on that connection, so the next query can start without waiting for JavaScript to process the previous result. A single connection still executes one SQL operation at a time. Transactions wait for earlier queries to finish and hold the connection until the callback completes.
+
 `NitroSQLite.native` bypasses this JavaScript queue. Native calls keep each individual SQLite handle safe, but mixing them with a session transaction can still run statements inside that transaction. A build with `SQLITE_THREADSAFE=0` also remains unsafe when different database handles run concurrently unless the caller serializes every SQLite call globally.
 
 ---

example/tests/unit/specs/DatabaseQueue.spec.ts

@@ -87,33 +87,20 @@ export default function registerDatabaseQueueUnitTests() {
       expect(testDbQueue.inProgress).toBe(false)
     })
 
-    it('multiple executeBatchAsync operations are queued', async () => {
+    it('submits multiple executeBatchAsync operations together', async () => {
       const executeBatch1Promise = testDb.executeBatchAsync(TEST_BATCH_COMMANDS)
-
-      expect(testDbQueue.queue.length).toBe(0)
-      expect(testDbQueue.inProgress).toBe(true)
-
       const executeBatch2Promise = testDb.executeBatchAsync(TEST_BATCH_COMMANDS)
-
-      expect(testDbQueue.queue.length).toBe(1)
-      expect(testDbQueue.inProgress).toBe(true)
-
       const executeBatch3Promise = testDb.executeBatchAsync(TEST_BATCH_COMMANDS)
 
-      expect(testDbQueue.queue.length).toBe(2)
-      expect(testDbQueue.inProgress).toBe(true)
-
-      await executeBatch1Promise
-
-      expect(testDbQueue.queue.length).toBe(1)
-      expect(testDbQueue.inProgress).toBe(true)
-
-      await executeBatch2Promise
-
       expect(testDbQueue.queue.length).toBe(0)
       expect(testDbQueue.inProgress).toBe(true)
+      expect(testDbQueue.activeStatements).toBe(3)
 
-      await executeBatch3Promise
+      await Promise.all([
+        executeBatch1Promise,
+        executeBatch2Promise,
+        executeBatch3Promise,
+      ])
 
       expect(testDbQueue.queue.length).toBe(0)
       expect(testDbQueue.inProgress).toBe(false)
@@ -220,9 +207,11 @@ export default function registerDatabaseQueueUnitTests() {
       })
 
       await transactionStarted.promise
-      const externalWrite = testDb.executeAsync(
-        'INSERT INTO User (id, name, age, networth) VALUES (?, ?, ?, ?)',
-        [2, 'external', 2, 2],
+      const externalWrites = Array.from({ length: 24 }, (_, index) =>
+        testDb.executeAsync(
+          'INSERT INTO User (id, name, age, networth) VALUES (?, ?, ?, ?)',
+          [index + 2, `external-${index}`, 2, 2],
+        ),
       )
       finishTransaction.resolve()
 
@@ -231,11 +220,11 @@ export default function registerDatabaseQueueUnitTests() {
       } catch (error) {
         expect((error as Error).message).toContain('rollback transaction')
       }
-      await externalWrite
+      await Promise.all(externalWrites)
 
       expect(
         testDb.execute<{ id: number }>('SELECT id FROM User').results,
-      ).toEqual([{ id: 2 }])
+      ).toEqual(Array.from({ length: 24 }, (_, index) => ({ id: index + 2 })))
     })
 
     it('returns distinct insert IDs from parallel async inserts', async () => {
@@ -257,6 +246,143 @@ export default function registerDatabaseQueueUnitTests() {
       )
     })
 
+    it('starts a transaction after an earlier burst of async writes finishes', async () => {
+      testDb.execute('CREATE TABLE TransactionBarrier (value INTEGER)')
+      const writes = Array.from({ length: 24 }, (_, index) =>
+        testDb.executeAsync(
+          'INSERT INTO TransactionBarrier (value) VALUES (?)',
+          [index],
+        ),
+      )
+      const transaction = testDb.transaction(
+        async (tx) =>
+          tx.execute<{ total: number }>(
+            'SELECT count(*) AS total FROM TransactionBarrier',
+          ).results[0]?.total,
+      )
+
+      await Promise.all(writes)
+      expect(await transaction).toBe(24)
+    })
+
+    it('keeps a batch atomic between async statements', async () => {
+      testDb.execute('CREATE TABLE BatchBarrier (value INTEGER PRIMARY KEY)')
+      const before = testDb.executeAsync(
+        'INSERT INTO BatchBarrier (value) VALUES (1)',
+      )
+      const batch = testDb.executeBatchAsync([
+        { query: 'INSERT INTO BatchBarrier (value) VALUES (2)' },
+        { query: 'INSERT INTO BatchBarrier (value) VALUES (1)' },
+      ])
+      const batchErrorPromise = batch.then(
+        () => undefined,
+        (error: unknown) => error,
+      )
+      const after = testDb.executeAsync(
+        'INSERT INTO BatchBarrier (value) VALUES (3)',
+      )
+
+      await before
+      const batchError = await batchErrorPromise
+      expect(batchError).toBeInstanceOf(NitroSQLiteError)
+      await after
+      expect(
+        testDb.execute<{ value: number }>(
+          'SELECT value FROM BatchBarrier ORDER BY value',
+        ).results,
+      ).toEqual([{ value: 1 }, { value: 3 }])
+    })
+
+    it('rejects synchronous transaction work while an async query is pending', async () => {
+      await testDb.transaction(async (tx) => {
+        const pending = tx.executeAsync('SELECT 1')
+        let syncError: unknown
+        try {
+          tx.execute('SELECT 2')
+        } catch (error) {
+          syncError = error
+        }
+        expect(syncError).toBeInstanceOf(NitroSQLiteError)
+        expect((syncError as Error).message).toContain(
+          'Await all tx.executeAsync',
+        )
+        await pending
+        expect(tx.execute('SELECT 2').results).toEqual([{ '2': 2 }])
+      })
+    })
+
+    it('runs native async statements in submission order', async () => {
+      const dbName = 'native-fifo-order'
+      dropDatabaseIfExists(dbName)
+      NitroSQLite.native.open(dbName)
+
+      try {
+        NitroSQLite.native.execute(
+          dbName,
+          'CREATE TABLE NativeQueueInsert (id INTEGER PRIMARY KEY AUTOINCREMENT, value INTEGER)',
+        )
+        const results = await Promise.all(
+          Array.from({ length: 64 }, (_, index) =>
+            NitroSQLite.native.executeAsync(
+              dbName,
+              'INSERT INTO NativeQueueInsert (value) VALUES (?)',
+              [index],
+            ),
+          ),
+        )
+
+        expect(results.map((result) => result.insertId)).toEqual(
+          Array.from({ length: 64 }, (_, index) => index + 1),
+        )
+
+        const batch = NitroSQLite.native.executeBatchAsync(dbName, [
+          {
+            query: 'INSERT INTO NativeQueueInsert (value) VALUES (?)',
+            params: [[64], [65]],
+          },
+        ])
+        const afterBatch = NitroSQLite.native.executeAsync(
+          dbName,
+          'INSERT INTO NativeQueueInsert (value) VALUES (?)',
+          [66],
+        )
+        await batch
+        expect((await afterBatch).insertId).toBe(67)
+      } finally {
+        NitroSQLite.native.close(dbName)
+        dropDatabaseIfExists(dbName)
+      }
+    })
+
+    it('continues the native FIFO after a query fails', async () => {
+      const dbName = 'native-fifo-recovery'
+      dropDatabaseIfExists(dbName)
+      NitroSQLite.native.open(dbName)
+
+      try {
+        const failed = NitroSQLite.native.executeAsync(
+          dbName,
+          'SELECT * FROM MissingTable',
+        )
+        const next = NitroSQLite.native.executeAsync(
+          dbName,
+          'SELECT 42 AS value',
+        )
+
+        let queryError: unknown
+        try {
+          await failed
+        } catch (error) {
+          queryError = error
+        }
+        expect(queryError).toBeInstanceOf(Error)
+        expect((await next).results).toEqual([{ value: 42 }])
+      } finally {
+        NitroSQLite.native.close(dbName)
+        dropDatabaseIfExists(dbName)
+      }
+    })
+
     it('rejects sync work and close while async work is pending', async () => {
       const dbName = 'busy-close'
       dropDatabaseIfExists(dbName)

packages/react-native-nitro-sqlite/cpp/NitroSQLiteOperations.cpp

@@ -4,8 +4,10 @@
 #include "NitroSQLiteUtils.hpp"
 #include "hybridObjects/HybridNitroSQLiteQueryResult.hpp"
 #include <NitroModules/ArrayBuffer.hpp>
+#include <NitroModules/Promise.hpp>
 #include <cmath>
 #include <ctime>
+#include <exception>
 #include <iostream>
 #include <limits>
 #include <map>
@@ -52,6 +54,38 @@ void SQLiteConnection::close() noexcept {
   database = nullptr;
 }
 
+void SQLiteConnection::enqueueAsync(std::function<void()> operation) {
+  std::lock_guard lock(asyncQueueMutex);
+  if (!asyncWorkerRunning) {
+    // The worker holds this connection alive until it has drained every operation.
+    Promise<void>::async([connection = shared_from_this()] { connection->drainAsync(); });
+    asyncWorkerRunning = true;
+  }
+  asyncQueue.push(std::move(operation));
+}
+
+void SQLiteConnection::drainAsync() {
+  while (true) {
+    std::function<void()> operation;
+    {
+      std::lock_guard lock(asyncQueueMutex);
+      if (asyncQueue.empty()) {
+        asyncWorkerRunning = false;
+        return;
+      }
+      operation = std::move(asyncQueue.front());
+      asyncQueue.pop();
+    }
+    try {
+      operation();
+    } catch (const std::exception& error) {
+      LOGE("Async operation on database %s failed while settling its promise: %s", name.c_str(), error.what());
+    } catch (...) {
+      LOGE("Async operation on database %s failed while settling its promise", name.c_str());
+    }
+  }
+}
+
 void sqliteOpenDb(const std::string& dbName, const std::string& docPath) {
   std::lock_guard lifecycleLock(dbLifecycleMutex);
   {

packages/react-native-nitro-sqlite/cpp/NitroSQLiteOperations.hpp

@@ -2,8 +2,10 @@
 
 #include "NitroSQLiteTypes.hpp"
 #include "hybridObjects/HybridNitroSQLiteQueryResult.hpp"
+#include <functional>
 #include <memory>
 #include <mutex>
+#include <queue>
 #include <sqlite3.h>
 #include <string>
 
@@ -12,18 +14,26 @@ namespace margelo::nitro::rnnitrosqlite {
 // Calls against one connection are serialized by `mutex`. Separate connections
 // intentionally remain independent, so SQLITE_THREADSAFE=0 still requires the
 // caller to serialize SQLite calls globally.
-struct SQLiteConnection final {
+struct SQLiteConnection final : std::enable_shared_from_this<SQLiteConnection> {
   SQLiteConnection(std::string name, sqlite3* database);
   ~SQLiteConnection();
 
   SQLiteConnection(const SQLiteConnection&) = delete;
   SQLiteConnection& operator=(const SQLiteConnection&) = delete;
 
   void close() noexcept;
+  void enqueueAsync(std::function<void()> operation);
 
   const std::string name;
   sqlite3* database;
   std::recursive_mutex mutex;
+
+private:
+  void drainAsync();
+
+  std::mutex asyncQueueMutex;
+  std::queue<std::function<void()>> asyncQueue;
+  bool asyncWorkerRunning = false;
 };
 
 using SQLiteConnectionPtr = std::shared_ptr<SQLiteConnection>;

packages/react-native-nitro-sqlite/cpp/hybridObjects/HybridNitroSQLite.cpp

@@ -13,13 +13,14 @@
 #include <map>
 #include <optional>
 #include <string>
+#include <utility>
 #include <variant>
 #include <vector>
 
 namespace margelo::nitro::rnnitrosqlite {
 
 // Copy any JS-backed ArrayBuffers on the JS thread so they can be safely
-// accessed from the background thread used by Promise::async.
+// accessed from the connection's background worker.
 static std::optional<SQLiteQueryParams> copyArrayBufferParamsForBackground(const std::optional<SQLiteQueryParams>& params) {
   if (!params) {
     return std::nullopt;
@@ -59,6 +60,36 @@ static std::vector<BatchQuery> copyArrayBufferParamsForBackground(const std::vec
   return copiedCommands;
 }
 
+template <typename Result, typename Operation>
+static std::shared_ptr<Promise<Result>> enqueueConnectionOperation(const SQLiteConnectionPtr& connection, Operation&& operation) {
+  auto promise = Promise<Result>::create();
+  try {
+    connection->enqueueAsync([promise, operation = std::forward<Operation>(operation)]() mutable {
+      std::optional<Result> result;
+      try {
+        result.emplace(operation());
+      } catch (...) {
+        promise->reject(std::current_exception());
+        return;
+      }
+      // Resolving may dispatch to JavaScript and throw after the native promise
+      // has settled. Do not try to reject that same promise again.
+      try {
+        promise->resolve(std::move(*result));
+      } catch (...) {
+        if (promise->isPending()) {
+          promise->reject(std::current_exception());
+          return;
+        }
+        throw;
+      }
+    });
+  } catch (...) {
+    promise->reject(std::current_exception());
+  }
+  return promise;
+}
+
 const std::string getDocPath(const std::optional<std::string>& location) {
   std::string tempDocPath = std::string(HybridNitroSQLite::docPath);
   if (location) {
@@ -139,8 +170,8 @@ HybridNitroSQLite::executeAsync(const std::string& dbName, const std::string& qu
     return Promise<std::shared_ptr<HybridNitroSQLiteQueryResultSpec>>::rejected(std::current_exception());
   }
 
-  return Promise<std::shared_ptr<HybridNitroSQLiteQueryResultSpec>>::async(
-      [connection, query, copiedParams]() -> std::shared_ptr<HybridNitroSQLiteQueryResultSpec> {
+  return enqueueConnectionOperation<std::shared_ptr<HybridNitroSQLiteQueryResultSpec>>(
+      connection, [connection, query, copiedParams]() -> std::shared_ptr<HybridNitroSQLiteQueryResultSpec> {
         auto result = sqliteExecute(connection, query, copiedParams);
         return result;
       });
@@ -166,7 +197,7 @@ std::shared_ptr<Promise<BatchQueryResult>> HybridNitroSQLite::executeBatchAsync(
     return Promise<BatchQueryResult>::rejected(std::current_exception());
   }
 
-  return Promise<BatchQueryResult>::async([connection, copiedCommands]() -> BatchQueryResult {
+  return enqueueConnectionOperation<BatchQueryResult>(connection, [connection, copiedCommands]() -> BatchQueryResult {
     auto result = sqliteExecuteBatch(connection, copiedCommands);
     return BatchQueryResult(result.rowsAffected);
   });
@@ -184,7 +215,7 @@ std::shared_ptr<Promise<FileLoadResult>> HybridNitroSQLite::loadFileAsync(const
   } catch (...) {
     return Promise<FileLoadResult>::rejected(std::current_exception());
   }
-  return Promise<FileLoadResult>::async([connection, location]() -> FileLoadResult {
+  return enqueueConnectionOperation<FileLoadResult>(connection, [connection, location]() -> FileLoadResult {
     const auto result = importSqlFile(connection, location);
     return FileLoadResult(result.commands, result.rowsAffected);
   });