Add diagnostics_channel tracing

docs/pages/features/_meta.js

@@ -9,4 +9,5 @@ export default {
   native: 'Native',
   esm: 'ESM',
   callbacks: 'Callbacks',
+  tracing: 'Tracing Channels',
 }

docs/pages/features/tracing.mdx

@@ -0,0 +1,236 @@
+---
+title: Tracing Channels
+---
+
+node-postgres publishes lifecycle events on a set of named tracing channels so
+you can instrument queries, connections, and pool activity without
+monkey-patching. If you're building an APM integration, custom tracer, or just
+want structured logging of your database calls, you can subscribe to these
+channels and node-postgres will tell you when things happen.
+
+Tracing is built on
+[`node:diagnostics_channel`](https://nodejs.org/api/diagnostics_channel.html)
+and is fully supported on Node.js 20+, where nothing is published when no one is
+listening. Node.js 18 has TracingChannel but can't report whether anyone is
+subscribed, so events are always published there, at a small cost per query.
+On Node.js 16 and non-Node runtimes tracing is a no-op.
+
+## Quick start
+
+Here's a minimal example that logs every query with its duration:
+
+```js
+import dc from 'node:diagnostics_channel'
+
+const timings = new WeakMap()
+const channel = dc.tracingChannel('pg:query')
+
+channel.subscribe({
+  start(ctx) {
+    timings.set(ctx, process.hrtime.bigint())
+  },
+  // asyncEnd fires for both successful and failed queries
+  asyncEnd(ctx) {
+    const elapsed = Number(process.hrtime.bigint() - timings.get(ctx)) / 1e6
+    const outcome = ctx.error ? 'failed' : 'completed'
+    console.log(`${ctx.query.text} ${outcome} in ${elapsed.toFixed(2)}ms`)
+    timings.delete(ctx)
+  },
+  error() {},
+  end() {},
+  asyncStart() {},
+})
+```
+
+You don't need to change how you create clients or pools. Just subscribe to the
+channel and node-postgres handles the rest.
+
+## The channels
+
+### TracingChannels
+
+These emit the full lifecycle: `start`, `end`, `asyncStart`, `asyncEnd`, and
+`error`.
+
+| Channel           | Fires for                                                     |
+| ----------------- | ------------------------------------------------------------- |
+| `pg:query`        | Each `client.query()` call                                    |
+| `pg:connect`      | Each `client.connect()` call                                  |
+| `pg:pool:connect` | Each `pool.connect()` call (acquiring a client from the pool) |
+
+### Plain channels
+
+These publish a single message. Subscribe with `dc.channel(name).subscribe(cb)`.
+
+| Channel           | Fires for                             |
+| ----------------- | ------------------------------------- |
+| `pg:pool:release` | A client is released back to the pool |
+| `pg:pool:remove`  | A client is removed from the pool     |
+
+### How they fit together
+
+A typical pooled request moves through the channels in this order:
+
+```text
+pg:pool:connect              acquire a client from the pool
+ └─ pg:connect               connect (only if the pool creates a new client)
+pg:query                     execute the query
+pg:pool:release              release the client back to the pool
+```
+
+`pg:connect` only fires when the pool has to establish a new connection. For
+reused clients it is skipped. `pg:pool:remove` fires separately when a client
+is evicted (e.g. after an error or when `maxUses` is exceeded).
+
+## Lifecycle events
+
+Each tracing channel exposes five sub-channels that fire in a fixed order
+depending on whether the operation completes synchronously or asynchronously:
+
+- **Synchronous success:** `start` -> `end`
+- **Synchronous failure:** `start` -> `error` -> `end`
+- **Asynchronous success:** `start` -> `end` -> `asyncStart` -> `asyncEnd`
+- **Asynchronous failure:** `start` -> `end` -> `error` -> `asyncStart` -> `asyncEnd`
+
+In practice queries and connections are always asynchronous, so you'll mostly
+work with `start` and `asyncEnd`. The context object is shared across all
+events for a single operation, and properties like `result` and `error` are
+added as the operation progresses.
+
+## Context payloads
+
+### pg:query
+
+```js
+{
+  query: {
+    text: 'SELECT $1::int',  // the query text
+    name: undefined,          // prepared statement name, if any
+  },
+  client: {
+    database: 'mydb',
+    host: 'localhost',
+    port: 5432,
+    user: 'postgres',
+    ssl: false,
+  },
+  // added before asyncStart: the query's Result, the same object your callback
+  // receives (an array of Results for multi-statement queries)
+  result: {
+    command: 'SELECT',
+    rowCount: 1,
+    rows: [{ int4: 1 }],
+    fields: [ /* ... */ ],
+  },
+}
+```
+
+### pg:connect
+
+```js
+{
+  connection: {
+    database: 'mydb',
+    host: 'localhost',
+    port: 5432,
+    user: 'postgres',
+    ssl: false,
+  },
+}
+```
+
+### pg:pool:connect
+
+```js
+{
+  pool: {
+    totalCount: 2,      // total clients in the pool
+    idleCount: 1,        // idle clients available
+    waitingCount: 0,     // callers waiting for a client
+    maxSize: 10,         // configured pool maximum
+  },
+  // added on asyncEnd:
+  client: {
+    processID: 123,
+    reused: true,        // whether the client was already in the pool
+  },
+}
+```
+
+### pg:pool:release
+
+```js
+{
+  client: { processID: 123 },
+  error: undefined,           // the Error passed to release(err), if any
+}
+```
+
+### pg:pool:remove
+
+```js
+{
+  client: { processID: 123 },
+}
+```
+
+## More examples
+
+### Monitoring pool usage
+
+```js
+import dc from 'node:diagnostics_channel'
+
+const poolConnect = dc.tracingChannel('pg:pool:connect')
+const poolRelease = dc.channel('pg:pool:release')
+
+poolConnect.subscribe({
+  start(ctx) {
+    console.log('pool checkout:', ctx.pool.idleCount, 'idle,', ctx.pool.waitingCount, 'waiting')
+  },
+  asyncEnd(ctx) {
+    if (ctx.client) console.log('checked out a', ctx.client.reused ? 'reused' : 'new', 'client')
+  },
+  error() {},
+  end() {},
+  asyncStart() {},
+})
+
+poolRelease.subscribe((msg) => {
+  console.log('client released', msg.error ? 'with an error' : '')
+})
+```
+
+### Subscribing to all query events
+
+```js
+import dc from 'node:diagnostics_channel'
+
+const channel = dc.tracingChannel('pg:query')
+
+channel.subscribe({
+  start(ctx) {
+    console.log('query started:', ctx.query.text)
+  },
+  asyncEnd(ctx) {
+    if (ctx.result) console.log('query completed:', ctx.result.command, ctx.result.rowCount, 'rows')
+  },
+  error(ctx) {
+    console.error('query failed:', ctx.error)
+  },
+  end() {},
+  asyncStart() {},
+})
+```
+
+## Notes
+
+- These channels report observability events. They are not hooks for altering
+  behavior; mutating a context payload does not change node-postgres internals.
+- The plain channels (`pg:pool:release`, `pg:pool:remove`) are not
+  TracingChannels because they represent point-in-time events with no async
+  continuation.
+- Queries without a callback or promise, such as `pg-cursor` and
+  `pg-query-stream`, are not traced.
+- The native client (`pg.native`) does not publish to `pg:query` or
+  `pg:connect`. The pool channels still fire when a pool uses native clients.

packages/pg-pool/diagnostics.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const noopChannel = { hasSubscribers: false }
+
+let poolConnectChannel = noopChannel
+let poolReleaseChannel = noopChannel
+let poolRemoveChannel = noopChannel
+
+try {
+  let dc
+  if (typeof process.getBuiltinModule === 'function') {
+    dc = process.getBuiltinModule('diagnostics_channel')
+  } else {
+    dc = require('diagnostics_channel')
+  }
+  if (typeof dc.tracingChannel === 'function') {
+    poolConnectChannel = dc.tracingChannel('pg:pool:connect')
+  }
+  if (typeof dc.channel === 'function') {
+    poolReleaseChannel = dc.channel('pg:pool:release')
+    poolRemoveChannel = dc.channel('pg:pool:remove')
+  }
+} catch (e) {
+  // diagnostics_channel not available (non-Node environment)
+}
+
+// Check explicitly for `false` rather than truthiness because the aggregated
+// `hasSubscribers` getter on TracingChannel is `undefined` on Node 18 (which
+// backported TracingChannel but not the getter). When `undefined`, we assume
+// there may be subscribers and trace unconditionally.
+function shouldTrace(channel) {
+  return channel.hasSubscribers !== false
+}
+
+module.exports = { poolConnectChannel, poolReleaseChannel, poolRemoveChannel, shouldTrace }

packages/pg-pool/index.js

@@ -1,5 +1,6 @@
 'use strict'
 const EventEmitter = require('events').EventEmitter
+const { poolConnectChannel, poolReleaseChannel, poolRemoveChannel, shouldTrace } = require('./diagnostics')
 
 const NOOP = function () {}
 
@@ -178,6 +179,10 @@ class Pool extends EventEmitter {
 
     this._clients = this._clients.filter((c) => c !== client)
     const context = this
+    if (shouldTrace(poolRemoveChannel)) {
+      poolRemoveChannel.publish({ client: { processID: client.processID } })
+    }
+
     client.end(() => {
       context.emit('remove', client)
 
@@ -196,6 +201,31 @@ class Pool extends EventEmitter {
     const response = promisify(this.Promise, cb)
     const result = response.result
 
+    if (shouldTrace(poolConnectChannel)) {
+      const context = {
+        pool: {
+          totalCount: this.totalCount,
+          idleCount: this.idleCount,
+          waitingCount: this.waitingCount,
+          maxSize: this.options.max,
+        },
+      }
+      const origCb = response.callback
+      const enrichedCb = (err, client, done) => {
+        if (client) context.client = { processID: client.processID, reused: !!client._poolUseCount }
+        return origCb(err, client, done)
+      }
+      poolConnectChannel.traceCallback(
+        (tracedCb) => {
+          response.callback = tracedCb
+        },
+        0,
+        context,
+        null,
+        enrichedCb
+      )
+    }
+
     // if we don't have to connect a new client, don't do so
     if (this._isFull() || this._idle.length) {
       // if we have idle clients schedule a pulse immediately
@@ -388,6 +418,10 @@ class Pool extends EventEmitter {
 
     this.emit('release', err, client)
 
+    if (shouldTrace(poolReleaseChannel)) {
+      poolReleaseChannel.publish({ client: { processID: client.processID }, error: err || undefined })
+    }
+
     // TODO(bmc): expose a proper, public interface _queryable and _ending
     if (err || this.ending || !client._queryable || client._ending || client._poolUseCount >= this.options.maxUses) {
       if (client._poolUseCount >= this.options.maxUses) {

packages/pg-pool/package.json

@@ -45,6 +45,7 @@
   },
   "files": [
     "index.js",
+    "diagnostics.js",
     "esm"
   ]
 }