improve commit hooks on close. Improve closing and thread management. improve errors for opening a DB.

Author: stevensJourney

cpp/ConnectionPool.cpp

@@ -169,6 +169,14 @@ void ConnectionPool::closeContext(ConnectionLockId contextId) {
 }
 
 void ConnectionPool::closeAll() {
+  // Prevent this pointer from being used if closed
+  rollbackPayload.dbName = NULL;
+  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

@@ -66,8 +66,8 @@ class ConnectionPool {
   std::vector<ConnectionLockId> writeQueue;
 
   // Cached constant payloads for c style commit/rollback callbacks
-  const TransactionCallbackPayload commitPayload;
-  const TransactionCallbackPayload rollbackPayload;
+  TransactionCallbackPayload commitPayload;
+  TransactionCallbackPayload rollbackPayload;
 
   void (*onContextCallback)(std::string, ConnectionLockId);
   void (*onCommitCallback)(const TransactionCallbackPayload *);

cpp/ConnectionState.cpp

@@ -10,24 +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);
-  isClosed = false;
-
-  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() {
@@ -65,10 +58,25 @@ std::future<void> ConnectionState::refreshSchema() {
 }
 
 void ConnectionState::close() {
+  // 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::lock_guard<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);
 }
 
@@ -77,14 +85,12 @@ void ConnectionState::queueWork(std::function<void(sqlite3 *)> task) {
     throw std::runtime_error("Connection is not open. Connection has been closed before queueing work.");
   }
 
-  // Grab the mutex
-  std::lock_guard<std::mutex> g(workQueueMutex);
-
-  // Push the request to the queue
-  workQueue.push(task);
+  {
+    std::lock_guard<std::mutex> g(workQueueMutex);
+    workQueue.push(task);
+  }
 
-  // Notify one thread that there are requests to process
-  workQueueConditionVariable.notify_all();
+  workQueueConditionVariable.notify_all(); 
 }
 
 void ConnectionState::doWork() {
@@ -124,11 +130,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 == false); });
+      g, [&] { return workQueue.empty() && !threadBusy; });
 }
 
 SQLiteOPResult genericSqliteOpenDb(string const dbName, string const docPath,

cpp/ConnectionState.h

@@ -25,15 +25,15 @@ 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;
-  bool threadBusy;
-  bool threadDone;
+  std::atomic<bool> threadBusy{false};
+  std::atomic<bool> threadDone{false};
 
 public:
-  bool isClosed;
+  std::atomic<bool> isClosed{false};
 
   ConnectionState(const std::string dbName, const std::string docPath,
                   int SQLFlags);

cpp/bindings.cpp

@@ -79,6 +79,11 @@ void transactionFinalizerHandler(const TransactionCallbackPayload *payload) {
    */
   invoker->invokeAsync([payload] {
     try {
+      // Prevent trying to create a JSI string for a potentially closed DB
+      if (payload == NULL || payload->dbName == NULL) {
+        return;
+      }
+      
       auto global = runtime->global();
       jsi::Function handlerFunction = global.getPropertyAsFunction(
           *runtime, "triggerTransactionFinalizerHook");

cpp/sqliteBridge.cpp

@@ -50,10 +50,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,

tests/tests/sqlite/rawQueries.spec.ts

@@ -71,7 +71,6 @@ export function registerBaseTests() {
 
       await db.execute('DROP TABLE IF EXISTS User; ');
       await db.execute('CREATE TABLE User ( id INT PRIMARY KEY, name TEXT NOT NULL, age INT, networth REAL) STRICT;');
-
       await db.execute('CREATE TABLE IF NOT EXISTS t1(id INTEGER PRIMARY KEY, a INTEGER, b INTEGER, c TEXT)');
     } catch (e) {
       console.warn('error on before each', e);