Fix Potential crash on db close

.changeset/purple-drinks-allow.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/react-native-quick-sqlite': minor
+---
+
+Improved behaviour for closing a database connection. This should prevent some crash issues.

.github/workflows/test.yaml

@@ -58,6 +58,9 @@ jobs:
           cd tests
           yarn install --frozen-lockfile
 
+      - name: Initialize Android Folder
+        run: mkdir -p ~/.android/avd
+
       - name: create AVD and generate snapshot for caching
         if: steps.avd-cache.outputs.cache-hit != 'true'
         uses: reactivecircus/[email protected]

cpp/ConnectionPool.cpp

@@ -19,6 +19,7 @@ ConnectionPool::ConnectionPool(std::string dbName, std::string docPath,
 
   onContextCallback = nullptr;
   isConcurrencyEnabled = maxReads > 0;
+  isClosed = false;
 
   readConnections = new ConnectionState *[maxReads];
   // Open the read connections
@@ -94,7 +95,14 @@ ConnectionPool::queueInContext(ConnectionLockId contextId,
     };
   }
 
-  state->queueWork(task);
+  try {
+    state->queueWork(task);
+  } catch (const std::exception &e) {
+    return SQLiteOPResult{
+        .errorMessage = e.what(),
+        .type = SQLiteError,
+    };
+  }
 
   return SQLiteOPResult{
       .type = SQLiteOk,
@@ -162,6 +170,14 @@ void ConnectionPool::closeContext(ConnectionLockId contextId) {
 }
 
 void ConnectionPool::closeAll() {
+  isClosed = true;
+  // Stop any callbacks
+  sqlite3_commit_hook(writeConnection.connection,
+                      NULL, NULL);
+  sqlite3_rollback_hook(writeConnection.connection, 
+                        NULL, NULL);
+  sqlite3_update_hook(writeConnection.connection, 
+                        NULL, NULL);
   writeConnection.close();
   for (int i = 0; i < maxReads; i++) {
     readConnections[i]->close();

cpp/ConnectionPool.h

@@ -75,6 +75,8 @@ class ConnectionPool {
   bool isConcurrencyEnabled;
 
 public:
+  bool isClosed;
+
   ConnectionPool(std::string dbName, std::string docPath,
                  unsigned int numReadConnections);
   ~ConnectionPool();

cpp/ConnectionState.cpp

@@ -10,23 +10,17 @@ SQLiteOPResult genericSqliteOpenDb(string const dbName, string const docPath,
 ConnectionState::ConnectionState(const std::string dbName,
                                  const std::string docPath, int SQLFlags) {
   auto result = genericSqliteOpenDb(dbName, docPath, &connection, SQLFlags);
-
-  this->clearLock();
-  threadDone = false;
-  thread = new std::thread(&ConnectionState::doWork, this);
+   if (result.type != SQLiteOk) {
+    throw std::runtime_error("Failed to open SQLite database: " + result.errorMessage);
+  }
+   thread = std::thread(&ConnectionState::doWork, this);
+   this->clearLock();
 }
 
 ConnectionState::~ConnectionState() {
-  // So threads know it's time to shut down
-  threadDone = true;
-
-  // Wake up all the threads, so they can finish and be joined
-  workQueueConditionVariable.notify_all();
-  if (thread->joinable()) {
-    thread->join();
+  if (!isClosed) {
+    close();
   }
-
-  delete thread;
 }
 
 void ConnectionState::clearLock() {
@@ -64,21 +58,41 @@ std::future<void> ConnectionState::refreshSchema() {
 }
 
 void ConnectionState::close() {
+  {
+    std::unique_lock<std::mutex> g(workQueueMutex);
+    // prevent any new work from being queued 
+    isClosed = true;
+  }
+
+  // Wait for the work queue to empty
   waitFinished();
-  // So that the thread can stop (if not already)
-  threadDone = true;
+
+  {
+    // Now signal the thread to stop and notify it
+    std::unique_lock<std::mutex> g(workQueueMutex);
+    threadDone = true;
+    workQueueConditionVariable.notify_all();
+  }
+
+  // Join the worker thread
+  if (thread.joinable()) {
+    thread.join();
+  }
+
+  // Safely close the SQLite connection
   sqlite3_close_v2(connection);
 }
 
 void ConnectionState::queueWork(std::function<void(sqlite3 *)> task) {
-  // Grab the mutex
-  std::lock_guard<std::mutex> g(workQueueMutex);
-
-  // Push the request to the queue
-  workQueue.push(task);
+  {
+    std::unique_lock<std::mutex> g(workQueueMutex);
+    if (isClosed) {
+      throw std::runtime_error("Connection is not open. Connection has been closed before queueing work.");
+    }
+    workQueue.push(task);
+  }
 
-  // Notify one thread that there are requests to process
-  workQueueConditionVariable.notify_all();
+  workQueueConditionVariable.notify_all(); 
 }
 
 void ConnectionState::doWork() {
@@ -104,9 +118,9 @@ void ConnectionState::doWork() {
       workQueue.pop();
     }
 
-    ++threadBusy;
+    threadBusy = true;
     task(connection);
-    --threadBusy;
+    threadBusy = false;
     // Need to notify in order for waitFinished to be updated when
     // the queue is empty and not busy
     {
@@ -118,11 +132,8 @@ void ConnectionState::doWork() {
 
 void ConnectionState::waitFinished() {
   std::unique_lock<std::mutex> g(workQueueMutex);
-  if (workQueue.empty()) {
-    return;
-  }
   workQueueConditionVariable.wait(
-      g, [&] { return workQueue.empty() && (threadBusy == 0); });
+      g, [&] { return workQueue.empty() && !threadBusy; });
 }
 
 SQLiteOPResult genericSqliteOpenDb(string const dbName, string const docPath,

cpp/ConnectionState.h

@@ -25,14 +25,16 @@ class ConnectionState {
   // Mutex to protect workQueue
   std::mutex workQueueMutex;
   // Store thread in order to stop it gracefully
-  std::thread *thread;
+  std::thread thread;
   // This condition variable is used for the threads to wait until there is work
   // to do
   std::condition_variable_any workQueueConditionVariable;
-  unsigned int threadBusy;
-  bool threadDone;
+  std::atomic<bool> threadBusy{false};
+  std::atomic<bool> threadDone{false};
 
 public:
+  std::atomic<bool> isClosed{false};
+
   ConnectionState(const std::string dbName, const std::string docPath,
                   int SQLFlags);
   ~ConnectionState();

cpp/bindings.cpp

@@ -77,17 +77,30 @@ void transactionFinalizerHandler(const TransactionCallbackPayload *payload) {
    * This function triggers an async invocation to call watch callbacks,
    * avoiding holding SQLite up.
    */
-  invoker->invokeAsync([payload] {
+
+  // Make a copy of the payload data, this avoids a potential race condition
+  // where the async invocation might occur after closing a connection
+  auto dbName = std::make_shared<std::string>(*payload->dbName);
+  int event = payload->event;
+  invoker->invokeAsync([dbName, event] {
     try {
+
+     ConnectionPool* connection = getConnection(*dbName);
+      if (connection == nullptr || connection->isClosed) {
+        return;
+      }
+
       auto global = runtime->global();
       jsi::Function handlerFunction = global.getPropertyAsFunction(
           *runtime, "triggerTransactionFinalizerHook");
 
-      auto jsiDbName = jsi::String::createFromAscii(*runtime, *payload->dbName);
-      auto jsiEventType = jsi::Value((int)payload->event);
+      auto jsiDbName = jsi::String::createFromAscii(*runtime, *dbName);
+      auto jsiEventType = jsi::Value(event);
       handlerFunction.call(*runtime, move(jsiDbName), move(jsiEventType));
     } catch (jsi::JSINativeException e) {
       std::cout << e.what() << std::endl;
+    } catch (const std::exception& e) {
+      std::cout << "Standard Exception: " << e.what() << std::endl;
     } catch (...) {
       std::cout << "Unknown error" << std::endl;
     }
@@ -384,7 +397,14 @@ void osp::install(jsi::Runtime &rt,
         }
       };
 
-      sqliteQueueInContext(dbName, contextLockId, task);
+      auto response = sqliteQueueInContext(dbName, contextLockId, task);
+      if (response.type == SQLiteError) {
+        auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error");
+        auto error = errorCtr.callAsConstructor(
+                      rt, jsi::String::createFromUtf8(
+                      rt, response.errorMessage));
+        reject->asObject(rt).asFunction(rt).call(rt, error);
+      }
       return {};
     }));
 

cpp/sqliteBridge.cpp

@@ -32,6 +32,16 @@ SQLiteOPResult generateNotOpenResult(std::string const &dbName) {
   };
 }
 
+ConnectionPool *getConnection(std::string const dbName) {
+  if (dbMap.count(dbName) == 0) {
+    // Connection is already closed
+    return nullptr;
+  }
+
+  return dbMap[dbName];
+}
+
+
 /**
  * Opens SQL database with default settings
  */
@@ -50,10 +60,18 @@ sqliteOpenDb(string const dbName, string const docPath,
     };
   }
 
-  dbMap[dbName] = new ConnectionPool(dbName, docPath, numReadConnections);
-  dbMap[dbName]->setOnContextAvailable(contextAvailableCallback);
-  dbMap[dbName]->setTableUpdateHandler(updateTableCallback);
-  dbMap[dbName]->setTransactionFinalizerHandler(onTransactionFinalizedCallback);
+  try {
+    // Open the database
+    dbMap[dbName] = new ConnectionPool(dbName, docPath, numReadConnections);
+    dbMap[dbName]->setOnContextAvailable(contextAvailableCallback);
+    dbMap[dbName]->setTableUpdateHandler(updateTableCallback);
+    dbMap[dbName]->setTransactionFinalizerHandler(onTransactionFinalizedCallback);
+  } catch (const std::exception &e) {
+    return SQLiteOPResult{
+        .type = SQLiteError,
+        .errorMessage = e.what(),
+    };
+  }
 
   return SQLiteOPResult{
       .type = SQLiteOk,
@@ -126,13 +144,6 @@ SQLiteOPResult sqliteRequestLock(std::string const dbName,
 
   ConnectionPool *connection = dbMap[dbName];
 
-  if (connection == nullptr) {
-    return SQLiteOPResult{
-        .type = SQLiteOk,
-
-    };
-  }
-
   switch (lockType) {
   case ConcurrentLockType::ReadLock:
     connection->readLock(contextId);
@@ -147,6 +158,7 @@ SQLiteOPResult sqliteRequestLock(std::string const dbName,
 
   return SQLiteOPResult{
       .type = SQLiteOk,
+
   };
 }
 

cpp/sqliteBridge.h

@@ -40,6 +40,8 @@ SQLiteOPResult sqliteCloseDb(string const dbName);
 
 void sqliteCloseAll();
 
+ConnectionPool *getConnection(std::string const dbName);
+
 SQLiteOPResult sqliteRemoveDb(string const dbName, string const docPath);
 
 /**

src/DBListenerManager.ts

@@ -39,6 +39,8 @@ export interface DBListener extends BaseListener {
    * is started, committed or rolled back.
    */
   writeTransaction: (event: WriteTransactionEvent) => void;
+
+  closed: () => void;
 }
 
 export class DBListenerManager extends BaseObserver<DBListener> {}