Bind async work to its JS runtime generation

cpp/DBHostObject.cpp

@@ -20,6 +20,9 @@ namespace react = facebook::react;
 #ifdef OP_SQLITE_USE_LIBSQL
 void DBHostObject::flush_pending_reactive_queries(
     const std::shared_ptr<jsi::Value> &resolve) {
+  if (alive != nullptr && !alive->load()) {
+    return;
+  }
   invoker->invokeAsync([resolve](jsi::Runtime &rt) {
     resolve->asObject(rt).asFunction(rt).call(rt, {});
   });
@@ -33,13 +36,19 @@ std::string turso_remote_db_name(const std::string &url) {
 
 void DBHostObject::flush_pending_reactive_queries(
   const std::shared_ptr<jsi::Value> &resolve) {
+  if (alive != nullptr && !alive->load()) {
+    return;
+  }
   invoker->invokeAsync([resolve](jsi::Runtime &rt) {
     resolve->asObject(rt).asFunction(rt).call(rt, {});
   });
 }
 #else
 void DBHostObject::flush_pending_reactive_queries(
     const std::shared_ptr<jsi::Value> &resolve) {
+  if (alive != nullptr && !alive->load()) {
+    return;
+  }
   for (const auto &query_ptr : pending_reactive_queries) {
     auto query = query_ptr.get();
 
@@ -67,19 +76,29 @@ void DBHostObject::flush_pending_reactive_queries(
 }
 
 void DBHostObject::on_commit() {
+  if (alive != nullptr && !alive->load()) {
+    return;
+  }
   invoker->invokeAsync([this](jsi::Runtime &rt) {
     commit_hook_callback->asObject(rt).asFunction(rt).call(rt);
   });
 }
 
 void DBHostObject::on_rollback() {
+  if (alive != nullptr && !alive->load()) {
+    return;
+  }
   invoker->invokeAsync([this](jsi::Runtime &rt) {
     rollback_hook_callback->asObject(rt).asFunction(rt).call(rt);
   });
 }
 
 void DBHostObject::on_update(const std::string &table,
                              const std::string &operation, long long row_id) {
+  if (alive != nullptr && !alive->load()) {
+    return;
+  }
+
   if (update_hook_callback != nullptr) {
     invoker->invokeAsync([callback = update_hook_callback, table, operation,
                           row_id](jsi::Runtime &rt) {
@@ -768,6 +787,18 @@ void DBHostObject::invalidate() {
   }
 
   invalidated = true;
+
+  // Abort whatever is currently inside sqlite3_step so the drain below can
+  // actually finish. Parity with the close and delete host functions, which
+  // already do this. Without it a long running query holds the pool past React
+  // Native's module invalidation budget, after which the runtime is destroyed
+  // anyway and the drain has bought nothing.
+#if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO)
+  if (db != nullptr) {
+    sqlite3_interrupt(db);
+  }
+#endif
+
   // Drain in-flight thread pool work before closing the db handle.
   // restartPool() joins threads (waiting for the current task) but then
   // needlessly re-creates the pool. waitFinished() is sufficient: it

cpp/DBHostObject.hpp

@@ -87,6 +87,18 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject {
 
   std::unordered_map<std::string, jsi::Value> function_map;
   std::string base_path;
+  // Bound at construction, on the JS thread, to the generation that created
+  // this database.
+  //
+  // NOTE: these deliberately shadow the process-global opsqlite::invoker and
+  // opsqlite::generation_alive inside every member function, which is what
+  // fixes the update/commit/rollback hooks and flush_pending_reactive_queries
+  // without touching each call site. Reading the globals at callback time
+  // instead lets a database belonging to a torn-down runtime post work into the
+  // runtime that replaced it, and then call asFunction() on a jsi::Value owned
+  // by the dead one.
+  std::shared_ptr<react::CallInvoker> invoker = opsqlite::invoker;
+  std::shared_ptr<std::atomic<bool>> alive = opsqlite::generation_alive;
   std::shared_ptr<ThreadPool> thread_pool;
   std::string db_name;
   std::string delete_db_name;

cpp/OPSqlite.cpp

@@ -12,6 +12,7 @@
 #include "utils.hpp"
 #include <functional>
 #include <iostream>
+#include <mutex>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -25,22 +26,42 @@ std::string _base_path;
 std::string _crsqlite_path;
 std::string _sqlite_vec_path;
 std::vector<std::shared_ptr<DBHostObject>> dbs;
+// Guards `dbs`. Two JS runtime generations overlap during a bridgeless reload,
+// so open() and invalidate() can touch this vector from different threads at
+// the same time.
+std::mutex dbs_mutex;
 bool invalidated = false;
 std::shared_ptr<react::CallInvoker> invoker;
+std::shared_ptr<std::atomic<bool>> generation_alive;
 
 // React native will try to clean the module on JS context invalidation
 // (CodePush/Hot Reload) The clearState function is called
 void invalidate() {
   // Global flag used by the threads to stop work
   invalidated = true;
 
-  for (const auto &db : dbs) {
-    db->invalidate();
+  // Mark THIS generation dead. Work queued by it holds a copy of the flag, so
+  // it drops its completions instead of resolving into a runtime that is being
+  // torn down.
+  if (generation_alive != nullptr) {
+    generation_alive->store(false);
+  }
+
+  // Take ownership of the registry under the lock before touching it. This runs
+  // on the outgoing generation's TurboModule queue, while the incoming
+  // generation's open() may already be emplacing into `dbs` on its own JS
+  // thread: RCTHost constructs the new RCTInstance without waiting for the old
+  // one to finish invalidating. Iterating the vector directly can therefore run
+  // off a reallocated buffer.
+  std::vector<std::shared_ptr<DBHostObject>> closing;
+  {
+    std::lock_guard<std::mutex> g(dbs_mutex);
+    closing.swap(dbs);
   }
 
-  // Clear our existing vector of shared pointers so they can be garbage
-  // collected
-  dbs.clear();
+  for (const auto &db : closing) {
+    db->invalidate();
+  }
 }
 
 void install(jsi::Runtime &rt,
@@ -53,6 +74,7 @@ void install(jsi::Runtime &rt,
   _sqlite_vec_path = std::string(sqlite_vec_path);
   opsqlite::invoker = _invoker;
   opsqlite::invalidated = false;
+  opsqlite::generation_alive = std::make_shared<std::atomic<bool>>(true);
 
   auto open = HFN0 {
     jsi::Object options = args[0].asObject(rt);
@@ -92,7 +114,10 @@ void install(jsi::Runtime &rt,
 
     std::shared_ptr<DBHostObject> db = std::make_shared<DBHostObject>(
         rt, path, name, path, readOnly, failOnCreate, encryption_key);
-    dbs.emplace_back(db);
+    {
+      std::lock_guard<std::mutex> g(dbs_mutex);
+      dbs.emplace_back(db);
+    }
     return jsi::Object::createFromHostObject(rt, db);
   });
 
@@ -146,7 +171,10 @@ void install(jsi::Runtime &rt,
         std::make_shared<DBHostObject>(rt, url, auth_token, path);
 #endif
 
-    dbs.emplace_back(db);
+    {
+      std::lock_guard<std::mutex> g(dbs_mutex);
+      dbs.emplace_back(db);
+    }
 
     return jsi::Object::createFromHostObject(rt, db);
   });
@@ -208,7 +236,10 @@ void install(jsi::Runtime &rt,
       rt, name, path, url, auth_token, remote_encryption_key);
   #endif
 
-    dbs.emplace_back(db);
+    {
+      std::lock_guard<std::mutex> g(dbs_mutex);
+      dbs.emplace_back(db);
+    }
 
     return jsi::Object::createFromHostObject(rt, db);
   });

cpp/OPThreadPool.cpp

@@ -31,6 +31,16 @@ ThreadPool::~ThreadPool() {
   workQueueConditionVariable.notify_all();
 
   for (auto &thread : threads) {
+    // Never join ourselves. If the pool's last owner is released on one of its
+    // own workers, join() throws std::system_error ("thread::join failed:
+    // Resource deadlock avoided") and takes the process with it. Not capturing
+    // the pool in the promisify task should make this unreachable; this is a
+    // backstop, not the fix.
+    if (thread.get_id() == std::this_thread::get_id()) {
+      thread.detach();
+      continue;
+    }
+
     if (thread.joinable()) {
       thread.join();
     }
@@ -48,8 +58,10 @@ void ThreadPool::queueWork(const std::function<void(void)> &task) {
   // Push the request to the queue
   workQueue.push(task);
 
-  // Notify one thread that there are requests to process
-  workQueueConditionVariable.notify_one();
+  // Wake every waiter. waitFinished() and doWork() share this condition
+  // variable, so notify_one() can hand the wakeup to the wrong one and leave
+  // the other asleep.
+  workQueueConditionVariable.notify_all();
 }
 
 // Function used by the threads to grab work from the queue
@@ -77,11 +89,15 @@ void ThreadPool::doWork() {
       ++busy;
     }
     task();
+    // Release the task (and everything it captured, e.g. JSI values) before
+    // signalling idle, so waitFinished()/close() can't observe busy == 0
+    // while task-owned resources are still pending destruction.
+    task = nullptr;
     {
       std::lock_guard<std::mutex> g(workQueueMutex);
       --busy;
     }
-    workQueueConditionVariable.notify_one();
+    workQueueConditionVariable.notify_all();
   }
 }
 

cpp/OPThreadPool.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <condition_variable>
 #include <exception>
 #include <mutex>
@@ -34,8 +35,9 @@ class ThreadPool {
   std::queue<std::function<void(void)>> workQueue;
 
   // This will be set to true when the thread pool is shutting down. This
-  // tells the threads to stop looping and finish
-  bool done;
+  // tells the threads to stop looping and finish.
+  // Atomic because doWork() reads it in `while (!done)` outside the mutex.
+  std::atomic<bool> done;
 
   // Function used by the threads to grab work from the queue
   void doWork();

cpp/types.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ReactCommon/CallInvoker.h>
+#include <atomic>
 #include <memory>
 #include <sqlite3.h>
 #include <string>
@@ -12,6 +13,18 @@ namespace opsqlite {
 extern std::shared_ptr<facebook::react::CallInvoker> invoker;
 extern bool invalidated;
 
+// Liveness of the current JS runtime generation. Replaced by install() and
+// cleared by invalidate(), so each generation gets its own flag rather than
+// sharing the process-global `invalidated` bool.
+//
+// Whoever queues work copies the shared_ptr when the work is created, so it
+// always observes ITS OWN generation's liveness. Checking a process-global
+// instead is wrong in both directions during a bridgeless reload, where two
+// generations overlap: an outgoing generation clearing it would suppress the
+// incoming generation's callbacks, and an incoming generation setting it would
+// re-enable the outgoing generation's.
+extern std::shared_ptr<std::atomic<bool>> generation_alive;
+
 struct ArrayBuffer {
   std::shared_ptr<uint8_t[]> data;
   size_t size;

cpp/utils.cpp

@@ -170,8 +170,8 @@ std::vector<JSVariant> to_variant_vec(jsi::Runtime &rt, jsi::Value const &xs) {
   res.reserve(arg_length);
 
   for (size_t ii = 0; ii < arg_length; ii++) {
-     res.emplace_back(to_variant(rt, values.getValueAtIndex(rt, ii)));
-   }
+    res.emplace_back(to_variant(rt, values.getValueAtIndex(rt, ii)));
+  }
 
   return res;
 }
@@ -383,18 +383,35 @@ promisify(jsi::Runtime &rt, std::shared_ptr<ThreadPool> thread_pool,
     auto resolve = std::make_shared<jsi::Value>(rt, args[0]);
     auto reject = std::make_shared<jsi::Value>(rt, args[1]);
 
-    auto task = [lambda = lambda, thread_pool,
-                 resolve_callback = resolve_callback,
-                 resolve = std::move(resolve), reject = std::move(reject)]() {
+    // Bind this generation's invoker and liveness flag here, on the JS thread,
+    // while the promise is being constructed. Reading the process globals from
+    // the worker instead lets a task queued by a torn-down runtime post into the
+    // runtime that replaced it, and then call asFunction() on a jsi::Value that
+    // belongs to the dead one.
+    auto invoker = opsqlite::invoker;
+    auto alive = opsqlite::generation_alive;
+
+    auto task = [lambda = lambda, resolve_callback = resolve_callback,
+                 resolve = std::move(resolve), reject = std::move(reject),
+                 invoker, alive]() {
+      if (invoker == nullptr) {
+        return;
+      }
+
       try {
         std::any result = lambda();
 
-        if (opsqlite::invalidated) {
+        // This generation is gone. Posting now would schedule onto a runtime
+        // that is being torn down, where asFunction() sees an already
+        // invalidated PointerValue.
+        if (alive != nullptr && !alive->load()) {
           return;
         }
 
-        opsqlite::invoker->invokeAsync(
-            [result = std::move(result), resolve = resolve,
+        // reject is also captured in the invokeAsync lambda
+        // so it can be safely disposed on the JS thread
+        invoker->invokeAsync(
+            [result = std::move(result), resolve = resolve, reject = reject,
              resolve_callback = resolve_callback](jsi::Runtime &rt) {
               auto jsi_result = resolve_callback(rt, result);
               resolve->asObject(rt).asFunction(rt).call(rt, jsi_result);
@@ -404,23 +421,34 @@ promisify(jsi::Runtime &rt, std::shared_ptr<ThreadPool> thread_pool,
         // runtime_error to the generic exception We have to
         // explicitly catch it
         // https://github.com/facebook/react-native/issues/48027
+        //
+        // resolve is also captured in the invokeAsync lambda
+        // so it can be safely disposed on the JS thread
         auto what = e.what();
-        opsqlite::invoker->invokeAsync(
-            [what = std::string(what), reject = reject](jsi::Runtime &rt) {
-              auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error");
-              auto error = errorCtr.callAsConstructor(
-                  rt, jsi::String::createFromAscii(rt, what));
-              reject->asObject(rt).asFunction(rt).call(rt, error);
-            });
+        if (alive != nullptr && !alive->load()) {
+          return;
+        }
+        invoker->invokeAsync([what = std::string(what), resolve = resolve,
+                              reject = reject](jsi::Runtime &rt) {
+          auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error");
+          auto error = errorCtr.callAsConstructor(
+              rt, jsi::String::createFromAscii(rt, what));
+          reject->asObject(rt).asFunction(rt).call(rt, error);
+        });
       } catch (std::exception &exc) {
         auto what = exc.what();
-        opsqlite::invoker->invokeAsync(
-            [what = std::string(what), reject = reject](jsi::Runtime &rt) {
-              auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error");
-              auto error = errorCtr.callAsConstructor(
-                  rt, jsi::String::createFromAscii(rt, what));
-              reject->asObject(rt).asFunction(rt).call(rt, error);
-            });
+        if (alive != nullptr && !alive->load()) {
+          return;
+        }
+        // resolve is also captured in the invokeAsync lambda
+        // so it can be safely disposed on the JS thread
+        invoker->invokeAsync([what = std::string(what), resolve = resolve,
+                              reject = reject](jsi::Runtime &rt) {
+          auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error");
+          auto error = errorCtr.callAsConstructor(
+              rt, jsi::String::createFromAscii(rt, what));
+          reject->asObject(rt).asFunction(rt).call(rt, error);
+        });
       }
     };