From fbaf0baeea6289a40719489810c14c87571401e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Guilherme=20Ara=C3=BAjo?= Date: Thu, 13 Aug 2026 15:01:04 -0300 Subject: [PATCH 1/4] sqlite: add diagnostic channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Guilherme Araújo --- benchmark/sqlite/sqlite-diagnostic-channel.js | 42 ++++ doc/api/diagnostics_channel.md | 31 +++ doc/api/sqlite.md | 6 +- lib/diagnostics_channel.js | 8 + src/base_object_types.h | 3 +- src/env_properties.h | 2 + src/node_diagnostics_channel.cc | 30 +++ src/node_diagnostics_channel.h | 10 + src/node_sqlite.cc | 135 ++++++++++++ src/node_sqlite.h | 35 ++++ .../test-sqlite-diagnostic-channel.js | 194 ++++++++++++++++++ 11 files changed, 494 insertions(+), 2 deletions(-) create mode 100644 benchmark/sqlite/sqlite-diagnostic-channel.js create mode 100644 test/parallel/test-sqlite-diagnostic-channel.js diff --git a/benchmark/sqlite/sqlite-diagnostic-channel.js b/benchmark/sqlite/sqlite-diagnostic-channel.js new file mode 100644 index 000000000000..0610839653df --- /dev/null +++ b/benchmark/sqlite/sqlite-diagnostic-channel.js @@ -0,0 +1,42 @@ +'use strict'; +const common = require('../common.js'); +const sqlite = require('node:sqlite'); +const dc = require('node:diagnostics_channel'); +const assert = require('node:assert'); + +const bench = common.createBenchmark(main, { + n: [1e5], + mode: ['none', 'subscribed', 'unsubscribed'], +}); + +function main(conf) { + const { n, mode } = conf; + + const db = new sqlite.DatabaseSync(':memory:'); + db.exec('CREATE TABLE t (x INTEGER)'); + const insert = db.prepare('INSERT INTO t VALUES (?)'); + + let subscriber; + if (mode === 'subscribed') { + subscriber = () => {}; + dc.subscribe('sqlite.db.query', subscriber); + } else if (mode === 'unsubscribed') { + subscriber = () => {}; + dc.subscribe('sqlite.db.query', subscriber); + dc.unsubscribe('sqlite.db.query', subscriber); + } + // mode === 'none': no subscription ever made + + let result; + bench.start(); + for (let i = 0; i < n; i++) { + result = insert.run(i); + } + bench.end(n); + + if (mode === 'subscribed') { + dc.unsubscribe('sqlite.db.query', subscriber); + } + + assert.ok(result !== undefined); +} diff --git a/doc/api/diagnostics_channel.md b/doc/api/diagnostics_channel.md index 22b70c444a9a..2d85bb5755f0 100644 --- a/doc/api/diagnostics_channel.md +++ b/doc/api/diagnostics_channel.md @@ -1924,10 +1924,41 @@ added: v16.18.0 Emitted when a new thread is created. +#### SQLite + + + +> Stability: 1 - Experimental + +##### Event: `'sqlite.db.query'` + +* `sql` {string} The expanded SQL with bound parameter values substituted. + If expansion fails, the source SQL with unsubstituted placeholders is used + instead. +* `database` {DatabaseSync} The [`DatabaseSync`][] instance that executed the + statement. +* `duration` {number} SQLite's internal estimate of the statement run time in + nanoseconds. This reflects C-layer execution time only and does not include + JavaScript binding overhead such as argument marshaling or result-row + construction. + +Emitted after a SQL statement finishes executing against a [`DatabaseSync`][] +instance. This is a **profiling** event: it fires once per statement upon +completion and reports an estimated duration from SQLite's internal profiler. +It is not a distributed-tracing span. There is no corresponding start event, +no async context propagation, and no parent-span linkage. If you need +OpenTelemetry-compatible spans or async context propagation, wrap your SQLite +calls with a [`TracingChannel`][] at the JavaScript layer instead. + +Publishing is zero-overhead when there are no subscribers. + [BoundedChannel Channels]: #boundedchannel-channels [TracingChannel Channels]: #tracingchannel-channels [`'uncaughtException'`]: process.md#event-uncaughtexception [`BoundedChannel`]: #class-boundedchannel +[`DatabaseSync`]: sqlite.md#class-databasesync [`TracingChannel`]: #class-tracingchannel [`asyncEnd` event]: #asyncendevent [`asyncStart` event]: #asyncstartevent diff --git a/doc/api/sqlite.md b/doc/api/sqlite.md index da7b4e12d8cd..d6ae87d0b28c 100644 --- a/doc/api/sqlite.md +++ b/doc/api/sqlite.md @@ -32,7 +32,9 @@ import sqlite from 'node:sqlite'; const sqlite = require('node:sqlite'); ``` -This module is only available under the `node:` scheme. +This module is only available under the `node:` scheme. SQL trace events can +be observed via the [`diagnostics_channel`][] module. See +[`'sqlite.db.query'`][] for details. The following example shows the basic usage of the `node:sqlite` module to open an in-memory database, write data to the database, and then read the data back. @@ -1890,6 +1892,7 @@ callback function to indicate what type of operation is being authorized. [Run-Time Limits]: https://www.sqlite.org/c3ref/limit.html [SQL injection]: https://en.wikipedia.org/wiki/SQL_injection [Type conversion between JavaScript and SQLite]: #type-conversion-between-javascript-and-sqlite +[`'sqlite.db.query'`]: diagnostics_channel.md#event-sqlitedbquery [`ATTACH DATABASE`]: https://www.sqlite.org/lang_attach.html [`ERR_INVALID_STATE`]: errors.md#err_invalid_state [`PRAGMA foreign_keys`]: https://www.sqlite.org/pragma.html#pragma_foreign_keys @@ -1903,6 +1906,7 @@ callback function to indicate what type of operation is being authorized. [`database.createTagStore()`]: #databasecreatetagstoremaxsize [`database.serialize()`]: #databaseserializedbname [`database.setAuthorizer()`]: #databasesetauthorizercallback +[`diagnostics_channel`]: diagnostics_channel.md [`sqlite3_backup_finish()`]: https://www.sqlite.org/c3ref/backup_finish.html#sqlite3backupfinish [`sqlite3_backup_init()`]: https://www.sqlite.org/c3ref/backup_finish.html#sqlite3backupinit [`sqlite3_backup_step()`]: https://www.sqlite.org/c3ref/backup_finish.html#sqlite3backupstep diff --git a/lib/diagnostics_channel.js b/lib/diagnostics_channel.js index 93a2e85857ab..998f17aa61d6 100644 --- a/lib/diagnostics_channel.js +++ b/lib/diagnostics_channel.js @@ -73,11 +73,19 @@ function markActive(channel) { ObjectSetPrototypeOf(channel, ActiveChannel.prototype); channel._subscribers = []; channel._stores = new SafeMap(); + + // Notify native modules that this channel just got its first subscriber. + if (channel._index !== undefined) + dc_binding.notifyChannelActive(channel._index); } function maybeMarkInactive(channel) { // When there are no more active subscribers or bound, restore to fast prototype. if (!channel._subscribers.length && !channel._stores.size) { + // Notify native modules that this channel just lost its last subscriber. + if (channel._index !== undefined) + dc_binding.notifyChannelInactive(channel._index); + // eslint-disable-next-line no-use-before-define ObjectSetPrototypeOf(channel, Channel.prototype); channel._subscribers = undefined; diff --git a/src/base_object_types.h b/src/base_object_types.h index cd1a06e41a30..1a63e1da1e84 100644 --- a/src/base_object_types.h +++ b/src/base_object_types.h @@ -24,7 +24,8 @@ namespace node { #define UNSERIALIZABLE_BINDING_TYPES(V) \ V(http2_binding_data, http2::BindingData) \ V(http_parser_binding_data, http_parser::BindingData) \ - V(quic_binding_data, quic::BindingData) + V(quic_binding_data, quic::BindingData) \ + V(sqlite_binding_data, sqlite::BindingData) // List of (non-binding) BaseObjects that are serializable in the snapshot. // The first argument should match what the type passes to diff --git a/src/env_properties.h b/src/env_properties.h index 9f69d07e92a3..eb26d3b6cf05 100644 --- a/src/env_properties.h +++ b/src/env_properties.h @@ -143,6 +143,7 @@ V(crypto_rsa_pss_string, "rsa-pss") \ V(cwd_string, "cwd") \ V(data_string, "data") \ + V(database_string, "database") \ V(default_is_true_string, "defaultIsTrue") \ V(defensive_string, "defensive") \ V(deserialize_info_string, "deserializeInfo") \ @@ -359,6 +360,7 @@ V(source_map_url_string, "sourceMapURL") \ V(source_url_string, "sourceURL") \ V(specifier_string, "specifier") \ + V(sql_string, "sql") \ V(stack_string, "stack") \ V(start_string, "start") \ V(state_string, "state") \ diff --git a/src/node_diagnostics_channel.cc b/src/node_diagnostics_channel.cc index ba0492a0df5e..2593f6eab90f 100644 --- a/src/node_diagnostics_channel.cc +++ b/src/node_diagnostics_channel.cc @@ -127,10 +127,38 @@ void BindingData::Deserialize(Local context, CHECK_NOT_NULL(binding); } +void BindingData::SetChannelStatusCallback(uint32_t index, + ChannelStatusCallback cb) { + channel_status_callbacks_[index] = std::move(cb); +} + +void BindingData::NotifyChannelActive(const FunctionCallbackInfo& args) { + Realm* realm = Realm::GetCurrent(args); + BindingData* binding = realm->GetBindingData(); + if (binding == nullptr) return; + CHECK(args[0]->IsUint32()); + uint32_t index = args[0].As()->Value(); + auto it = binding->channel_status_callbacks_.find(index); + if (it != binding->channel_status_callbacks_.end()) it->second(true); +} + +void BindingData::NotifyChannelInactive( + const FunctionCallbackInfo& args) { + Realm* realm = Realm::GetCurrent(args); + BindingData* binding = realm->GetBindingData(); + if (binding == nullptr) return; + CHECK(args[0]->IsUint32()); + uint32_t index = args[0].As()->Value(); + auto it = binding->channel_status_callbacks_.find(index); + if (it != binding->channel_status_callbacks_.end()) it->second(false); +} + void BindingData::CreatePerIsolateProperties(IsolateData* isolate_data, Local target) { Isolate* isolate = isolate_data->isolate(); SetMethod(isolate, target, "linkNativeChannel", LinkNativeChannel); + SetMethod(isolate, target, "notifyChannelActive", NotifyChannelActive); + SetMethod(isolate, target, "notifyChannelInactive", NotifyChannelInactive); } void BindingData::CreatePerContextProperties(Local target, @@ -145,6 +173,8 @@ void BindingData::CreatePerContextProperties(Local target, void BindingData::RegisterExternalReferences( ExternalReferenceRegistry* registry) { registry->Register(LinkNativeChannel); + registry->Register(NotifyChannelActive); + registry->Register(NotifyChannelInactive); } Channel::Channel(Environment* env, diff --git a/src/node_diagnostics_channel.h b/src/node_diagnostics_channel.h index ca68e75a4361..c8c1a79994b2 100644 --- a/src/node_diagnostics_channel.h +++ b/src/node_diagnostics_channel.h @@ -4,6 +4,7 @@ #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS #include +#include #include #include #include @@ -52,6 +53,14 @@ class BindingData : public SnapshotableObject { static void LinkNativeChannel( const v8::FunctionCallbackInfo& args); + using ChannelStatusCallback = std::function; + void SetChannelStatusCallback(uint32_t index, ChannelStatusCallback cb); + + static void NotifyChannelActive( + const v8::FunctionCallbackInfo& args); + static void NotifyChannelInactive( + const v8::FunctionCallbackInfo& args); + static void CreatePerIsolateProperties(IsolateData* isolate_data, v8::Local target); static void CreatePerContextProperties(v8::Local target, @@ -62,6 +71,7 @@ class BindingData : public SnapshotableObject { private: InternalFieldInfo* internal_field_info_ = nullptr; + std::unordered_map channel_status_callbacks_; }; class Channel : public BaseObject { diff --git a/src/node_sqlite.cc b/src/node_sqlite.cc index 1b7c11cb69e7..8057f7ce0aae 100644 --- a/src/node_sqlite.cc +++ b/src/node_sqlite.cc @@ -5,7 +5,9 @@ #include "env-inl.h" #include "memory_tracker-inl.h" #include "node.h" +#include "node_diagnostics_channel.h" #include "node_errors.h" +#include "node_external_reference.h" #include "node_mem-inl.h" #include "node_url.h" #include "simdutf.h" @@ -79,6 +81,30 @@ inline MaybeLocal Utf8StringMaybeOneByte(Isolate* isolate, isolate, input.data(), NewStringType::kNormal, len); } +BindingData::BindingData(Realm* realm, Local wrap) + : BaseObject(realm, wrap) { + MakeWeak(); +} + +void BindingData::MemoryInfo(MemoryTracker* tracker) const { + tracker->TrackFieldWithSize("open_databases", + open_databases.size() * sizeof(DatabaseSync*), + "open_databases"); +} + +void BindingData::CreatePerContextProperties(Local target, + Local unused, + Local context, + void* priv) { + Realm* realm = Realm::GetCurrent(context); + Environment* env = realm->env(); + Realm* principal = env->principal_realm(); + + if (principal->GetBindingData() != nullptr) return; + + principal->AddBindingData(target); +} + #define CHECK_ERROR_OR_THROW(isolate, db, expr, expected, ret) \ do { \ int r_ = (expr); \ @@ -957,6 +983,9 @@ DatabaseSync::DatabaseSync(Environment* env, enable_load_extension_ = allow_load_extension; ignore_next_sqlite_error_ = false; + BindingData* binding = env->principal_realm()->GetBindingData(); + if (binding != nullptr) binding->open_databases.insert(this); + if (open) { Open(); } @@ -979,6 +1008,10 @@ void DatabaseSync::DeleteSessions() { } DatabaseSync::~DatabaseSync() { + BindingData* binding = + env()->principal_realm()->GetBindingData(); + if (binding != nullptr) binding->open_databases.erase(this); + FinalizeBackups(); if (IsOpen()) { @@ -1076,10 +1109,29 @@ bool DatabaseSync::Open() { env()->isolate(), this, load_extension_ret, SQLITE_OK, false); } + trace_channel_ = diagnostics_channel::Channel::Get(env(), "sqlite.db.query"); + if (trace_channel_ != nullptr && trace_channel_->HasSubscribers()) { + sqlite3_trace_v2(connection_, SQLITE_TRACE_PROFILE, TraceCallback, this); + } + opened = true; return true; } +void DatabaseSync::EnableTracing() { + if (!IsOpen()) return; + if (trace_channel_ == nullptr) { + trace_channel_ = + diagnostics_channel::Channel::Get(env(), "sqlite.db.query"); + } + sqlite3_trace_v2(connection_, SQLITE_TRACE_PROFILE, TraceCallback, this); +} + +void DatabaseSync::DisableTracing() { + if (!IsOpen()) return; + sqlite3_trace_v2(connection_, 0, nullptr, nullptr); +} + void DatabaseSync::FinalizeBackups() { for (auto backup : backups_) { backup->Cleanup(); @@ -2730,6 +2782,65 @@ int DatabaseSync::AuthorizerCallback(void* user_data, return int_result; } +int DatabaseSync::TraceCallback(unsigned int type, + void* user_data, + void* p, + void* x) { + if (type != SQLITE_TRACE_PROFILE) { + return 0; + } + + DatabaseSync* db = static_cast(user_data); + Environment* env = db->env(); + + diagnostics_channel::Channel* ch = db->trace_channel_; + if (ch == nullptr || !ch->HasSubscribers()) { + return 0; + } + + Isolate* isolate = env->isolate(); + HandleScope handle_scope(isolate); + + char* expanded = sqlite3_expanded_sql(static_cast(p)); + Local sql_string; + if (expanded != nullptr) { + bool ok = String::NewFromUtf8(isolate, expanded).ToLocal(&sql_string); + sqlite3_free(expanded); + if (!ok) { + return 0; + } + } else { + // Fallback to source SQL if expanded is unavailable + const char* source = sqlite3_sql(static_cast(p)); + if (source == nullptr || + !String::NewFromUtf8(isolate, source).ToLocal(&sql_string)) { + return 0; + } + } + + // x points to the estimated statement run time in nanoseconds. A double is + // sufficient since 2^53 ns (~104 days) exceeds any realistic query duration. + sqlite3_int64 duration_ns = *static_cast(x); + + Local keys[3] = { + env->sql_string().As(), + env->database_string().As(), + env->duration_string().As(), + }; + + Local values[3] = { + sql_string, + db->object(), + Number::New(isolate, static_cast(duration_ns)), + }; + + Local payload = Object::New(isolate, Null(isolate), keys, values, 3); + + ch->Publish(env, payload); + + return 0; +} + StatementSync::StatementSync(Environment* env, Local object, BaseObjectPtr db, @@ -4309,7 +4420,31 @@ static void Initialize(Local target, Local context, void* priv) { Environment* env = Environment::GetCurrent(context); + Realm* realm = env->principal_realm(); Isolate* isolate = env->isolate(); + + // Set up the per-Environment database registry. + BindingData::CreatePerContextProperties(target, unused, context, priv); + + // Register a native callback on the sqlite.db.query diagnostic channel so + // that SQLite tracing is enabled/disabled as subscribers come and go. + auto* diag_binding = + realm->GetBindingData(); + auto* sqlite_bd = realm->GetBindingData(); + if (diag_binding != nullptr && sqlite_bd != nullptr) { + uint32_t idx = diag_binding->GetOrCreateChannelIndex("sqlite.db.query"); + BaseObjectPtr bd_ptr(sqlite_bd); + diag_binding->SetChannelStatusCallback(idx, [bd_ptr](bool is_active) { + BindingData* bd = bd_ptr.get(); + if (bd == nullptr) return; + for (DatabaseSync* db : bd->open_databases) { + if (is_active) + db->EnableTracing(); + else + db->DisableTracing(); + } + }); + } Local db_tmpl = NewFunctionTemplate(isolate, DatabaseSync::New); db_tmpl->InstanceTemplate()->SetInternalFieldCount( diff --git a/src/node_sqlite.h b/src/node_sqlite.h index daa8d5b8a37e..ac710d1013f1 100644 --- a/src/node_sqlite.h +++ b/src/node_sqlite.h @@ -19,6 +19,13 @@ #include namespace node { + +namespace diagnostics_channel { +class Channel; +} // namespace diagnostics_channel + +class ExternalReferenceRegistry; + namespace sqlite { // Mapping from JavaScript property names to SQLite limit constants @@ -195,6 +202,27 @@ class StatementExecutionHelper { bool use_big_ints); }; +class DatabaseSync; + +class BindingData : public BaseObject { + public: + SET_BINDING_ID(sqlite_binding_data) + + BindingData(Realm* realm, v8::Local wrap); + + void MemoryInfo(MemoryTracker* tracker) const override; + SET_MEMORY_INFO_NAME(BindingData) + SET_SELF_SIZE(BindingData) + + std::unordered_set open_databases; + + static void CreatePerContextProperties(v8::Local target, + v8::Local unused, + v8::Local context, + void* priv); + static void RegisterExternalReferences(ExternalReferenceRegistry* registry); +}; + class DatabaseSync : public BaseObject { public: enum InternalFields { @@ -239,6 +267,10 @@ class DatabaseSync : public BaseObject { const char* param2, const char* param3, const char* param4); + static int TraceCallback(unsigned int type, + void* user_data, + void* p, + void* x); void FinalizeStatements(); void RemoveBackup(BackupJob* backup); void AddBackup(BackupJob* backup); @@ -261,6 +293,8 @@ class DatabaseSync : public BaseObject { // enable that use case. void SetIgnoreNextSQLiteError(bool ignore); bool ShouldIgnoreSQLiteError(); + void EnableTracing(); + void DisableTracing(); void IncrementCallbackDepth() { ++callback_depth_; } void DecrementCallbackDepth() { --callback_depth_; } @@ -306,6 +340,7 @@ class DatabaseSync : public BaseObject { std::set backups_; std::unordered_set sessions_; std::unordered_set statements_; + diagnostics_channel::Channel* trace_channel_ = nullptr; friend class DatabaseSyncLimits; friend class Session; diff --git a/test/parallel/test-sqlite-diagnostic-channel.js b/test/parallel/test-sqlite-diagnostic-channel.js new file mode 100644 index 000000000000..2e27425d452e --- /dev/null +++ b/test/parallel/test-sqlite-diagnostic-channel.js @@ -0,0 +1,194 @@ +'use strict'; + +const { skipIfSQLiteMissing } = require('../common'); +skipIfSQLiteMissing(); + +const assert = require('node:assert'); +const dc = require('node:diagnostics_channel'); +const { DatabaseSync } = require('node:sqlite'); +const { suite, it } = require('node:test'); + +suite('sqlite.db.query diagnostics channel', () => { + it('subscriber receives SQL string for exec() statements', (t) => { + const calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + db.exec('INSERT INTO t VALUES (1)'); + + assert.strictEqual(calls.length, 2); + assert.strictEqual(calls[0].sql, 'CREATE TABLE t (x INTEGER)'); + assert.strictEqual(calls[1].sql, 'INSERT INTO t VALUES (1)'); + }); + + it('subscriber receives SQL string for prepared INSERT statements', (t) => { + let calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + calls = []; // reset after setup + + const stmt = db.prepare('INSERT INTO t VALUES (?)'); + stmt.run(42); + + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0].sql, 'INSERT INTO t VALUES (42.0)'); + }); + + it('subscriber receives SQL string for prepared SELECT statements', (t) => { + let calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + db.exec('INSERT INTO t VALUES (1)'); + calls = []; // reset after setup + + const stmt = db.prepare('SELECT x FROM t WHERE x = ?'); + stmt.get(1); + + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0].sql, 'SELECT x FROM t WHERE x = 1.0'); + }); + + it('subscriber receives SQL string for prepared UPDATE statements', (t) => { + let calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + db.exec('INSERT INTO t VALUES (1)'); + calls = []; // reset after setup + + const stmt = db.prepare('UPDATE t SET x = ? WHERE x = ?'); + stmt.run(2, 1); + + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0].sql, 'UPDATE t SET x = 2.0 WHERE x = 1.0'); + }); + + it('subscriber receives SQL string for prepared DELETE statements', (t) => { + let calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + db.exec('INSERT INTO t VALUES (1)'); + calls = []; // reset after setup + + const stmt = db.prepare('DELETE FROM t WHERE x = ?'); + stmt.run(1); + + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0].sql, 'DELETE FROM t WHERE x = 1.0'); + }); + + it('no calls received after unsubscribe', (t) => { + const calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + + db.exec('CREATE TABLE t (x INTEGER)'); + assert.strictEqual(calls.length, 1); + + dc.unsubscribe('sqlite.db.query', handler); + db.exec('INSERT INTO t VALUES (1)'); + assert.strictEqual(calls.length, 1); // No new calls after unsubscribe + }); + + it('falls back to source SQL when expansion fails', (t) => { + let calls = []; + const db = new DatabaseSync(':memory:', { limits: { length: 1000 } }); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x TEXT)'); + calls = []; // reset after setup + + const stmt = db.prepare('INSERT INTO t VALUES (?)'); + + const longValue = 'a'.repeat(977); + stmt.run(longValue); + + assert.strictEqual(calls.length, 1); + // Falls back to source SQL with unexpanded '?' placeholder + assert.strictEqual(calls[0].sql, 'INSERT INTO t VALUES (?)'); + }); + + it('database property identifies the correct database', (t) => { + const calls = []; + const db1 = new DatabaseSync(':memory:'); + const db2 = new DatabaseSync(':memory:'); + t.after(() => { db1.close(); db2.close(); }); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db1.exec('CREATE TABLE t (x INTEGER)'); + db2.exec('CREATE TABLE t (x INTEGER)'); + + assert.strictEqual(calls.length, 2); + assert.strictEqual(calls[0].database, db1); + assert.strictEqual(calls[1].database, db2); + assert.notStrictEqual(calls[0].database, calls[1].database); + }); + + it('duration is a number', (t) => { + const calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + + assert.strictEqual(calls.length, 1); + assert.strictEqual(typeof calls[0].duration, 'number'); + }); + + it('duration is non-negative', (t) => { + const calls = []; + const db = new DatabaseSync(':memory:'); + t.after(() => db.close()); + + const handler = (msg) => calls.push(msg); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + db.exec('CREATE TABLE t (x INTEGER)'); + + assert.strictEqual(calls.length, 1); + assert.ok(calls[0].duration >= 0); + }); +}); From 13af5d18fa2c4a2460736adac6dee181c45d703e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Guilherme=20Ara=C3=BAjo?= Date: Thu, 13 Aug 2026 15:22:59 -0300 Subject: [PATCH 2/4] sqlite: fix diagnostic channel ptr --- src/node_sqlite.cc | 6 +++--- src/node_sqlite.h | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/node_sqlite.cc b/src/node_sqlite.cc index 8057f7ce0aae..2dcdd8f07e4e 100644 --- a/src/node_sqlite.cc +++ b/src/node_sqlite.cc @@ -1110,7 +1110,7 @@ bool DatabaseSync::Open() { } trace_channel_ = diagnostics_channel::Channel::Get(env(), "sqlite.db.query"); - if (trace_channel_ != nullptr && trace_channel_->HasSubscribers()) { + if (trace_channel_ && trace_channel_->HasSubscribers()) { sqlite3_trace_v2(connection_, SQLITE_TRACE_PROFILE, TraceCallback, this); } @@ -1120,7 +1120,7 @@ bool DatabaseSync::Open() { void DatabaseSync::EnableTracing() { if (!IsOpen()) return; - if (trace_channel_ == nullptr) { + if (!trace_channel_) { trace_channel_ = diagnostics_channel::Channel::Get(env(), "sqlite.db.query"); } @@ -2793,7 +2793,7 @@ int DatabaseSync::TraceCallback(unsigned int type, DatabaseSync* db = static_cast(user_data); Environment* env = db->env(); - diagnostics_channel::Channel* ch = db->trace_channel_; + diagnostics_channel::Channel* ch = db->trace_channel_.get(); if (ch == nullptr || !ch->HasSubscribers()) { return 0; } diff --git a/src/node_sqlite.h b/src/node_sqlite.h index ac710d1013f1..d9070ac5d9f1 100644 --- a/src/node_sqlite.h +++ b/src/node_sqlite.h @@ -340,7 +340,7 @@ class DatabaseSync : public BaseObject { std::set backups_; std::unordered_set sessions_; std::unordered_set statements_; - diagnostics_channel::Channel* trace_channel_ = nullptr; + BaseObjectPtr trace_channel_; friend class DatabaseSyncLimits; friend class Session; From 46e551a6505246df2736fbd7698e9b444f49c9f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Guilherme=20Ara=C3=BAjo?= Date: Thu, 13 Aug 2026 15:33:02 -0300 Subject: [PATCH 3/4] sqlite: use erm --- .../test-sqlite-diagnostic-channel.js | 42 +++++++------------ 1 file changed, 16 insertions(+), 26 deletions(-) diff --git a/test/parallel/test-sqlite-diagnostic-channel.js b/test/parallel/test-sqlite-diagnostic-channel.js index 2e27425d452e..3fa8b76b7e49 100644 --- a/test/parallel/test-sqlite-diagnostic-channel.js +++ b/test/parallel/test-sqlite-diagnostic-channel.js @@ -11,8 +11,7 @@ const { suite, it } = require('node:test'); suite('sqlite.db.query diagnostics channel', () => { it('subscriber receives SQL string for exec() statements', (t) => { const calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -28,8 +27,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('subscriber receives SQL string for prepared INSERT statements', (t) => { let calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -38,7 +36,7 @@ suite('sqlite.db.query diagnostics channel', () => { db.exec('CREATE TABLE t (x INTEGER)'); calls = []; // reset after setup - const stmt = db.prepare('INSERT INTO t VALUES (?)'); + using stmt = db.prepare('INSERT INTO t VALUES (?)'); stmt.run(42); assert.strictEqual(calls.length, 1); @@ -47,8 +45,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('subscriber receives SQL string for prepared SELECT statements', (t) => { let calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -58,7 +55,7 @@ suite('sqlite.db.query diagnostics channel', () => { db.exec('INSERT INTO t VALUES (1)'); calls = []; // reset after setup - const stmt = db.prepare('SELECT x FROM t WHERE x = ?'); + using stmt = db.prepare('SELECT x FROM t WHERE x = ?'); stmt.get(1); assert.strictEqual(calls.length, 1); @@ -67,8 +64,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('subscriber receives SQL string for prepared UPDATE statements', (t) => { let calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -78,7 +74,7 @@ suite('sqlite.db.query diagnostics channel', () => { db.exec('INSERT INTO t VALUES (1)'); calls = []; // reset after setup - const stmt = db.prepare('UPDATE t SET x = ? WHERE x = ?'); + using stmt = db.prepare('UPDATE t SET x = ? WHERE x = ?'); stmt.run(2, 1); assert.strictEqual(calls.length, 1); @@ -87,8 +83,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('subscriber receives SQL string for prepared DELETE statements', (t) => { let calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -98,7 +93,7 @@ suite('sqlite.db.query diagnostics channel', () => { db.exec('INSERT INTO t VALUES (1)'); calls = []; // reset after setup - const stmt = db.prepare('DELETE FROM t WHERE x = ?'); + using stmt = db.prepare('DELETE FROM t WHERE x = ?'); stmt.run(1); assert.strictEqual(calls.length, 1); @@ -107,8 +102,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('no calls received after unsubscribe', (t) => { const calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -123,8 +117,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('falls back to source SQL when expansion fails', (t) => { let calls = []; - const db = new DatabaseSync(':memory:', { limits: { length: 1000 } }); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:', { limits: { length: 1000 } }); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -133,7 +126,7 @@ suite('sqlite.db.query diagnostics channel', () => { db.exec('CREATE TABLE t (x TEXT)'); calls = []; // reset after setup - const stmt = db.prepare('INSERT INTO t VALUES (?)'); + using stmt = db.prepare('INSERT INTO t VALUES (?)'); const longValue = 'a'.repeat(977); stmt.run(longValue); @@ -145,9 +138,8 @@ suite('sqlite.db.query diagnostics channel', () => { it('database property identifies the correct database', (t) => { const calls = []; - const db1 = new DatabaseSync(':memory:'); - const db2 = new DatabaseSync(':memory:'); - t.after(() => { db1.close(); db2.close(); }); + using db1 = new DatabaseSync(':memory:'); + using db2 = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -164,8 +156,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('duration is a number', (t) => { const calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); @@ -179,8 +170,7 @@ suite('sqlite.db.query diagnostics channel', () => { it('duration is non-negative', (t) => { const calls = []; - const db = new DatabaseSync(':memory:'); - t.after(() => db.close()); + using db = new DatabaseSync(':memory:'); const handler = (msg) => calls.push(msg); dc.subscribe('sqlite.db.query', handler); From 43fa6d1298720089a7b5e98772a3dd97f0a432f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Guilherme=20Ara=C3=BAjo?= Date: Sat, 15 Aug 2026 09:59:04 -0300 Subject: [PATCH 4/4] sqlite: fix crashes in query diagnostics channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Guilherme Araújo --- doc/api/diagnostics_channel.md | 8 +++ doc/api/sqlite.md | 16 ++++-- src/node_sqlite.cc | 5 +- src/node_sqlite.h | 19 +++++++ .../test-sqlite-diagnostic-channel.js | 55 ++++++++++++++++++- 5 files changed, 97 insertions(+), 6 deletions(-) diff --git a/doc/api/diagnostics_channel.md b/doc/api/diagnostics_channel.md index 2d85bb5755f0..40d3372b5cf9 100644 --- a/doc/api/diagnostics_channel.md +++ b/doc/api/diagnostics_channel.md @@ -1954,6 +1954,12 @@ calls with a [`TracingChannel`][] at the JavaScript layer instead. Publishing is zero-overhead when there are no subscribers. +No event is emitted for a statement that is abandoned mid-iteration and later +finalized, either explicitly through [`statement.close()`][] or when the +statement is garbage collected. Subscribers must not close the database or the +statement, since both are still in use while the event is being delivered; see +[`database.close()`][] and [`statement.close()`][]. + [BoundedChannel Channels]: #boundedchannel-channels [TracingChannel Channels]: #tracingchannel-channels [`'uncaughtException'`]: process.md#event-uncaughtexception @@ -1969,6 +1975,7 @@ Publishing is zero-overhead when there are no subscribers. [`channel.unsubscribe(onMessage)`]: #channelunsubscribeonmessage [`channel.withStoreScope(data)`]: #channelwithstorescopedata [`child_process.spawn()`]: child_process.md#child_processspawncommand-args-options +[`database.close()`]: sqlite.md#databaseclose [`diagnostics_channel.channel(name)`]: #diagnostics_channelchannelname [`diagnostics_channel.subscribe(name, onMessage)`]: #diagnostics_channelsubscribename-onmessage [`diagnostics_channel.tracingChannel()`]: #diagnostics_channeltracingchannelnameorchannels @@ -1978,6 +1985,7 @@ Publishing is zero-overhead when there are no subscribers. [`net.Server.listen()`]: net.md#serverlisten [`process.execve()`]: process.md#processexecvefile-args-env [`start` event]: #startevent +[`statement.close()`]: sqlite.md#statementclose [`worker_threads.locks`]: worker_threads.md#worker_threadslocks [context loss]: async_context.md#troubleshooting-context-loss [thenable object]: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise#thenables diff --git a/doc/api/sqlite.md b/doc/api/sqlite.md index d6ae87d0b28c..4382b18bea01 100644 --- a/doc/api/sqlite.md +++ b/doc/api/sqlite.md @@ -313,8 +313,8 @@ added: v22.5.0 Closes the database connection. An exception is thrown if the database is not open. An [`ERR_INVALID_STATE`][] error is thrown if the method is called while a statement is executing, such as inside a user-defined function, an aggregate -function, or an authorizer callback. This method is a wrapper around -[`sqlite3_close_v2()`][]. +function, an authorizer callback, or a [`'sqlite.db.query'`][] subscriber. This +method is a wrapper around [`sqlite3_close_v2()`][]. ### `database.loadExtension(path[, entryPoint])` @@ -1124,7 +1124,12 @@ added: REPLACEME --> Finalizes the prepared statement. An exception is thrown if the statement is -already finalized. This method is a wrapper around [`sqlite3_finalize()`][]. +already finalized. An [`ERR_INVALID_STATE`][] error is thrown if this statement +is currently executing, which happens when the method is called from a callback +that the statement itself triggered, such as a user-defined function, an +aggregate function, or a [`'sqlite.db.query'`][] subscriber. Other statements +on the same connection can be finalized from such a callback. This method is a +wrapper around [`sqlite3_finalize()`][]. ### `statement.columns()` @@ -1371,7 +1376,9 @@ added: REPLACEME --> Finalizes the prepared statement. If the prepared statement is already -finalized, then this is a no-op. +finalized, then this is a no-op. An [`ERR_INVALID_STATE`][] error is thrown if +this statement is currently executing, under the same conditions as +[`statement.close()`][]. ### `statement.stat(counter)` @@ -1938,6 +1945,7 @@ callback function to indicate what type of operation is being authorized. [`sqlite3session_create()`]: https://www.sqlite.org/session/sqlite3session_create.html [`sqlite3session_delete()`]: https://www.sqlite.org/session/sqlite3session_delete.html [`sqlite3session_patchset()`]: https://www.sqlite.org/session/sqlite3session_patchset.html +[`statement.close()`]: #statementclose [`statement.setAllowBareNamedParameters()`]: #statementsetallowbarenamedparametersenabled [`statement.setAllowUnknownNamedParameters()`]: #statementsetallowunknownnamedparametersenabled [`statement.stat()`]: #statementstatcounter diff --git a/src/node_sqlite.cc b/src/node_sqlite.cc index 2dcdd8f07e4e..8e6a230f625a 100644 --- a/src/node_sqlite.cc +++ b/src/node_sqlite.cc @@ -2794,7 +2794,8 @@ int DatabaseSync::TraceCallback(unsigned int type, Environment* env = db->env(); diagnostics_channel::Channel* ch = db->trace_channel_.get(); - if (ch == nullptr || !ch->HasSubscribers()) { + if (ch == nullptr || !ch->HasSubscribers() || + db->AreTraceEventsSuppressed()) { return 0; } @@ -2836,6 +2837,7 @@ int DatabaseSync::TraceCallback(unsigned int type, Local payload = Object::New(isolate, Null(isolate), keys, values, 3); + CallbackDepthGuard guard(db); ch->Publish(env, payload); return 0; @@ -2868,6 +2870,7 @@ void StatementSync::Close() { } void StatementSync::Finalize() { + TraceEventSuppressionGuard trace_guard(db_.get()); statement_.reset(); InvalidateColumnNameCache(); } diff --git a/src/node_sqlite.h b/src/node_sqlite.h index d9070ac5d9f1..475a759e75e8 100644 --- a/src/node_sqlite.h +++ b/src/node_sqlite.h @@ -320,6 +320,10 @@ class DatabaseSync : public BaseObject { stmt) != stepping_statements_.end(); } + void IncrementTraceSuppressionDepth() { ++trace_suppression_depth_; } + void DecrementTraceSuppressionDepth() { --trace_suppression_depth_; } + bool AreTraceEventsSuppressed() const { return trace_suppression_depth_ > 0; } + SET_MEMORY_INFO_NAME(DatabaseSync) SET_SELF_SIZE(DatabaseSync) @@ -335,6 +339,7 @@ class DatabaseSync : public BaseObject { bool ignore_next_sqlite_error_; int callback_depth_ = 0; int authorizer_depth_ = 0; + int trace_suppression_depth_ = 0; std::vector stepping_statements_; std::set backups_; @@ -518,6 +523,20 @@ class CallbackDepthGuard { DatabaseSync* db_; }; +class TraceEventSuppressionGuard { + public: + explicit TraceEventSuppressionGuard(DatabaseSync* db) : db_(db) { + db_->IncrementTraceSuppressionDepth(); + } + ~TraceEventSuppressionGuard() { db_->DecrementTraceSuppressionDepth(); } + TraceEventSuppressionGuard(const TraceEventSuppressionGuard&) = delete; + TraceEventSuppressionGuard& operator=(const TraceEventSuppressionGuard&) = + delete; + + private: + DatabaseSync* db_; +}; + class SteppingStatementGuard { public: SteppingStatementGuard(DatabaseSync* db, sqlite3_stmt* stmt) : db_(db) { diff --git a/test/parallel/test-sqlite-diagnostic-channel.js b/test/parallel/test-sqlite-diagnostic-channel.js index 3fa8b76b7e49..8b0776c969d7 100644 --- a/test/parallel/test-sqlite-diagnostic-channel.js +++ b/test/parallel/test-sqlite-diagnostic-channel.js @@ -1,12 +1,14 @@ +// Flags: --expose-gc 'use strict'; -const { skipIfSQLiteMissing } = require('../common'); +const { mustCall, skipIfSQLiteMissing } = require('../common'); skipIfSQLiteMissing(); const assert = require('node:assert'); const dc = require('node:diagnostics_channel'); const { DatabaseSync } = require('node:sqlite'); const { suite, it } = require('node:test'); +const { gcUntil } = require('../common/gc'); suite('sqlite.db.query diagnostics channel', () => { it('subscriber receives SQL string for exec() statements', (t) => { @@ -181,4 +183,55 @@ suite('sqlite.db.query diagnostics channel', () => { assert.strictEqual(calls.length, 1); assert.ok(calls[0].duration >= 0); }); + + it('does not publish when an unfinished statement is collected', async (t) => { + let calls = 0; + const handler = () => calls++; + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + let collected = false; + const registry = new FinalizationRegistry(() => { collected = true; }); + + (() => { + const db = new DatabaseSync(':memory:'); + db.exec('CREATE TABLE t (x INTEGER)'); + for (let i = 0; i < 10; i++) { + db.exec(`INSERT INTO t VALUES (${i})`); + } + + const stmt = db.prepare('SELECT x FROM t'); + registry.register(stmt); + stmt.iterate().next(); // Leave the statement unfinished. + })(); + + calls = 0; // reset after setup + await gcUntil('unfinished statement is collected', () => collected); + + assert.strictEqual(calls, 0); + }); + + it('subscriber cannot close the database or statement', (t) => { + using db = new DatabaseSync(':memory:'); + + db.exec('CREATE TABLE t (x INTEGER)'); + using stmt = db.prepare('INSERT INTO t VALUES (?)'); + + const handler = mustCall(() => { + assert.throws(() => db.close(), { code: 'ERR_INVALID_STATE' }); + assert.throws(() => stmt.close(), { code: 'ERR_INVALID_STATE' }); + assert.throws(() => stmt[Symbol.dispose](), { + code: 'ERR_INVALID_STATE', + }); + }); + dc.subscribe('sqlite.db.query', handler); + t.after(() => dc.unsubscribe('sqlite.db.query', handler)); + + stmt.run(1); + + dc.unsubscribe('sqlite.db.query', handler); + assert.deepStrictEqual(db.prepare('SELECT x FROM t').all(), [ + { __proto__: null, x: 1 }, + ]); + }); });