sqlite: add diagnostic channel

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('diagnostics_channel');
+const assert = require('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);
+}

doc/api/diagnostics_channel.md

@@ -1913,10 +1913,33 @@ added: v16.18.0
 
 Emitted when a new thread is created.
 
+#### SQLite
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> 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} The estimated statement run time in nanoseconds.
+
+Emitted when a SQL statement is executed against a [`DatabaseSync`][] instance.
+This allows subscribers to observe every SQL statement executed without
+modifying the database code itself. Tracing is zero-cost 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

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.
@@ -1623,6 +1625,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
 [`PRAGMA foreign_keys`]: https://www.sqlite.org/pragma.html#pragma_foreign_keys
 [`SQLITE_DBCONFIG_DEFENSIVE`]: https://www.sqlite.org/c3ref/c_dbconfig_defensive.html#sqlitedbconfigdefensive
@@ -1634,6 +1637,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

lib/diagnostics_channel.js

@@ -72,11 +72,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;

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

src/env_properties.h

@@ -134,6 +134,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")                                \
@@ -344,6 +345,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")                                                     \

src/node_diagnostics_channel.cc

@@ -127,12 +127,40 @@ void BindingData::Deserialize(Local<Context> 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<Value>& args) {
+  Realm* realm = Realm::GetCurrent(args);
+  BindingData* binding = realm->GetBindingData<BindingData>();
+  if (binding == nullptr) return;
+  CHECK(args[0]->IsUint32());
+  uint32_t index = args[0].As<v8::Uint32>()->Value();
+  auto it = binding->channel_status_callbacks_.find(index);
+  if (it != binding->channel_status_callbacks_.end()) it->second(true);
+}
+
+void BindingData::NotifyChannelInactive(
+    const FunctionCallbackInfo<Value>& args) {
+  Realm* realm = Realm::GetCurrent(args);
+  BindingData* binding = realm->GetBindingData<BindingData>();
+  if (binding == nullptr) return;
+  CHECK(args[0]->IsUint32());
+  uint32_t index = args[0].As<v8::Uint32>()->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<ObjectTemplate> target) {
   Isolate* isolate = isolate_data->isolate();
   SetMethod(
       isolate, target, "getOrCreateChannelIndex", GetOrCreateChannelIndex);
   SetMethod(isolate, target, "linkNativeChannel", LinkNativeChannel);
+  SetMethod(isolate, target, "notifyChannelActive", NotifyChannelActive);
+  SetMethod(isolate, target, "notifyChannelInactive", NotifyChannelInactive);
 }
 
 void BindingData::CreatePerContextProperties(Local<Object> target,
@@ -148,6 +176,8 @@ void BindingData::RegisterExternalReferences(
     ExternalReferenceRegistry* registry) {
   registry->Register(GetOrCreateChannelIndex);
   registry->Register(LinkNativeChannel);
+  registry->Register(NotifyChannelActive);
+  registry->Register(NotifyChannelInactive);
 }
 
 Channel::Channel(Environment* env,

src/node_diagnostics_channel.h

@@ -4,6 +4,7 @@
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
 #include <cinttypes>
+#include <functional>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -53,6 +54,14 @@ class BindingData : public SnapshotableObject {
   static void LinkNativeChannel(
       const v8::FunctionCallbackInfo<v8::Value>& args);
 
+  using ChannelStatusCallback = std::function<void(bool is_active)>;
+  void SetChannelStatusCallback(uint32_t index, ChannelStatusCallback cb);
+
+  static void NotifyChannelActive(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void NotifyChannelInactive(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
+
   static void CreatePerIsolateProperties(IsolateData* isolate_data,
                                          v8::Local<v8::ObjectTemplate> target);
   static void CreatePerContextProperties(v8::Local<v8::Object> target,
@@ -63,6 +72,7 @@ class BindingData : public SnapshotableObject {
 
  private:
   InternalFieldInfo* internal_field_info_ = nullptr;
+  std::unordered_map<uint32_t, ChannelStatusCallback> channel_status_callbacks_;
 };
 
 class Channel : public BaseObject {