Bind hooks and reactive flush to their runtime generation

cpp/DBHostObject.cpp

@@ -6,9 +6,9 @@
 #include "bridge.hpp"
 #endif
 #include "logs.h"
-#include <functional>
 #include "macros.hpp"
 #include "utils.hpp"
+#include <functional>
 #include <iostream>
 #include <utility>
 
@@ -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, {});
   });
@@ -32,14 +35,20 @@ std::string turso_remote_db_name(const std::string &url) {
 }
 
 void DBHostObject::flush_pending_reactive_queries(
-  const std::shared_ptr<jsi::Value> &resolve) {
+    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) {
@@ -207,8 +226,8 @@ DBHostObject::DBHostObject(jsi::Runtime &rt, std::string &db_name,
 
   thread_pool = std::make_shared<ThreadPool>();
 
-  db = opsqlite_open_sync(db_name, path, url, auth_token,
-                          remote_encryption_key);
+  db =
+      opsqlite_open_sync(db_name, path, url, auth_token, remote_encryption_key);
 
   create_jsi_functions(rt);
 }
@@ -242,12 +261,13 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     auto obj_params = args[0].asObject(rt);
 
     std::string secondary_db_name =
-      obj_params.getProperty(rt, "secondaryDbFileName").asString(rt).utf8(rt);
-    std::string alias = obj_params.getProperty(rt, "alias").asString(rt).utf8(rt);
+        obj_params.getProperty(rt, "secondaryDbFileName").asString(rt).utf8(rt);
+    std::string alias =
+        obj_params.getProperty(rt, "alias").asString(rt).utf8(rt);
 
     if (obj_params.hasProperty(rt, "location")) {
       std::string location =
-        obj_params.getProperty(rt, "location").asString(rt).utf8(rt);
+          obj_params.getProperty(rt, "location").asString(rt).utf8(rt);
       secondary_db_path = secondary_db_path + location;
     }
 
@@ -256,8 +276,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     // SQLite and Turso bind with explicit lengths. Failing loudly across
     // all backends keeps behaviour consistent.
     if (secondary_db_name.find('\0') != std::string::npos) {
-      throw std::runtime_error(
-          "[op-sqlite] attach secondaryDbFileName must not contain a zero byte");
+      throw std::runtime_error("[op-sqlite] attach secondaryDbFileName must "
+                               "not contain a zero byte");
     }
     if (alias.find('\0') != std::string::npos) {
       throw std::runtime_error(
@@ -303,7 +323,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     // Drain any in-flight async queries before closing the db handle.
     // Without this, a queued/running execute() on the thread pool may
     // dereference the freed sqlite3* pointer → heap corruption / SIGABRT.
-    thread_pool->waitFinished();
+    thread_pool->wait_finished();
 #ifdef OP_SQLITE_USE_LIBSQL
     opsqlite_libsql_close(db);
     db = {};
@@ -350,11 +370,11 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
 #endif
     // Drain any in-flight async queries before closing/removing the db handle.
     // Without this, queued/running work may dereference a freed sqlite handle.
-    thread_pool->waitFinished();
+    thread_pool->wait_finished();
 
     if (delete_db_name.empty()) {
-      throw std::runtime_error(
-        "[op-sqlite][delete] delete() is not supported for remote-only databases");
+      throw std::runtime_error("[op-sqlite][delete] delete() is not supported "
+                               "for remote-only databases");
     }
 
 #ifdef OP_SQLITE_USE_LIBSQL
@@ -625,7 +645,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     auto query = args[0].asObject(rt);
 
     const std::string query_str =
-      query.getProperty(rt, "query").asString(rt).utf8(rt);
+        query.getProperty(rt, "query").asString(rt).utf8(rt);
     auto js_args = query.getProperty(rt, "arguments");
     auto js_discriminators =
         query.getProperty(rt, "fireOn").asObject(rt).asArray(rt);
@@ -642,7 +662,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     for (size_t i = 0; i < js_discriminators.length(rt); i++) {
       auto js_discriminator =
           js_discriminators.getValueAtIndex(rt, i).asObject(rt);
-        std::string table =
+      std::string table =
           js_discriminator.getProperty(rt, "table").asString(rt).utf8(rt);
       std::vector<int> ids;
       if (js_discriminator.hasProperty(rt, "ids")) {
@@ -725,7 +745,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
         flush_pending_reactive_queries(resolve);
       };
 
-      thread_pool->queueWork(task);
+      thread_pool->queue_work(task);
 
       return {};
     }));
@@ -768,12 +788,20 @@ 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
-  // blocks until the queue is empty and no worker is busy, then the
-  // ThreadPool destructor (via shared_ptr release) joins the threads.
-  thread_pool->waitFinished();
+  thread_pool->wait_finished();
 
 #ifdef OP_SQLITE_USE_LIBSQL
   opsqlite_libsql_close(db);

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);
   });