perf: drain SQLite operations through a native FIFO

.github/workflows/build-ios.yml

@@ -112,7 +112,7 @@ jobs:
             -scheme NitroSQLiteExample \
             -sdk iphonesimulator \
             -configuration Debug \
-            -destination 'platform=iOS Simulator,name=iPhone 17 Pro' \
+            -destination 'generic/platform=iOS Simulator' \
             -showBuildTimingSummary \
             ONLY_ACTIVE_ARCH=YES \
             build \

README.md

@@ -76,6 +76,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

@@ -220,9 +220,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 +233,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 +259,68 @@ 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('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('rejects sync work and close while async work is pending', async () => {
       const dbName = 'busy-close'
       dropDatabaseIfExists(dbName)

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,23 @@ 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 {
+      try {
+        promise->resolve(operation());
+      } catch (...) {
+        promise->reject(std::current_exception());
+      }
+    });
+  } 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 +157,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 +184,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 +202,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);
   });

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

@@ -4,6 +4,7 @@
 #include "logs.hpp"
 #include "utils.hpp"
 #include <NitroModules/ArrayBuffer.hpp>
+#include <NitroModules/Promise.hpp>
 #include <cmath>
 #include <ctime>
 #include <iostream>
@@ -54,6 +55,32 @@ 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();
+    }
+    operation();
+  }
+}
+
 void sqliteOpenDb(const std::string& dbName, const std::string& docPath) {
   std::lock_guard lifecycleLock(dbLifecycleMutex);
   {

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

@@ -2,8 +2,10 @@
 
 #include "hybridObjects/HybridNitroSQLiteQueryResult.hpp"
 #include "types.hpp"
+#include <functional>
 #include <memory>
 #include <mutex>
+#include <queue>
 #include <sqlite3.h>
 #include <string>
 
@@ -12,18 +14,26 @@ namespace margelo::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/src/DatabaseQueue.ts

@@ -1,6 +1,7 @@
 import NitroSQLiteError from './NitroSQLiteError'
 
 export interface QueuedOperation {
+  kind: 'statement' | 'exclusive'
   /**
    * Starts the operation
    */
@@ -10,6 +11,8 @@ export interface QueuedOperation {
 export type DatabaseQueue = {
   queue: QueuedOperation[]
   inProgress: boolean
+  activeStatements: number
+  draining: boolean
 }
 
 const databaseQueues = new Map<string, DatabaseQueue>()
@@ -21,7 +24,12 @@ export function openDatabaseQueue(dbName: string) {
     )
   }
 
-  databaseQueues.set(dbName, { queue: [], inProgress: false })
+  databaseQueues.set(dbName, {
+    queue: [],
+    inProgress: false,
+    activeStatements: 0,
+    draining: false,
+  })
 }
 
 export function closeDatabaseQueue(dbName: string) {
@@ -57,7 +65,22 @@ export function getDatabaseQueue(dbName: string) {
 export function queueOperationAsync<Result>(
   dbName: string,
   callback: () => Promise<Result>,
-) {
+): Promise<Result> {
+  return enqueueOperation(dbName, 'exclusive', callback)
+}
+
+export function queueStatementAsync<Result>(
+  dbName: string,
+  callback: () => Promise<Result>,
+): Promise<Result> {
+  return enqueueOperation(dbName, 'statement', callback)
+}
+
+function enqueueOperation<Result>(
+  dbName: string,
+  kind: QueuedOperation['kind'],
+  callback: () => Promise<Result>,
+): Promise<Result> {
   const databaseQueue = getDatabaseQueue(dbName)
 
   return new Promise<Result>((resolve, reject) => {
@@ -68,32 +91,59 @@ export function queueOperationAsync<Result>(
       } catch (error) {
         reject(error)
       } finally {
-        databaseQueue.inProgress = false
-        startOperationAsync(databaseQueue)
+        if (kind === 'statement') {
+          databaseQueue.activeStatements--
+          if (databaseQueue.activeStatements === 0) {
+            databaseQueue.inProgress = false
+          }
+        } else {
+          databaseQueue.inProgress = false
+        }
+        startNextOperations(databaseQueue)
       }
     }
 
     const operation: QueuedOperation = {
+      kind,
       start,
     }
 
     databaseQueue.queue.push(operation)
-    startOperationAsync(databaseQueue)
+    startNextOperations(databaseQueue)
   })
 }
 
-function startOperationAsync(queue: DatabaseQueue) {
-  // Queue is empty or in progress. Bail out.
-  if (queue.inProgress || queue.queue.length === 0) {
+function startNextOperations(queue: DatabaseQueue) {
+  if (queue.draining || (queue.inProgress && queue.activeStatements === 0)) {
     return
   }
 
-  queue.inProgress = true
+  queue.draining = true
+  try {
+    while (queue.queue.length > 0) {
+      const exclusiveIndex = queue.queue.findIndex(
+        (operation) => operation.kind === 'exclusive',
+      )
+      const statementCount =
+        exclusiveIndex === -1 ? queue.queue.length : exclusiveIndex
+
+      if (statementCount > 0) {
+        const statements = queue.queue.splice(0, statementCount)
+        queue.inProgress = true
+        queue.activeStatements += statements.length
+        for (const statement of statements) statement.start()
+        continue
+      }
 
-  const operation = queue.queue.shift()!
-  setImmediate(() => {
-    operation.start()
-  })
+      if (queue.activeStatements > 0) return
+
+      queue.inProgress = true
+      queue.queue.shift()!.start()
+      return
+    }
+  } finally {
+    queue.draining = false
+  }
 }
 
 export function startOperationSync<Result>(
@@ -115,5 +165,6 @@ export function startOperationSync<Result>(
     return callback()
   } finally {
     databaseQueue.inProgress = false
+    startNextOperations(databaseQueue)
   }
 }