Skip to content

fix: don't touch a closed connection when unsubscribing a reactive query #442

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ospfranco merged 3 commits into from
Aug 20, 2026
Merged

fix: don't touch a closed connection when unsubscribing a reactive query #442

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f2f7b8d
fix: don't touch a closed connection when unsubscribing a reactive query
mlecoq Aug 20, 2026
3394195
add test
mlecoq Aug 20, 2026
5acb90f
Throw instead of crashing when the database is used after close
mlecoq Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 76 additions & 7 deletions cpp/DBHostObject.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,10 @@ void DBHostObject::on_update(const std::string &table,
}

void DBHostObject::auto_register_update_hook() {
if (invalidated || db == nullptr) {
return;
}

if (update_hook_callback == nullptr && reactive_queries.empty() &&
is_update_hook_registered) {
opsqlite_deregister_update_hook(db);
Expand All @@ -171,6 +175,22 @@ void DBHostObject::auto_register_update_hook() {
}
#endif

void DBHostObject::throw_if_closed(const char *function_name) const {
if (invalidated) {
throw std::runtime_error(std::string("[op-sqlite][") + function_name +
"] database is closed");
}
}

void DBHostObject::release_hooks() {
reactive_queries.clear();
pending_reactive_queries.clear();
update_hook_callback = nullptr;
commit_hook_callback = nullptr;
rollback_hook_callback = nullptr;
is_update_hook_registered = false;
}

// _____ _ _
// / ____| | | | |
// | | ___ _ __ ___| |_ _ __ _ _ ___| |_ ___ _ __
Expand Down Expand Up @@ -256,6 +276,8 @@ DBHostObject::DBHostObject(jsi::Runtime &rt, std::string &base_path,

void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
function_map["attach"] = HFN(this) {
throw_if_closed("attach");

std::string secondary_db_path = std::string(base_path);

auto obj_params = args[0].asObject(rt);
Expand Down Expand Up @@ -294,6 +316,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["detach"] = HFN(this) {
throw_if_closed("detach");

if (!args[0].isString()) {
throw std::runtime_error("[op-sqlite] alias must be a strings");
}
Expand Down Expand Up @@ -324,6 +348,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
// Without this, a queued/running execute() on the thread pool may
// dereference the freed sqlite3* pointer → heap corruption / SIGABRT.
thread_pool->wait_finished();
release_hooks();
#ifdef OP_SQLITE_USE_LIBSQL
opsqlite_libsql_close(db);
db = {};
Expand Down Expand Up @@ -377,16 +402,21 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
"for remote-only databases");
}

release_hooks();
#ifdef OP_SQLITE_USE_LIBSQL
opsqlite_libsql_remove(db, delete_db_name, base_path);
#else
opsqlite_remove(db, delete_db_name, base_path);
auto *closing_db = db;
db = nullptr;
opsqlite_remove(closing_db, delete_db_name, base_path);
#endif

return {};
});

function_map["executeRaw"] = HFN(this) {
throw_if_closed("executeRaw");

const std::string query = args[0].asString(rt).utf8(rt);
const std::vector<JSVariant> params = count == 2 && args[1].isObject()
? to_variant_vec(rt, args[1])
Expand Down Expand Up @@ -414,6 +444,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["executeSync"] = HFN(this) {
throw_if_closed("executeSync");

std::string query = args[0].asString(rt).utf8(rt);
std::vector<JSVariant> params;

Expand All @@ -430,6 +462,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["executeRawSync"] = HFN(this) {
throw_if_closed("executeRawSync");

const std::string query = args[0].asString(rt).utf8(rt);
std::vector<JSVariant> params = count == 2 && args[1].isObject()
? to_variant_vec(rt, args[1])
Expand All @@ -447,6 +481,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["execute"] = HFN(this) {
throw_if_closed("execute");

const std::string query = args[0].asString(rt).utf8(rt);
std::vector<JSVariant> params = count == 2 && args[1].isObject()
? to_variant_vec(rt, args[1])
Expand All @@ -469,6 +505,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["executeWithHostObjects"] = HFN(this) {
throw_if_closed("executeWithHostObjects");

const std::string query = args[0].asString(rt).utf8(rt);
std::vector<JSVariant> params = count == 2 && args[1].isObject()
? to_variant_vec(rt, args[1])
Expand Down Expand Up @@ -501,6 +539,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["executeBatch"] = HFN(this) {
throw_if_closed("executeBatch");

if (count < 1) {
throw std::runtime_error(
"[op-sqlite][executeAsyncBatch] Incorrect parameter count");
Expand Down Expand Up @@ -540,6 +580,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {

#if defined(OP_SQLITE_USE_LIBSQL) || defined(OP_SQLITE_USE_TURSO)
function_map["sync"] = HFN(this) {
throw_if_closed("sync");

#ifdef OP_SQLITE_USE_LIBSQL
opsqlite_libsql_sync(db);
#else
Expand All @@ -551,12 +593,16 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
#ifdef OP_SQLITE_USE_LIBSQL

function_map["setReservedBytes"] = HFN(this) {
throw_if_closed("setReservedBytes");

auto reserved_bytes = static_cast<int32_t>(args[0].asNumber());
opsqlite_libsql_set_reserved_bytes(db, reserved_bytes);
return {};
});

function_map["getReservedBytes"] = HFN(this) {
throw_if_closed("getReservedBytes");

return {opsqlite_libsql_get_reserved_bytes(db)};
});
#endif
Expand All @@ -565,6 +611,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {

#if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO)
function_map["loadFile"] = HFN(this) {
throw_if_closed("loadFile");

if (count < 1) {
throw std::runtime_error(
"[op-sqlite][loadFile] Incorrect parameter count");
Expand All @@ -585,6 +633,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["updateHook"] = HFN(this) {
throw_if_closed("updateHook");

auto callback = std::make_shared<jsi::Value>(rt, args[0]);

if (callback->isUndefined() || callback->isNull()) {
Expand All @@ -598,6 +648,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["commitHook"] = HFN(this) {
throw_if_closed("commitHook");

if (count < 1) {
throw std::runtime_error("[op-sqlite][commitHook] callback needed");
}
Expand All @@ -614,6 +666,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["rollbackHook"] = HFN(this) {
throw_if_closed("rollbackHook");

if (count < 1) {
throw std::runtime_error("[op-sqlite][rollbackHook] callback needed");
}
Expand All @@ -631,6 +685,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["loadExtension"] = HFN(this) {
throw_if_closed("loadExtension");

auto path = args[0].asString(rt).utf8(rt);
std::string entry_point;
if (count > 1 && args[1].isString()) {
Expand All @@ -642,6 +698,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["reactiveExecute"] = HFN(this) {
throw_if_closed("reactiveExecute");

auto query = args[0].asObject(rt);

const std::string query_str =
Expand Down Expand Up @@ -684,13 +742,19 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {

auto_register_update_hook();

auto unsubscribe = HFN2(this, reactiveQuery) {
auto it = std::find(reactive_queries.begin(), reactive_queries.end(),
reactiveQuery);
if (it != reactive_queries.end()) {
reactive_queries.erase(it);
auto weak_self = weak_from_this();

auto unsubscribe = HFN2(weak_self, reactiveQuery) {
auto self = weak_self.lock();
if (self == nullptr) {
return {};
}
auto it = std::find(self->reactive_queries.begin(),
self->reactive_queries.end(), reactiveQuery);
if (it != self->reactive_queries.end()) {
self->reactive_queries.erase(it);
}
auto_register_update_hook();
self->auto_register_update_hook();
return {};
});

Expand All @@ -699,6 +763,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
#endif

function_map["prepareStatement"] = HFN(this) {
throw_if_closed("prepareStatement");

auto query = args[0].asString(rt).utf8(rt);
#ifdef OP_SQLITE_USE_LIBSQL
libsql_stmt_t statement = opsqlite_libsql_prepare_statement(db, query);
Expand Down Expand Up @@ -737,6 +803,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
});

function_map["flushPendingReactiveQueries"] = HFN(this) {
throw_if_closed("flushPendingReactiveQueries");

auto promiseCtr = rt.global().getPropertyAsFunction(rt, "Promise");
auto promise = promiseCtr.callAsConstructor(rt, HFN(this) {
auto resolve = std::make_shared<jsi::Value>(rt, args[0]);
Expand Down Expand Up @@ -802,6 +870,7 @@ void DBHostObject::invalidate() {

// Drain in-flight thread pool work before closing the db handle.
thread_pool->wait_finished();
release_hooks();

#ifdef OP_SQLITE_USE_LIBSQL
opsqlite_libsql_close(db);
Expand Down
7 changes: 6 additions & 1 deletion cpp/DBHostObject.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
#include <sqlite3.h>
#endif
#endif
#include <memory>
#include <unordered_map>
#include <vector>

Expand Down Expand Up @@ -41,7 +42,9 @@ struct ReactiveQuery {
std::shared_ptr<jsi::Value> callback;
};

class JSI_EXPORT DBHostObject : public jsi::HostObject {
class JSI_EXPORT DBHostObject
: public jsi::HostObject,
public std::enable_shared_from_this<DBHostObject> {
public:
// Normal constructor shared between all backends
DBHostObject(jsi::Runtime &rt, std::string &base_path, std::string &db_name,
Expand Down Expand Up @@ -82,6 +85,8 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject {
private:
std::set<std::shared_ptr<ReactiveQuery>> pending_reactive_queries;
void auto_register_update_hook();
void release_hooks();
void throw_if_closed(const char *function_name) const;
void create_jsi_functions(jsi::Runtime &rt);
void flush_pending_reactive_queries(const std::shared_ptr<jsi::Value> &resolve);

Expand Down
29 changes: 29 additions & 0 deletions example/src/tests/dbsetup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,35 @@ describe("DB setup tests", () => {
}
});

it("Throws instead of crashing when the database is used after close", () => {
const db = open({
name: "closedDbGuard.sqlite",
});

db.close();

const callsOnClosedDb = [
() => db.executeSync("SELECT 1;"),
() => db.prepareStatement("SELECT 1;"),
() => db.updateHook(() => {}),
() => db.commitHook(() => {}),
() => db.rollbackHook(() => {}),
() => db.attach({ secondaryDbFileName: "other", alias: "other" }),
];

for (const call of callsOnClosedDb) {
let error: unknown = null;
try {
call();
} catch (e) {
error = e;
}
expect(!!error).toEqual(true);
}

db.delete();
});

it("Should delete db", async () => {
const db = open({
name: "deleteTest",
Expand Down
14 changes: 14 additions & 0 deletions example/src/tests/reactive.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,6 +337,20 @@ describe("Reactive queries", () => {
nickname: "Johnny",
});

unsubscribe();
});

it("Unsubscribing after the database is closed does not crash", async () => {
const unsubscribe = db.reactiveExecute({
query: "SELECT * FROM User;",
arguments: [],
fireOn: [{ table: "User" }],
callback: () => {},
});

db.close();


unsubscribe();
});
});
Loading