diff --git a/cpp/DBHostObject.cpp b/cpp/DBHostObject.cpp index b3fbcc37..c2b2f631 100644 --- a/cpp/DBHostObject.cpp +++ b/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 &resolve) { + if (alive != nullptr && !alive->load()) { + return; + } invoker->invokeAsync([resolve](jsi::Runtime &rt) { resolve->asObject(rt).asFunction(rt).call(rt, {}); }); @@ -33,6 +36,9 @@ std::string turso_remote_db_name(const std::string &url) { void DBHostObject::flush_pending_reactive_queries( const std::shared_ptr &resolve) { + if (alive != nullptr && !alive->load()) { + return; + } invoker->invokeAsync([resolve](jsi::Runtime &rt) { resolve->asObject(rt).asFunction(rt).call(rt, {}); }); @@ -40,6 +46,9 @@ void DBHostObject::flush_pending_reactive_queries( #else void DBHostObject::flush_pending_reactive_queries( const std::shared_ptr &resolve) { + if (alive != nullptr && !alive->load()) { + return; + } for (const auto &query_ptr : pending_reactive_queries) { auto query = query_ptr.get(); @@ -67,12 +76,18 @@ 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); }); @@ -80,6 +95,10 @@ void DBHostObject::on_rollback() { 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 diff --git a/cpp/DBHostObject.hpp b/cpp/DBHostObject.hpp index 19b1180d..81ba8281 100644 --- a/cpp/DBHostObject.hpp +++ b/cpp/DBHostObject.hpp @@ -87,6 +87,18 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject { std::unordered_map 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 invoker = opsqlite::invoker; + std::shared_ptr> alive = opsqlite::generation_alive; std::shared_ptr thread_pool; std::string db_name; std::string delete_db_name; diff --git a/cpp/OPSqlite.cpp b/cpp/OPSqlite.cpp index 31401f2a..79f228f2 100644 --- a/cpp/OPSqlite.cpp +++ b/cpp/OPSqlite.cpp @@ -12,6 +12,7 @@ #include "utils.hpp" #include #include +#include #include #include #include @@ -25,8 +26,13 @@ std::string _base_path; std::string _crsqlite_path; std::string _sqlite_vec_path; std::vector> 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 invoker; +std::shared_ptr> generation_alive; // React native will try to clean the module on JS context invalidation // (CodePush/Hot Reload) The clearState function is called @@ -34,13 +40,28 @@ 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> closing; + { + std::lock_guard 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>(true); auto open = HFN0 { jsi::Object options = args[0].asObject(rt); @@ -92,7 +114,10 @@ void install(jsi::Runtime &rt, std::shared_ptr db = std::make_shared( rt, path, name, path, readOnly, failOnCreate, encryption_key); - dbs.emplace_back(db); + { + std::lock_guard 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(rt, url, auth_token, path); #endif - dbs.emplace_back(db); + { + std::lock_guard 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 g(dbs_mutex); + dbs.emplace_back(db); + } return jsi::Object::createFromHostObject(rt, db); }); diff --git a/cpp/OPThreadPool.cpp b/cpp/OPThreadPool.cpp index e9b06857..d44960c1 100644 --- a/cpp/OPThreadPool.cpp +++ b/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 &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 g(workQueueMutex); --busy; } - workQueueConditionVariable.notify_one(); + workQueueConditionVariable.notify_all(); } } diff --git a/cpp/OPThreadPool.hpp b/cpp/OPThreadPool.hpp index 9405681d..6cadfea3 100644 --- a/cpp/OPThreadPool.hpp +++ b/cpp/OPThreadPool.hpp @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -34,8 +35,9 @@ class ThreadPool { std::queue> 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 done; // Function used by the threads to grab work from the queue void doWork(); diff --git a/cpp/types.hpp b/cpp/types.hpp index 6c2ff2df..5844e615 100644 --- a/cpp/types.hpp +++ b/cpp/types.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -12,6 +13,18 @@ namespace opsqlite { extern std::shared_ptr 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> generation_alive; + struct ArrayBuffer { std::shared_ptr data; size_t size; diff --git a/cpp/utils.cpp b/cpp/utils.cpp index 1f2ad590..0bf6d2d9 100644 --- a/cpp/utils.cpp +++ b/cpp/utils.cpp @@ -170,8 +170,8 @@ std::vector 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 thread_pool, auto resolve = std::make_shared(rt, args[0]); auto reject = std::make_shared(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 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); + }); } };