diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 41cd1cd9..6776b864 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -101,6 +101,21 @@ jobs: WORKERD_TLS_PORT: 55433 WORKERD_TLS_CA: ${{ runner.temp }}/server.crt + tracing: + name: Tracing integrations (dd-trace, Sentry) + runs-on: ubuntu-24.04 + timeout-minutes: 15 + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + - uses: ./.github/actions/setup-postgres + with: + version: '18' + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 + with: + node-version: '24' + - run: echo /usr/lib/postgresql/18/bin >> "$GITHUB_PATH" + - run: npm run test:tracing + lint: name: Lint runs-on: ubuntu-24.04 @@ -118,7 +133,7 @@ jobs: publish: name: Stage on npm if: startsWith(github.ref, 'refs/tags/v') - needs: [node, deno, cloudflare, lint] + needs: [node, deno, cloudflare, lint, tracing] runs-on: ubuntu-24.04 timeout-minutes: 45 environment: npm diff --git a/.github/workflows/workerd-ab.yml b/.github/workflows/workerd-ab.yml index f8455a39..55457059 100644 --- a/.github/workflows/workerd-ab.yml +++ b/.github/workflows/workerd-ab.yml @@ -14,6 +14,11 @@ on: ab_base: description: Ref whose cloudflare:sockets build this branch is compared against default: main + ab_subscribe: + description: Channels the head build subscribes to with no-op handlers; the base build never does + type: choice + options: [none, query, all] + default: none permissions: contents: read @@ -24,7 +29,7 @@ concurrency: jobs: ab: - name: Workers CPU A/B against ${{ inputs.ab_base }} on ${{ inputs.ab_target }} + name: Workers CPU A/B against ${{ inputs.ab_base }} on ${{ inputs.ab_target }} with subscribers ${{ inputs.ab_subscribe }} runs-on: ubuntu-24.04 timeout-minutes: 25 env: @@ -44,6 +49,8 @@ jobs: - run: npm run build:cf working-directory: ab-base - run: node tests-startup/workerd-edge.mjs prepare "$RUNNER_TEMP/edge" 2026-10-12 "" ab-base + env: + AB_SUBSCRIBE: ${{ inputs.ab_subscribe }} - name: Deploy the Preview and measure env: NEON_DATABASE_URL: ${{ secrets.NEON_DATABASE_URL }} diff --git a/README.md b/README.md index 50dca684..d9cb665b 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ - **Correctness fixes.** Queries get the right results, transactions stay isolated, and rollbacks are reported as failures. - **Fewer hangs and crashes.** Fixes for stuck queries, shutdown hangs and crashes after disconnects. - **Pool monitoring.** See busy connections, idle connections and waiting queries with `sql.stats()`. +- **Tracing without patches.** Queries, connections and transactions publish [`diagnostics_channel` events](docs/api.md#tracing-channels) for APM and OpenTelemetry subscribers, and nothing is built while no one subscribes. - **Faster queries behind poolers.** Removes an extra network round trip for supported values (opt-in). - **Vercel Fluid Compute support.** Reuse connections between requests and close idle ones before suspension to avoid connection leaks. - **Cloudflare Workers on the cheapest path.** Connects through Cloudflare's own `cloudflare:sockets` API, which costs less CPU per connection than workerd's `node:net`, tested locally and on Cloudflare's network with Neon, Supabase and Hyperdrive. @@ -115,6 +116,7 @@ With `prepare: false`, parameterized queries can pay for an extra network round ### New features - **See pool pressure.** [`sql.stats()`](docs/api.md#pool-stats) reports busy and idle connections and waiting queries. +- **Trace queries natively.** [`postgres:query`, `postgres:connect`, `postgres:pool:connect` and `postgres:transaction`](docs/api.md#tracing-channels) are `diagnostics_channel` tracing channels, published only when something subscribes. - **Handle serverless suspension.** Expired idle connections are replaced on resume; the [Vercel Fluid Compute adapter](docs/api.md#vercel-fluid-compute) keeps instances alive until idle connections close. - **Automatic cleanup.** [`await using`](docs/api.md#teardown--cleanup) closes a pool and `using` releases a reserved connection when its scope ends. Requires Node.js 24+ or TypeScript 5.2+. - **Cloudflare Workers fixes.** On [Workers](docs/api.md#cloudflare-workers-support), `COPY TO` streams and `subscribe()` pause and resume the socket ([#975](https://github.com/porsager/postgres/issues/975)), `ssl: 'prefer'` falls back to plaintext when the server declines TLS, and a certificate from a private CA fails at once with [`TLS_CLOSED`](docs/api.md#tls_closed) instead of a timeout ([#1213](https://github.com/porsager/postgres/issues/1213)). diff --git a/docs/api.md b/docs/api.md index f1a3da73..dfb204eb 100644 --- a/docs/api.md +++ b/docs/api.md @@ -19,6 +19,7 @@ The full API of [@steadybrew/postgresjs](../README.md). Installing, switching fr * [Custom Types](#custom-types) * [Teardown / Cleanup](#teardown--cleanup) * [Reserving connections](#reserving-connections) +* [Tracing channels](#tracing-channels) * [Error handling](#error-handling) * [TypeScript support](#typescript-support) * [Migration tools](#migration-tools) @@ -1509,6 +1510,160 @@ await reserved`select * from users` After release, queries through that reserved handle reject with [`CONNECTION_RELEASED`](#connection_released). Call `sql.reserve()` again to acquire a new handle. A reserved handle whose connection has closed rejects with `CONNECTION_CLOSED` until it is released, and cannot be reused after reconnect. +## Tracing channels + +The driver publishes `diagnostics_channel` events for queries, connections, pool checkouts and transactions, so an APM or OpenTelemetry integration can subscribe without patching the driver. The names and fields follow the tracing channels of node-postgres, so a subscriber written for one driver reads the other with few changes. Nothing is built or published while no one subscribes: a query costs one `hasSubscribers` check. + +### TracingChannels + +These publish `start`, `end`, `asyncStart`, `asyncEnd` and `error`. + +| Channel | Fires for | +|---|---| +| `postgres:query` | Each query: from the call that runs it until the server answers or the query fails | +| `postgres:connect` | Each connection acquisition: socket, TLS, authentication, startup and type fetching, up to the connection serving queries. Retries and other hosts stay inside one span. | +| `postgres:pool:connect` | Each wait for a connection: `sql.reserve()`, `sql.begin()` and every plain pooled query that has a `postgres:query` subscriber. For a plain query it is a child of the query span and ends when the connection starts running the query. Statements sent on a reserved connection or in a transaction do not publish it. | +| `postgres:transaction` | `sql.begin()` and, nested in it, each `sql.savepoint()` | + +### Plain channels + +These publish one message. Subscribe with `dc.channel(name).subscribe(callback)`. + +| Channel | Fires for | +|---|---| +| `postgres:pool:release` | A reserved connection or a transaction gives its connection back | +| `postgres:pool:remove` | A connection closes, including a failed connection attempt | + +### How they fit together + +```text +postgres:pool:connect sql.reserve() or sql.begin() acquires a connection + └─ postgres:connect only if the acquisition had to open the connection +postgres:transaction sql.begin() runs its function + └─ postgres:query each statement, BEGIN and COMMIT included +postgres:pool:release the connection goes back to the pool +postgres:pool:remove later, when the connection closes + +postgres:query a plain pooled query + └─ postgres:pool:connect the wait for a connection + └─ postgres:connect only if the query had to open the connection +``` + +A plain query publishes `postgres:query` and, inside it, `postgres:pool:connect`. The pool span ends when the connection makes the query its active query, so a query that waits behind others on a busy connection, or in the pool's queue, is waiting inside it. A query that settles before then, because the pool is ending, a `COPY` holds every connection, it is cancelled, the connection attempt fails or the connection is lost, fails the pool span first and then itself. + +The `TracingChannel`s publish in the order Node's `tracePromise` does: `start` and `end` in the caller's synchronous code, then for a failure `error`, and for both outcomes `asyncStart` and `asyncEnd`, with `context.result` or `context.error` set first. `start` is published with `runStores`, so a store bound to it, such as an `AsyncLocalStorage` span context, is current in the code that starts the call. Every query publishes `asyncStart` and `asyncEnd` exactly once. That includes cursors read with `for await` or a callback, `forEach`, `COPY` streams (once the copy is over), `describe()`, `simple()`, `unsafe()`, `file()`, pipelined queries, retried queries and queries cancelled before they were sent. The driver's own startup queries (type fetching, `target_session_attrs`) are not published; they belong to the connect span. + +The `result` and `error` keys do not exist on a context until they are set, so a subscriber can tell a finished call from a started one by `'result' in context || 'error' in context`. + +## Context payloads + +The types are `QueryTraceContext`, `ConnectTraceContext`, `PoolConnectTraceContext`, `TransactionTraceContext`, `PoolReleaseMessage` and `PoolRemoveMessage`. No payload contains parameter values. + +### postgres:query + +| Field | Set | +|---|---| +| `query.text` | when the query is written to a connection: SQL with `$n` placeholders | +| `query.name` | when the query is written: the prepared statement name, undefined for an unnamed statement | +| `client.database`, `client.user` | at `start` | +| `client.host`, `client.port` | at `start` when one host is configured, otherwise when the query is written. For a Unix socket the path is the host and there is no port. | +| `client.ssl` | at `start` from the `ssl` option; when the query is written, whether the connection is encrypted | +| `client.processID` | when the query is written: the backend process id, which `postgres:pool:connect` and `postgres:pool:remove` also report | +| `result` | before `asyncStart`: the `Result` with `command` and `count`; for `describe()` the statement; for a `COPY` stream the `Result` of the finished copy, not the stream | +| `error` | before `error` | + +The text is known only after the driver has serialized the query for a connection, so read it at `asyncEnd` or `error`, or rename the span then. The `postgres:pool:connect` span inside a plain query is the wait for a connection, and the rest of the query span is the time on the server. + +### postgres:connect + +| Field | Set | +|---|---| +| `connection.database`, `connection.user` | at `start` | +| `connection.host`, `connection.port` | at `start` when one host is configured; updated as each attempt opens its socket | +| `connection.ssl` | at `start` from the `ssl` option; before `asyncStart`, whether the connection is encrypted | +| `error` | before `error`: the failure that ends the acquisition | + +### postgres:pool:connect + +| Field | Set | +|---|---| +| `pool.totalCount`, `pool.idleCount`, `pool.waitingCount` | at `start`, as `sql.stats()` reports them | +| `pool.maxSize` | at `start`: the `max` option | +| `handoverTime` | before `asyncStart`, once a connection took the call over: the epoch milliseconds at which it did | +| `queuedBehind` | with `handoverTime`: how many queries the connection already had running or waiting when it took the call over | +| `client.processID` | with `handoverTime` | +| `client.reused` | with `handoverTime`: false when this call had to open the connection; a call that waited while another call opened it reports true | +| `error` | before `error`: a connection error, `CONNECTION_ENDED` after `sql.end()`, `COPY_IN_PROGRESS`, or the cancellation or connection error of a plain query that settled before it started | + +`handoverTime` minus the time of `start` is the wait for a connection, which is what OpenTelemetry's `db.client.connection.wait_time` measures. For a plain query with `queuedBehind` above 0, the time from `handoverTime` to `asyncStart` was spent behind those queries on the same connection. A lease span ends a microtask after the hand-over, so the time behind those queries falls inside the lease's first statement. Take the `start` time yourself in the `start` subscriber. On Cloudflare Workers `performance.now()` advances only after I/O, so within a request both times can be equal. + +```js +import dc from 'node:diagnostics_channel' +import { trace } from '@opentelemetry/api' + +const pool = dc.tracingChannel('postgres:pool:connect') + +pool.subscribe({ + start(context) { + context.span = trace.getActiveSpan() + }, + end() {}, + asyncStart() {}, + asyncEnd(context) { + if (context.handoverTime === undefined) + return + + context.span?.addEvent('connection handed over', { 'postgres.queued_behind': context.queuedBehind }, context.handoverTime) + }, + error() {} +}) +``` + +### postgres:transaction + +| Field | Set | +|---|---| +| `client` | as in `postgres:query`; the connection fields are filled in once the transaction has its connection | +| `options` | at `start`: the mode string given to `sql.begin()`; undefined for a savepoint | +| `savepoint` | at `start` for `sql.savepoint()`: the savepoint name as sent, such as `s0_name` | +| `prepared` | when `sql.prepare(name)` is called | +| `result` | before `asyncStart` for `sql.begin()`: `{ committed }`, false when the transaction was prepared | +| `error` | before `error` | + +### postgres:pool:release + +`{ client: { processID }, error }`. `error` is the error the lease ended with: the failure that ended a transaction before its connection was released, or the error that closed the connection. It is undefined after a normal release. A commit the server answered with a rollback is reported after the release, so the release carries no error then. + +### postgres:pool:remove + +`{ client: { processID }, error }`. `error` is undefined for a normal close and is not part of the node-postgres message. + +## Example + +```js +import dc from 'node:diagnostics_channel' + +const channel = dc.tracingChannel('postgres:query') + +channel.subscribe({ + start(context) { + context.startedAt = performance.now() + }, + end() {}, + asyncStart() {}, + asyncEnd(context) { + console.log(context.query.text, context.result?.count, performance.now() - context.startedAt) + }, + error(context) { + console.error(context.query.text, context.error) + } +}) +``` + +To make a span current for the code that runs the query, bind an `AsyncLocalStorage` to the start channel with `channel.start.bindStore(storage, context => span)`. + +On Node.js the channels need Node 22 or newer, as the package does. On a runtime without `diagnostics_channel.tracingChannel` nothing is published. On Cloudflare Workers the module needs the `nodejs_compat` flag, which the driver requires already. + ## Error handling Errors are all thrown to related queries and never globally. Errors coming from database itself are always in the [native Postgres format](https://www.postgresql.org/docs/current/errcodes-appendix.html), and the same goes for any [Node.js errors](https://nodejs.org/api/errors.html#errors_common_system_errors) eg. coming from the underlying connection. diff --git a/docs/changes-since-3.4.9.diff b/docs/changes-since-3.4.9.diff index 3c8f9a75..22453f1c 100644 --- a/docs/changes-since-3.4.9.diff +++ b/docs/changes-since-3.4.9.diff @@ -1,8 +1,8 @@ diff --git a/src/connection.js b/src/connection.js -index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f47b7e2bf6 100644 +index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..2522150c9db4dfaa6553587ee54c3cc5e63f7fd3 100644 --- a/src/connection.js +++ b/src/connection.js -@@ -4,24 +4,41 @@ import crypto from 'crypto' +@@ -4,24 +4,55 @@ import crypto from 'crypto' import Stream from 'stream' import { performance } from 'perf_hooks' @@ -16,8 +16,23 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 +import { SSLRequest, tlsConfig, cancelRequest } from './transport.js' import { Query, CLOSE } from './query.js' import b from './bytes.js' - +- -export default Connection ++import { ++ connectChannel, ++ watched, ++ run, ++ resolved, ++ rejected, ++ connectContext, ++ connectAttempt, ++ connectSecured, ++ queryBuilt, ++ poolHandover, ++ poolStarted, ++ poolRemoved ++} from './tracing.js' ++ +export { Connection } let uid = 1 @@ -48,7 +63,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 const retryRoutines = new Set([ 'FetchPreparedStatement', 'RevalidateCachedQuery', -@@ -49,7 +66,7 @@ const errorFields = { +@@ -49,7 +80,7 @@ const errorFields = { 82 : 'routine' // R } @@ -57,7 +72,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 const { sslnegotiation, ssl, -@@ -72,12 +89,20 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -72,12 +103,20 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose const sent = Queue() , id = uid++ , backend = { pid: null, secret: null } @@ -82,7 +97,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 , errorResponse = null , result = new Result() , incoming = Buffer.alloc(0) -@@ -86,82 +111,442 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -86,102 +125,514 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose , statements = {} , statementId = Math.random().toString(36).slice(2) , statementCount = 1 @@ -90,6 +105,8 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 , remaining = 0 , hostIndex = 0 - , retries = 0 ++ , serverIndex = 0 ++ , secured = false , length = 0 - , delay = 0 , rows = 0 @@ -231,6 +248,8 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + get active() { return query ? 1 : 0 }, + get pipelined() { return sent.length }, + get accepting() { return accepting() }, ++ get session() { return session() }, ++ get inflight() { return inflight() }, + get copying() { return stream !== null }, + get copyHeld() { return stream !== null && !stream.writableEnded }, + get acquiring() { return acquisition && !acquisition.owner.cancelled ? 1 : 0 }, @@ -298,12 +317,36 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + return backoffSeconds(backoff, options.shared.retries) * 1000 + } + ++ function serverAddress() { ++ return options.path || host[serverIndex] ++ } ++ ++ function inflight() { ++ return (query ? 1 : 0) + sent.length ++ } ++ ++ function session() { ++ return { host: serverAddress(), port: serverPort(), ssl: secured, processID: backend.pid } ++ } ++ ++ function serverPort() { ++ return options.path ? undefined : port[serverIndex] ++ } ++ + function standbyPass() { + return target_session_attrs === 'prefer-standby' && host.length > 1 && acquisition.pass === 'standby' + } + + function acquire(owner) { -+ acquisition = { owner, ending: false, pass: 'standby', hostsTried: 0, attempting: null, mismatches: [], lastError: null } ++ acquisition = { owner, ending: false, pass: 'standby', hostsTried: 0, attempting: null, mismatches: [], lastError: null, trace: undefined } ++ if (!watched(connectChannel)) ++ return startAcquisition() ++ ++ acquisition.trace = connectContext(options) ++ run(connectChannel, acquisition.trace, startAcquisition) ++ } ++ ++ function startAcquisition() { + const wait = inheritedBackoff ? inheritedBackoff.at + inheritedBackoff.delay - performance.now() : 0 + inheritedBackoff = null + wait > 0 ? enterBackoff(wait) : enterConnecting() @@ -346,6 +389,9 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + function attach(created) { + listen(created) + socket = created ++ serverIndex = hostIndex ++ secured = false ++ acquisition.trace && connectAttempt(acquisition.trace, serverAddress(), serverPort()) + acquisition.attempting = options.socket ? 'custom socket' : options.path || host[hostIndex] + ':' + port[hostIndex] + + if (options.socket) @@ -384,13 +430,14 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + const config = tlsConfig(options, raw) + raw.removeAllListeners() + socket = tls.connect(config) ++ secured = true + socket.on('secureConnect', () => dispatch('secure')) + listen(socket) + } catch (err) { + afterFailure(err, 'protocol') -+ } -+ } -+ + } + } + + function authenticate() { + try { + statements = {} @@ -495,8 +542,10 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + acquisition = null + endWaiters = [] + transition(Phase.Closed) ++ a && a.trace && rejected(connectChannel, a.trace, err) + a && queryError(a.owner, err) + waiters.forEach(resolve => resolve()) ++ poolRemoved(a ? undefined : backend.pid, reported) + onclose(connection, reported) + } + @@ -519,14 +568,15 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + ms && (closeTimer = clampedTimeout(() => (closeTimer = null, dispatch('closeTimeout')), ms)) + } else { + socket.destroy() - } - } - ++ } ++ } ++ + function handoff() { -+ const { owner, ending } = acquisition ++ const { owner, ending, trace } = acquisition + transition(Phase.Ready) + clearAcquisitionTimers() + acquisition = null ++ trace && (connectSecured(trace, secured), resolved(connectChannel, trace)) + options.shared.retries = 0 + lifeTimer.start() + @@ -560,19 +610,21 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + return true + + q.owner === undefined && (q.owner = connection.owner) ++ q.poolWait && poolHandover(q, inflight(), session()) try { q.state = backend -@@ -169,19 +554,35 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose + query ? sent.push(q) - : (query = q, query.active = true) - +- : (query = q, query.active = true) ++ : (query = q, query.active = true, q.poolWait && poolStarted(q)) ++ + if (q.cancelled) { + q.cancelled = null + abort(q, Errors.generic('57014', 'canceling statement due to user request')) + return accepting() + } -+ + build(q) - return write(toBuffer(q)) - && !q.describeFirst @@ -607,7 +659,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 function toBuffer(q) { if (q.parameters.length >= 65534) throw Errors.generic('MAX_PARAMETERS_EXCEEDED', 'Max number of parameters (65534) exceeded') -@@ -206,7 +607,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -206,7 +657,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose function prepared(q) { return Buffer.concat([ @@ -616,7 +668,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 q.cursorFn ? Execute('', q.cursorRows) : ExecuteUnnamed -@@ -234,13 +635,16 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -234,13 +685,17 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose q.signature = q.prepare && types + string q.onlyDescribe && (delete statements[q.signature]) q.parameters = q.parameters || parameters @@ -632,10 +684,11 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 - typeof options.debug === 'function' && options.debug(id, string, parameters, types) + typeof options.debug === 'function' && callback(options.debug, id, string, parameters, types) ++ q.trace && queryBuilt(q, session()) } function write(x, fn) { -@@ -252,52 +656,12 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -252,52 +707,12 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } function nextWrite(fn) { @@ -689,7 +742,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 function data(x) { if (incomings) { incomings.push(x) -@@ -312,6 +676,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -312,6 +727,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose ? x : Buffer.concat([incoming, x], incoming.length + x.length) @@ -697,7 +750,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 while (incoming.length > 4) { length = incoming.readUInt32BE(1) if (length >= incoming.length) { -@@ -323,74 +688,34 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -323,74 +739,34 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose try { handle(incoming.subarray(0, length + 1)) } catch (e) { @@ -744,12 +797,8 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 - - function reconnect() { - setTimeout(connect, closedTime ? Math.max(0, closedTime + delay - performance.now()) : 0) -+ function lost(err, counted = true) { -+ counted && options.shared.retries++ -+ inheritedBackoff = { at: performance.now(), delay: backoffMs() } -+ enterClosed(err, err.code === 'CONNECTION_CLOSED' ? err : closedBy(err)) - } - +- } +- - function connected() { - try { - statements = {} @@ -764,8 +813,12 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 - } catch (err) { - error(err) - } -- } -- ++ function lost(err, counted = true) { ++ counted && options.shared.retries++ ++ inheritedBackoff = { at: performance.now(), delay: backoffMs() } ++ enterClosed(err, err.code === 'CONNECTION_CLOSED' ? err : closedBy(err)) + } + - function error(err) { - if (connection.queue === queues.connecting && options.host[retries + 1]) - return @@ -786,7 +839,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } function queryError(query, err) { -@@ -401,7 +726,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -401,7 +777,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose err = new Error(err) 'query' in err || 'parameters' in err || Object.defineProperties(err, { @@ -795,7 +848,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 query: { value: query.string, enumerable: options.debug }, parameters: { value: query.parameters, enumerable: options.debug }, args: { value: query.args, enumerable: options.debug }, -@@ -410,51 +735,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -410,51 +786,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose query.reject(err) } @@ -855,7 +908,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } /* Handlers */ -@@ -495,6 +783,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -495,6 +834,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose let column let value @@ -864,7 +917,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 const row = query.isRaw ? new Array(query.statement.columns.length) : {} for (let i = 0; i < query.statement.columns.length; i++) { column = query.statement.columns[i] -@@ -514,13 +804,13 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -514,13 +855,13 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose query.isRaw ? (row[i] = query.isRaw === true ? value @@ -882,7 +935,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } function ParameterStatus(x) { -@@ -528,19 +818,25 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -528,19 +869,25 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose backendParameters[k] = v if (options.parameters[k] !== v) { options.parameters[k] = v @@ -910,7 +963,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 query.resolve(results || result) } } else if (errorResponse) { -@@ -549,42 +845,34 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -549,42 +896,35 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose query = results = errorResponse = null result = new Result() @@ -940,6 +993,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + query = sent.length ? sent.shift() : null + if (query) { + query.active = true ++ query.poolWait && poolStarted(query) + query.cancelled && cancelRequest(options, query.state).then(query.cancelled.resolve, query.cancelled.reject) } @@ -958,9 +1012,6 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 + if (reason) + return mismatched(reason) + } -+ -+ if (needsTypes) -+ return fetchArrayTypes() - connection.reserved - ? !connection.reserved.release && x[5] === 73 // I @@ -971,11 +1022,14 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 - : ending - ? terminate() - : onopen(connection) ++ if (needsTypes) ++ return fetchArrayTypes() ++ + handoff() } function CommandComplete(x) { -@@ -602,7 +890,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -602,7 +942,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose final && (final(), final = null) @@ -984,7 +1038,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 return errored(Errors.generic('UNSAFE_TRANSACTION', 'Only use sql.begin, sql.reserved or max: 1')) if (query.options.simple) -@@ -654,7 +942,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -654,7 +994,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose const number = x.readUInt16BE(index + 4) const type = x.readUInt32BE(index + 6) query.statement.columns[i] = { @@ -993,7 +1047,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 ? transform.column.from(x.toString('utf8', start, index - 1)) : x.toString('utf8', start, index - 1), parser: parsers[type], -@@ -670,8 +958,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -670,8 +1010,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose return (query.resolve(query.statement), write(Sync)) } @@ -1005,7 +1059,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 type === 3 ? AuthenticationCleartextPassword : type === 5 ? AuthenticationMD5Password : type === 10 ? SASL : -@@ -679,18 +968,20 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -679,18 +1020,20 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose type === 12 ? SASLFinal : type !== 0 ? UnknownAuth : noop @@ -1029,7 +1083,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 const payload = 'md5' + ( await md5( Buffer.concat([ -@@ -699,40 +990,42 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -699,40 +1042,42 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose ]) ) ) @@ -1063,10 +1117,10 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 - parseInt(res.i), 32, - 'sha256' - ) +- +- const clientKey = await hmac(saltedPassword, 'Client Key') + const { clientKey, storedKey, serverKey } = await scramKeys(await Pass(), res.s, res.i) -- const clientKey = await hmac(saltedPassword, 'Client Key') -- - const auth = 'n=*,r=' + nonce + ',' + const auth = 'n=*,r=' + clientNonce + ',' + 'r=' + res.r + ',s=' + res.s + ',i=' + res.i @@ -1086,7 +1140,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 write( b().p().str(payload).end() ) -@@ -742,8 +1035,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -742,8 +1087,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose if (x.toString('utf8', 9).split(b.N, 1)[0].slice(2) === serverSignature) return /* c8 ignore next 5 */ @@ -1096,7 +1150,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } function Pass() { -@@ -765,48 +1057,47 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -765,48 +1109,47 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose backend.secret = x.readUInt32BE(9) } @@ -1113,8 +1167,9 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 order by b.oid - `], [], execute) - types.forEach(({ oid, typarray }) => addArrayType(oid, typarray)) -- } -- ++ `, types => types.forEach(({ oid, typarray }) => addArrayType(options, oid, typarray))) + } + - function addArrayType(oid, typarray) { - if (!!options.parsers[typarray] && !!options.serializers[typarray]) return - const parser = options.parsers[oid] @@ -1122,9 +1177,8 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 - options.parsers[typarray] = (xs) => arrayParser(xs, parser, typarray) - options.parsers[typarray].array = true - options.serializers[typarray] = (xs) => arraySerializer(xs, options.serializers[oid], options, typarray) -+ `, types => types.forEach(({ oid, typarray }) => addArrayType(options, oid, typarray))) - } - +- } +- - function tryNext(x, xs) { + function mismatchReason(x, xs) { return ( @@ -1168,7 +1222,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } function ErrorResponse(x) { -@@ -814,12 +1105,11 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -814,12 +1157,11 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose (query.cursorFn || query.describeFirst) && write(Sync) errorResponse = Errors.postgres(parseError(x)) } else { @@ -1182,7 +1236,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 q.retried = error execute(q) } -@@ -830,22 +1120,27 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -830,22 +1172,27 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose let index = 9 while (x[index++] !== 0); @@ -1216,7 +1270,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } } -@@ -862,7 +1157,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -862,7 +1209,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose }, destroy(error, callback) { callback(error) @@ -1225,7 +1279,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 stream = null }, final(callback) { -@@ -872,27 +1167,29 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -872,27 +1219,29 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } }) query.resolve(stream) @@ -1258,7 +1312,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 stream = null }, final(callback) { -@@ -901,22 +1198,36 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -901,22 +1250,36 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } }) query.resolve(stream) @@ -1298,7 +1352,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } /* c8 ignore next 3 */ -@@ -945,20 +1256,21 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -945,20 +1308,21 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } /* Messages */ @@ -1327,7 +1381,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 prev = b.i b.inc(4).str(x).i32(b.i - prev - 4, prev) -@@ -966,7 +1278,27 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -966,7 +1330,27 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose b.i16(0) @@ -1356,7 +1410,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 } function Parse(str, parameters, types, name = '') { -@@ -994,7 +1326,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose +@@ -994,7 +1378,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } function StartupMessage() { @@ -1365,7 +1419,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 Object.entries(Object.assign({ user, database, -@@ -1019,6 +1351,33 @@ function parseError(x) { +@@ -1019,6 +1403,33 @@ function parseError(x) { return error } @@ -1399,7 +1453,7 @@ index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..4e05655b08eb1c8981ba627f49d650f4 function md5(x) { return crypto.createHash('md5').update(x).digest('hex') } -@@ -1042,21 +1401,44 @@ function xor(a, b) { +@@ -1042,21 +1453,44 @@ function xor(a, b) { function timer(fn, seconds) { seconds = typeof seconds === 'function' ? seconds() : seconds if (!seconds) @@ -1485,7 +1539,7 @@ index 0ff83c428bec774e11417b8e1f033dc3305c18cf..023c3ba1bd113e6f5508ab079efcf220 code: x, errno: x, diff --git a/src/index.js b/src/index.js -index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd02329497fd7 100644 +index c7fba3dace20173f4cb13f3f138bb6d213b43683..462dfe482408620ba51d557cb5cd1d8b98949438 100644 --- a/src/index.js +++ b/src/index.js @@ -3,10 +3,12 @@ import fs from 'fs' @@ -1501,7 +1555,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 toPascal, pascal, toCamel, -@@ -18,13 +20,17 @@ import { +@@ -18,12 +20,29 @@ import { fromKebab } from './types.js' @@ -1515,13 +1569,25 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 +import { Errors, PostgresError, callback } from './errors.js' import Subscribe from './subscribe.js' import largeObject from './large.js' - -+const Lease = { Active: 0, Settling: 1, Released: 2, Ended: 3, Closed: 4 } ++import { ++ transactionChannel, ++ poolConnectChannel, ++ watched, ++ traceRoute, ++ traceAsync, ++ transactionContext, ++ transactionSession, ++ transactionCommitted, ++ poolContext, ++ poolAcquired, ++ poolReleased ++} from './tracing.js' + ++const Lease = { Active: 0, Settling: 1, Released: 2, Ended: 3, Closed: 4 } + Object.assign(Postgres, { PostgresError, - toPascal, -@@ -51,6 +57,7 @@ function Postgres(a, b) { +@@ -51,6 +70,7 @@ function Postgres(a, b) { , subscribe = options.no_subscribe || Subscribe(Postgres, { ...options }) let ending = false @@ -1529,7 +1595,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 const queries = Queue() , connecting = Queue() -@@ -62,7 +69,7 @@ function Postgres(a, b) { +@@ -62,7 +82,7 @@ function Postgres(a, b) { , full = Queue() , queues = { connecting, reserved, closed, ended, open, busy, full } @@ -1538,7 +1604,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 const sql = Sql(handler) -@@ -75,10 +82,12 @@ function Postgres(a, b) { +@@ -75,16 +95,19 @@ function Postgres(a, b) { PostgresError, options, reserve, @@ -1552,7 +1618,34 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 }) return sql -@@ -144,6 +153,56 @@ function Postgres(a, b) { + + function Sql(handler) { + handler.debug = options.debug ++ handler.config = options + + Object.entries(options.types).reduce((acc, [name, type]) => { + acc[name] = (x) => new Parameter(x, type.to) +@@ -128,7 +151,7 @@ function Postgres(a, b) { + + function file(path, args = [], options = {}) { + arguments.length === 2 && !Array.isArray(args) && (options = args, args = []) +- const query = new Query([], args, (query) => { ++ const read = query => { + fs.readFile(path, 'utf8', (err, string) => { + if (err) + return query.reject(err) +@@ -136,7 +159,9 @@ function Postgres(a, b) { + query.strings = [string] + handler(query) + }) +- }, cancel, { ++ } ++ read.config = handler.config ++ const query = new Query([], args, read, cancel, { + ...options, + simple: 'simple' in options ? options.simple : args.length === 0 + }) +@@ -144,6 +169,56 @@ function Postgres(a, b) { } } @@ -1609,7 +1702,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 async function listen(name, fn, onlisten) { const listener = { fn, onlisten } -@@ -153,6 +212,7 @@ function Postgres(a, b) { +@@ -153,6 +228,7 @@ function Postgres(a, b) { idle_timeout: null, max_lifetime: null, fetch_types: false, @@ -1617,7 +1710,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 onclose() { Object.entries(listen.channels).forEach(([name, { listeners }]) => { delete listen.channels[name] -@@ -160,7 +220,7 @@ function Postgres(a, b) { +@@ -160,7 +236,7 @@ function Postgres(a, b) { }) }, onnotify(c, x) { @@ -1626,7 +1719,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } })) -@@ -170,7 +230,7 @@ function Postgres(a, b) { +@@ -170,7 +246,7 @@ function Postgres(a, b) { if (exists) { channels[name].listeners.push(listener) const result = await channels[name].result @@ -1635,7 +1728,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 return { state: result.state, unlisten } } -@@ -178,7 +238,7 @@ function Postgres(a, b) { +@@ -178,7 +254,7 @@ function Postgres(a, b) { sql.unsafe('"' + name.replace(/"/g, '""') + '"') }`, listeners: [listener] } const result = await channels[name].result @@ -1644,7 +1737,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 return { state: result.state, unlisten } async function unlisten() { -@@ -200,62 +260,146 @@ function Postgres(a, b) { +@@ -200,62 +276,171 @@ function Postgres(a, b) { return await sql`select pg_notify(${ channel }, ${ '' + payload })` } @@ -1657,29 +1750,40 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 - queries.push(query) - closed.length && connect(closed.shift(), query) - }) -+ async function lease(auto) { ++ function lease(auto) { ++ if (!watched(poolConnectChannel)) ++ return acquire(auto) ++ ++ const trace = poolContext(stats()) ++ return traceAsync(poolConnectChannel, trace, () => acquire(auto, trace)) ++ } ++ ++ async function acquire(auto, trace) { + if (ending) + throw Errors.connection('CONNECTION_ENDED', options) + -+ const l = { state: null, connection: null, error: null, onclose: null, queue: Queue() } ++ const l = { state: null, connection: null, error: null, outcome: undefined, behind: 0, onclose: null, queue: Queue() } ++ let waiter = null + const idle = takeOpen() + idle + ? own(idle, l) + : auto && !closed.length && busy.length + ? own(busy.shift(), l) + : await new Promise((resolve, reject) => { -+ const query = { reserve: c => resolve(own(c, l)), reject } ++ const query = waiter = { reserve: c => resolve(own(c, l)), reject, fresh: false } + closed.length + ? connect(closed.shift(), query) + : allHeldByCopy() + ? reject(copyInProgress()) + : queries.push(query) + }) ++ trace && poolAcquired(trace, l.connection.session.processID, !(waiter && waiter.fresh), l.behind) + return l + } + + function own(c, l) { + l.connection = c ++ l.behind = c.inflight + l.state = Lease.Active + c.owner = l + l.next = () => { @@ -1708,20 +1812,22 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 + ? Errors.generic('CONNECTION_RELEASED', 'Reserved connection used after release()') + : Errors.generic('TRANSACTION_ENDED', 'Transaction handle used after its transaction ended') + } ++ ++ function stop(l, state) { ++ if (l.state === Lease.Released || l.state === Lease.Ended) ++ return null - move(c, reserved) - c.reserved = () => queue.length - ? c.execute(queue.shift()) - : move(c, reserved) - c.reserved.release = true -+ function stop(l, state) { -+ if (l.state === Lease.Released || l.state === Lease.Ended) -+ return null - ++ const lost = l.state === Lease.Closed + l.state = state + while (l.queue.length) + l.queue.shift().reject(leaseError(l)) -+ + ++ poolReleased(l.connection, lost ? l.error : l.outcome) + const c = l.connection + if (c.owner !== l) + return null @@ -1754,10 +1860,21 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } } - async function begin(options, fn) { - !fn && (fn = options, options = '') +- async function begin(options, fn) { +- !fn && (fn = options, options = '') - const queries = Queue() ++ function begin(mode, fn) { ++ !fn && (fn = mode, mode = '') ++ if (!watched(transactionChannel)) ++ return transact(mode, fn) ++ ++ const trace = transactionContext(options, undefined, mode) ++ return traceAsync(transactionChannel, trace, () => transact(mode, fn, trace)) ++ } ++ ++ async function transact(options, fn, trace) { + const l = await lease(true) ++ trace && transactionSession(trace, l.connection.session) let savepoints = 0 - , connection , prepare = null @@ -1776,27 +1893,29 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 + + async function transaction() { + await Sql(handler).unsafe('begin ' + options.replace(/[^a-z ]/ig, ''), []).execute().catch(e => { ++ l.outcome = e + e instanceof PostgresError && release() + throw e + }) + return scope(fn) - } - -- async function scope(c, fn, name) { -- const sql = Sql(handler) ++ } ++ + async function finish(statement) { + let result + l.state = Lease.Settling + try { + result = await statement(Sql(q => handler(q, Lease.Settling))) + } catch (e) { ++ l.outcome = l.outcome || e + terminate() + throw e + } + release() + return result -+ } -+ + } + +- async function scope(c, fn, name) { +- const sql = Sql(handler) + function release() { + const c = stop(l, Lease.Ended) + c && c.release() @@ -1816,6 +1935,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 + throw Errors.generic('INVALID_TRANSACTION_NAME', 'sql.prepare requires a non-empty string without null bytes') + + prepare = 'E\'' + x.replace(/\\/g, '\\\\').replace(/'/g, '\'\'') + '\'' ++ trace && (trace.prepared = x) + } let uncaughtError , result @@ -1825,13 +1945,14 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 try { result = await new Promise((resolve, reject) => { const x = fn(sql) -@@ -265,17 +409,27 @@ function Postgres(a, b) { +@@ -265,17 +450,30 @@ function Postgres(a, b) { if (uncaughtError) throw uncaughtError } catch (e) { - await (name - ? sql`rollback to ${ sql(name) }` - : sql`rollback` ++ name || (l.outcome = e) + l.state === Lease.Active && await (name + ? sql.unsafe('rollback to ' + escapeIdentifier(name)) + : finish(final => final.unsafe('rollback')) @@ -1856,15 +1977,20 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 + uncaughtError && { cause: uncaughtError } + ) + } ++ ++ trace && transactionCommitted(trace, prepare === null) } return result -@@ -285,23 +439,18 @@ function Postgres(a, b) { +@@ -285,23 +483,21 @@ function Postgres(a, b) { return savepoint(sql => sql.apply(sql, arguments)) arguments.length === 1 && (fn = name, name = null) - return scope(c, fn, 's' + savepoints++ + (name ? '_' + name : '')) -+ return scope(fn, 's' + savepoints++ + (name ? '_' + name : '')) ++ const id = 's' + savepoints++ + (name ? '_' + name : '') ++ return watched(transactionChannel) ++ ? traceAsync(transactionChannel, transactionContext(options, l.connection.session, undefined, id), () => scope(fn, id)) ++ : scope(fn, id) } - function handler(q) { @@ -1889,7 +2015,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } } -@@ -310,7 +459,7 @@ function Postgres(a, b) { +@@ -310,7 +506,7 @@ function Postgres(a, b) { queue.push(c) c.queue = queue queue === open @@ -1898,28 +2024,40 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 : c.idleTimer.cancel() return c } -@@ -330,15 +479,51 @@ function Postgres(a, b) { +@@ -327,18 +523,64 @@ function Postgres(a, b) { + } + + function handler(query) { ++ const refusal = query.trace && watched(poolConnectChannel) ++ ? traceRoute(query, poolContext(stats()), route) ++ : route(query) ++ refusal && query.reject(refusal) ++ } ++ ++ function route(query) { if (ending) - return query.reject(Errors.connection('CONNECTION_ENDED', options, options)) +- return query.reject(Errors.connection('CONNECTION_ENDED', options, options)) ++ return Errors.connection('CONNECTION_ENDED', options, options) - if (open.length) - return go(open.shift(), query) + const idle = takeOpen() + if (idle) -+ return go(idle, query) ++ return (go(idle, query), null) if (closed.length) - return connect(closed.shift(), query) - -- busy.length -- ? go(busy.shift(), query) -- : queries.push(query) +- return connect(closed.shift(), query) ++ return (connect(closed.shift(), query), null) ++ + const c = takeBusy() -+ c -+ ? go(c, query) -+ : allHeldByCopy() -+ ? query.reject(copyInProgress()) -+ : queries.push(query) ++ if (c) ++ return (go(c, query), null) ++ ++ if (allHeldByCopy()) ++ return copyInProgress() ++ ++ queries.push(query) ++ return null + } + + function allHeldByCopy() { @@ -1935,7 +2073,10 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 + const c = busy.shift() + if (c.accepting) + return c -+ + +- busy.length +- ? go(busy.shift(), query) +- : queries.push(query) + move(c, full) + } + @@ -1955,7 +2096,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } function go(c, query) { -@@ -348,17 +533,18 @@ function Postgres(a, b) { +@@ -348,17 +590,18 @@ function Postgres(a, b) { } function cancel(query) { @@ -1984,7 +2125,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 }) } -@@ -368,13 +554,23 @@ function Postgres(a, b) { +@@ -368,13 +611,23 @@ function Postgres(a, b) { await 1 let timer @@ -2015,7 +2156,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } async function close() { -@@ -382,9 +578,9 @@ function Postgres(a, b) { +@@ -382,13 +635,14 @@ function Postgres(a, b) { } async function destroy(resolve) { @@ -2026,7 +2167,12 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 resolve() } -@@ -398,12 +594,18 @@ function Postgres(a, b) { + function connect(c, query) { ++ query.reserve ? query.fresh = true : query.poolWait && (query.poolWait.fresh = true) + move(c, connecting) + c.connect(query) + return c +@@ -398,12 +652,18 @@ function Postgres(a, b) { move(c, ended) } @@ -2048,7 +2194,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 while (ready && queries.length && max-- > 0) { const query = queries.shift() -@@ -418,12 +620,60 @@ function Postgres(a, b) { +@@ -418,12 +678,60 @@ function Postgres(a, b) { : move(c, full) } @@ -2113,7 +2259,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } } -@@ -453,16 +703,17 @@ function parseOptions(a, b) { +@@ -453,16 +761,17 @@ function parseOptions(a, b) { connect_timeout : 30, max_lifetime : max_lifetime, max_pipeline : 100, @@ -2133,7 +2279,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 host : Array.isArray(host) ? host : host.split(',').map(x => x.split(':')[0]), port : Array.isArray(port) ? port : host.split(',').map(x => parseInt(x.split(':')[1] || port)), path : o.path || host.indexOf('/') > -1 && host + '/.s.PGSQL.' + port, -@@ -472,8 +723,8 @@ function parseOptions(a, b) { +@@ -472,8 +781,8 @@ function parseOptions(a, b) { ...Object.entries(defaults).reduce( (acc, [k, d]) => { const value = k in o ? o[k] : k in query @@ -2144,7 +2290,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 acc[k] = typeof value === 'string' && ints.includes(k) ? +value : value -@@ -488,16 +739,26 @@ function parseOptions(a, b) { +@@ -488,16 +797,26 @@ function parseOptions(a, b) { }, types : o.types || {}, target_session_attrs: tsa(o, url, env), @@ -2172,7 +2318,7 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 } function tsa(o, url, env) { -@@ -508,8 +769,12 @@ function tsa(o, url, env) { +@@ -508,8 +827,12 @@ function tsa(o, url, env) { throw new Error('target_session_attrs ' + x + ' is not supported') } @@ -2188,25 +2334,35 @@ index c7fba3dace20173f4cb13f3f138bb6d213b43683..db93ccaa423e3162f42f99c9e52cd023 function max_lifetime() { diff --git a/src/query.js b/src/query.js -index 0d44a15cd9e42beaf11d4be6ccde60223e09508a..c1777a904c2f214a041fb8af0b56f0ac38418081 100644 +index 0d44a15cd9e42beaf11d4be6ccde60223e09508a..575c601452ab89fe411fb848a2391c07d6e13fbd 100644 --- a/src/query.js +++ b/src/query.js -@@ -23,10 +23,12 @@ export class Query extends Promise { +@@ -1,3 +1,5 @@ ++import { queryChannel, watched, traceQuery, querySettled, queryFailed } from './tracing.js' ++ + const originCache = new Map() + , originStackCache = new Map() + , originError = Symbol('OriginError') +@@ -23,12 +25,16 @@ export class Query extends Promise { this.state = null this.statement = null - this.resolve = x => (this.active = false, resolve(x)) - this.reject = x => (this.active = false, reject(x)) + // dd-trace's load-time rewrite only accepts single-expression arrows here -+ this.resolve = x => (this.active = false, this.settled = true, resolve(x)) -+ this.reject = x => (this.active = false, this.settled = true, reject(x)) ++ this.resolve = x => (this.trace && querySettled(this, x), this.active = false, this.settled = true, resolve(x)) ++ this.reject = x => (this.trace && queryFailed(this, x), this.active = false, this.settled = true, reject(x)) this.active = false + this.settled = false this.cancelled = null this.executed = false ++ this.trace = undefined ++ this.poolWait = undefined this.signature = '' -@@ -50,7 +52,14 @@ export class Query extends Promise { + + this[originError] = this.handler.debug +@@ -50,7 +56,14 @@ export class Query extends Promise { } cancel() { @@ -2222,6 +2378,46 @@ index 0d44a15cd9e42beaf11d4be6ccde60223e09508a..c1777a904c2f214a041fb8af0b56f0ac } simple() { +@@ -96,8 +109,8 @@ export class Query extends Promise { + resolve({ value, done: false }) + return new Promise(r => prev = r) + } +- this.resolve = () => (this.active = false, resolve({ done: true })) +- this.reject = x => (this.active = false, reject(x)) ++ this.resolve = x => (this.trace && querySettled(this, x), this.active = false, resolve({ done: true })) ++ this.reject = x => (this.trace && queryFailed(this, x), this.active = false, reject(x)) + }) + this.execute() + return promise +@@ -137,7 +150,15 @@ export class Query extends Promise { + } + + async handle() { +- !this.executed && (this.executed = true) && await 1 && this.handler(this) ++ if (this.executed) ++ return ++ ++ this.executed = true ++ if (!this.initialization && !this.settled && watched(queryChannel)) ++ return traceQuery(this, dispatch) ++ ++ await 1 ++ this.handler(this) + } + + execute() { +@@ -161,6 +182,11 @@ export class Query extends Promise { + } + } + ++async function dispatch(q) { ++ await 1 ++ q.handler(q) ++} ++ + function cachedError(xs) { + if (originCache.has(xs)) + return originCache.get(xs) diff --git a/src/subscribe.js b/src/subscribe.js index 4f8934cc8277fc58d2c37838cfb1afc315bc0744..0fd5d4201123e4c08d85fd1c46396baa60ba53cf 100644 --- a/src/subscribe.js @@ -2536,6 +2732,198 @@ index 0000000000000000000000000000000000000000..bcdf4c1894e0f84957d7ff3890bea9d8 +export default function clampedTimeout(fn, ms) { + return setTimeout(fn, Math.min(ms, maxDelay)) +} +diff --git a/src/tracing.js b/src/tracing.js +new file mode 100644 +index 0000000000000000000000000000000000000000..42cefb340333894c94f101919e479387a169ebe0 +--- /dev/null ++++ b/src/tracing.js +@@ -0,0 +1,186 @@ ++import dc from 'diagnostics_channel' ++import { performance } from 'perf_hooks' ++ ++const supported = Boolean(dc) && typeof dc.tracingChannel === 'function' ++ ++export const queryChannel = supported ? dc.tracingChannel('postgres:query') : undefined ++export const connectChannel = supported ? dc.tracingChannel('postgres:connect') : undefined ++export const transactionChannel = supported ? dc.tracingChannel('postgres:transaction') : undefined ++export const poolConnectChannel = supported ? dc.tracingChannel('postgres:pool:connect') : undefined ++export const poolReleaseChannel = supported ? dc.channel('postgres:pool:release') : undefined ++export const poolRemoveChannel = supported ? dc.channel('postgres:pool:remove') : undefined ++ ++export function watched(channel) { ++ return channel !== undefined && (channel.hasSubscribers ?? channel.start.hasSubscribers) ++} ++ ++export function run(channel, context, fn, ...args) { ++ try { ++ return channel.start.runStores(context, fn, undefined, ...args) ++ } finally { ++ channel.end.publish(context) ++ } ++} ++ ++export function resolved(channel, context) { ++ channel.asyncStart.publish(context) ++ channel.asyncEnd.publish(context) ++} ++ ++export function rejected(channel, context, error) { ++ context.error = error ++ channel.error.publish(context) ++ resolved(channel, context) ++} ++ ++export function traceAsync(channel, context, fn) { ++ return run(channel, context, fn).then( ++ result => (resolved(channel, context), result), ++ error => { ++ rejected(channel, context, error) ++ throw error ++ } ++ ) ++} ++ ++function serverOf({ host, port, path }) { ++ return path ++ ? { host: path, port: undefined } ++ : host && host.length === 1 ++ ? { host: host[0], port: port[0] } ++ : { host: undefined, port: undefined } ++} ++ ++function clientOf(config) { ++ return { database: config.database, ...serverOf(config), user: config.user, ssl: Boolean(config.ssl), processID: undefined } ++} ++ ++export function connectContext(config) { ++ const { database, host, port, user, ssl } = clientOf(config) ++ return { connection: { database, host, port, user, ssl } } ++} ++ ++export function connectAttempt(context, host, port) { ++ context.connection.host = host ++ context.connection.port = port ++} ++ ++export function connectSecured(context, ssl) { ++ context.connection.ssl = ssl ++} ++ ++export function transactionContext(config, session, options, savepoint) { ++ return { client: { ...clientOf(config), ...session }, options, savepoint, prepared: undefined } ++} ++ ++export function transactionSession(context, session) { ++ Object.assign(context.client, session) ++} ++ ++export function transactionCommitted(context, committed) { ++ context.result = { committed } ++} ++ ++export function poolContext({ totalCount, idleCount, waitingCount, max }) { ++ return { pool: { totalCount, idleCount, waitingCount, maxSize: max } } ++} ++ ++export function poolAcquired(context, processID, reused, queuedBehind) { ++ context.handoverTime = performance.timeOrigin + performance.now() ++ context.queuedBehind = queuedBehind ++ context.client = { processID, reused } ++} ++ ++export function poolHandover(q, queuedBehind, session) { ++ const wait = q.poolWait ++ wait.context.handoverTime = performance.timeOrigin + performance.now() ++ wait.context.queuedBehind = queuedBehind ++ wait.context.client = { processID: session.processID, reused: !wait.fresh } ++} ++ ++export function traceRoute(q, context, fn) { ++ const wait = q.poolWait = { context, fresh: false, routing: true, activated: false, failure: undefined } ++ try { ++ return run(poolConnectChannel, context, fn, q) ++ } finally { ++ wait.routing = false ++ settleRouted(q, wait) ++ } ++} ++ ++function settleRouted(q, wait) { ++ if (q.poolWait !== wait || !(wait.activated || wait.failure)) ++ return ++ ++ q.poolWait = undefined ++ if (wait.activated || wait.context.handoverTime !== undefined) ++ resolved(poolConnectChannel, wait.context) ++ else ++ rejected(poolConnectChannel, wait.context, wait.failure.error) ++ ++ wait.failure && queryFailed(q, wait.failure.error) ++} ++ ++export function poolStarted(q) { ++ const wait = q.poolWait ++ if (wait.routing) { ++ wait.activated = true ++ return ++ } ++ ++ q.poolWait = undefined ++ resolved(poolConnectChannel, wait.context) ++} ++ ++function poolFailed(q, error) { ++ const wait = q.poolWait ++ q.poolWait = undefined ++ rejected(poolConnectChannel, wait.context, error) ++} ++ ++export function poolReleased(connection, error) { ++ if (poolReleaseChannel && poolReleaseChannel.hasSubscribers) ++ poolReleaseChannel.publish({ client: { processID: connection.session.processID }, error }) ++} ++ ++export function poolRemoved(processID, error) { ++ poolRemoveChannel && poolRemoveChannel.hasSubscribers && poolRemoveChannel.publish({ client: { processID }, error }) ++} ++ ++export function traceQuery(q, fn) { ++ q.trace = { query: { text: undefined, name: undefined }, client: clientOf(q.handler.config || {}) } ++ return run(queryChannel, q.trace, fn, q) ++} ++ ++export function queryBuilt(q, session) { ++ const context = q.trace ++ context.query.text = q.string ++ context.query.name = q.statement.name || undefined ++ Object.assign(context.client, session) ++} ++ ++export function querySettled(q, result) { ++ if (q.streaming && result && typeof result.pipe === 'function') ++ return ++ ++ q.poolWait && poolStarted(q) ++ const context = q.trace ++ q.trace = undefined ++ context.result = result ++ resolved(queryChannel, context) ++} ++ ++export function queryFailed(q, error) { ++ if (q.poolWait) { ++ if (q.poolWait.routing) { ++ q.poolWait.failure = q.poolWait.failure || { error } ++ return ++ } ++ ++ poolFailed(q, error) ++ } ++ ++ const context = q.trace ++ q.trace = undefined ++ rejected(queryChannel, context, error) ++} diff --git a/src/transport.js b/src/transport.js new file mode 100644 index 0000000000000000000000000000000000000000..990ba27822fbfaca189a3a973c87ba10e58f1ec2 @@ -2766,7 +3154,7 @@ index 0000000000000000000000000000000000000000..e8a4bddf7c0f4b6c55ec7c5f33865888 + } +} diff --git a/types/index.d.ts b/types/index.d.ts -index 13c3432f761e2c2ee5b7dc1bdafe93e49f993dc7..7574d4b038b922d31236b0e9bfc3d27f5d18fe75 100644 +index 13c3432f761e2c2ee5b7dc1bdafe93e49f993dc7..09ebd712094980b56257a6095a80510adec52af5 100644 --- a/types/index.d.ts +++ b/types/index.d.ts @@ -1,4 +1,4 @@ @@ -2948,8 +3336,91 @@ index 13c3432f761e2c2ee5b7dc1bdafe93e49f993dc7..7574d4b038b922d31236b0e9bfc3d27f } interface TransactionSql = {}> extends ISql { -@@ -729,6 +800,12 @@ declare namespace postgres { - +@@ -727,8 +798,95 @@ declare namespace postgres { + prepare(name: string): Promise>; + } + ++ /** The connection a traced call runs on. Host, port and ssl are those of the connection once it is known. */ ++ interface TraceClient { ++ database: string; ++ host: string | undefined; ++ port: number | undefined; ++ user: string; ++ ssl: boolean; ++ /** Backend process id; set once the call is written to a connection */ ++ processID: number | undefined; ++ } ++ ++ /** Context of the `postgres:query` tracing channel. `query` and the connection fields of `client` are set when the query is written to a connection, after `start`. */ ++ interface QueryTraceContext { ++ query: { ++ /** SQL text with `$n` placeholders as sent to the server */ ++ text: string | undefined; ++ /** Prepared statement name; undefined for an unnamed statement */ ++ name: string | undefined; ++ }; ++ client: TraceClient; ++ /** Set before `asyncStart` when the query succeeded */ ++ result?: ResultMeta | Statement | Readable | Writable; ++ /** Set before `error` when the query failed */ ++ error?: Error; ++ } ++ ++ /** Context of the `postgres:connect` tracing channel, one span for each connection acquisition. */ ++ interface ConnectTraceContext { ++ connection: Omit; ++ error?: Error; ++ } ++ ++ /** Context of the `postgres:pool:connect` tracing channel: every wait for a connection, for `sql.reserve()`, `sql.begin()` and plain pooled queries. */ ++ interface PoolConnectTraceContext { ++ pool: { ++ totalCount: number; ++ idleCount: number; ++ waitingCount: number; ++ maxSize: number; ++ }; ++ /** Epoch milliseconds at which the connection took the call over; set before `asyncStart` once a connection took the call over. `handoverTime` minus the `start` time is the wait for a connection. */ ++ handoverTime?: number; ++ /** Queries the connection already had running or waiting when it took this one over. For a plain query, the time from `handoverTime` to `asyncStart` is spent behind them; a lease span ends a microtask after the hand-over, so that time falls inside its first statement. */ ++ queuedBehind?: number; ++ /** Set with `handoverTime` */ ++ client?: { ++ processID: number | undefined; ++ /** False when the acquisition had to open the connection */ ++ reused: boolean; ++ }; ++ error?: Error; ++ } ++ ++ /** Context of the `postgres:transaction` tracing channel, for `sql.begin()` and `sql.savepoint()`. */ ++ interface TransactionTraceContext { ++ client: TraceClient; ++ /** The mode string passed to `sql.begin()`; undefined for a savepoint */ ++ options: string | undefined; ++ /** The savepoint name as sent to the server; undefined for `sql.begin()` */ ++ savepoint: string | undefined; ++ /** The name given to `sql.prepare()` */ ++ prepared: string | undefined; ++ /** Set by `sql.begin()` on success: `committed` is false when the transaction was prepared */ ++ result?: { committed: boolean }; ++ error?: Error; ++ } ++ ++ /** Message of the `postgres:pool:release` channel. */ ++ interface PoolReleaseMessage { ++ client: { processID: number | undefined }; ++ /** The error the lease ended with */ ++ error: Error | undefined; ++ } ++ ++ /** Message of the `postgres:pool:remove` channel. */ ++ interface PoolRemoveMessage { ++ client: { processID: number | undefined }; ++ /** The error that closed the connection; undefined for a normal close */ ++ error: Error | undefined; ++ } ++ interface ReservedSql = {}> extends Sql { release(): void; + diff --git a/src/connection.js b/src/connection.js index 4e05655b..2522150c 100644 --- a/src/connection.js +++ b/src/connection.js @@ -12,6 +12,20 @@ import clampedTimeout from './timeout.js' import { SSLRequest, tlsConfig, cancelRequest } from './transport.js' import { Query, CLOSE } from './query.js' import b from './bytes.js' +import { + connectChannel, + watched, + run, + resolved, + rejected, + connectContext, + connectAttempt, + connectSecured, + queryBuilt, + poolHandover, + poolStarted, + poolRemoved +} from './tracing.js' export { Connection } @@ -113,6 +127,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose , statementCount = 1 , remaining = 0 , hostIndex = 0 + , serverIndex = 0 + , secured = false , length = 0 , rows = 0 , copyPaused = false @@ -243,6 +259,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose get active() { return query ? 1 : 0 }, get pipelined() { return sent.length }, get accepting() { return accepting() }, + get session() { return session() }, + get inflight() { return inflight() }, get copying() { return stream !== null }, get copyHeld() { return stream !== null && !stream.writableEnded }, get acquiring() { return acquisition && !acquisition.owner.cancelled ? 1 : 0 }, @@ -297,12 +315,36 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose return backoffSeconds(backoff, options.shared.retries) * 1000 } + function serverAddress() { + return options.path || host[serverIndex] + } + + function inflight() { + return (query ? 1 : 0) + sent.length + } + + function session() { + return { host: serverAddress(), port: serverPort(), ssl: secured, processID: backend.pid } + } + + function serverPort() { + return options.path ? undefined : port[serverIndex] + } + function standbyPass() { return target_session_attrs === 'prefer-standby' && host.length > 1 && acquisition.pass === 'standby' } function acquire(owner) { - acquisition = { owner, ending: false, pass: 'standby', hostsTried: 0, attempting: null, mismatches: [], lastError: null } + acquisition = { owner, ending: false, pass: 'standby', hostsTried: 0, attempting: null, mismatches: [], lastError: null, trace: undefined } + if (!watched(connectChannel)) + return startAcquisition() + + acquisition.trace = connectContext(options) + run(connectChannel, acquisition.trace, startAcquisition) + } + + function startAcquisition() { const wait = inheritedBackoff ? inheritedBackoff.at + inheritedBackoff.delay - performance.now() : 0 inheritedBackoff = null wait > 0 ? enterBackoff(wait) : enterConnecting() @@ -345,6 +387,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose function attach(created) { listen(created) socket = created + serverIndex = hostIndex + secured = false + acquisition.trace && connectAttempt(acquisition.trace, serverAddress(), serverPort()) acquisition.attempting = options.socket ? 'custom socket' : options.path || host[hostIndex] + ':' + port[hostIndex] if (options.socket) @@ -377,6 +422,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose const config = tlsConfig(options, raw) raw.removeAllListeners() socket = tls.connect(config) + secured = true socket.on('secureConnect', () => dispatch('secure')) listen(socket) } catch (err) { @@ -488,8 +534,10 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose acquisition = null endWaiters = [] transition(Phase.Closed) + a && a.trace && rejected(connectChannel, a.trace, err) a && queryError(a.owner, err) waiters.forEach(resolve => resolve()) + poolRemoved(a ? undefined : backend.pid, reported) onclose(connection, reported) } @@ -516,10 +564,11 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } function handoff() { - const { owner, ending } = acquisition + const { owner, ending, trace } = acquisition transition(Phase.Ready) clearAcquisitionTimers() acquisition = null + trace && (connectSecured(trace, secured), resolved(connectChannel, trace)) options.shared.retries = 0 lifeTimer.start() @@ -547,12 +596,13 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose return true q.owner === undefined && (q.owner = connection.owner) + q.poolWait && poolHandover(q, inflight(), session()) try { q.state = backend query ? sent.push(q) - : (query = q, query.active = true) + : (query = q, query.active = true, q.poolWait && poolStarted(q)) if (q.cancelled) { q.cancelled = null @@ -645,6 +695,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose : { string, types, name: q.prepare ? statementId + statementCount++ : '' } typeof options.debug === 'function' && callback(options.debug, id, string, parameters, types) + q.trace && queryBuilt(q, session()) } function write(x, fn) { @@ -854,6 +905,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose query = sent.length ? sent.shift() : null if (query) { query.active = true + query.poolWait && poolStarted(query) query.cancelled && cancelRequest(options, query.state).then(query.cancelled.resolve, query.cancelled.reject) } diff --git a/src/index.js b/src/index.js index db93ccaa..462dfe48 100644 --- a/src/index.js +++ b/src/index.js @@ -28,6 +28,19 @@ import Queue from './queue.js' import { Errors, PostgresError, callback } from './errors.js' import Subscribe from './subscribe.js' import largeObject from './large.js' +import { + transactionChannel, + poolConnectChannel, + watched, + traceRoute, + traceAsync, + transactionContext, + transactionSession, + transactionCommitted, + poolContext, + poolAcquired, + poolReleased +} from './tracing.js' const Lease = { Active: 0, Settling: 1, Released: 2, Ended: 3, Closed: 4 } @@ -94,6 +107,7 @@ function Postgres(a, b) { function Sql(handler) { handler.debug = options.debug + handler.config = options Object.entries(options.types).reduce((acc, [name, type]) => { acc[name] = (x) => new Parameter(x, type.to) @@ -137,7 +151,7 @@ function Postgres(a, b) { function file(path, args = [], options = {}) { arguments.length === 2 && !Array.isArray(args) && (options = args, args = []) - const query = new Query([], args, (query) => { + const read = query => { fs.readFile(path, 'utf8', (err, string) => { if (err) return query.reject(err) @@ -145,7 +159,9 @@ function Postgres(a, b) { query.strings = [string] handler(query) }) - }, cancel, { + } + read.config = handler.config + const query = new Query([], args, read, cancel, { ...options, simple: 'simple' in options ? options.simple : args.length === 0 }) @@ -260,29 +276,40 @@ function Postgres(a, b) { return await sql`select pg_notify(${ channel }, ${ '' + payload })` } - async function lease(auto) { + function lease(auto) { + if (!watched(poolConnectChannel)) + return acquire(auto) + + const trace = poolContext(stats()) + return traceAsync(poolConnectChannel, trace, () => acquire(auto, trace)) + } + + async function acquire(auto, trace) { if (ending) throw Errors.connection('CONNECTION_ENDED', options) - const l = { state: null, connection: null, error: null, onclose: null, queue: Queue() } + const l = { state: null, connection: null, error: null, outcome: undefined, behind: 0, onclose: null, queue: Queue() } + let waiter = null const idle = takeOpen() idle ? own(idle, l) : auto && !closed.length && busy.length ? own(busy.shift(), l) : await new Promise((resolve, reject) => { - const query = { reserve: c => resolve(own(c, l)), reject } + const query = waiter = { reserve: c => resolve(own(c, l)), reject, fresh: false } closed.length ? connect(closed.shift(), query) : allHeldByCopy() ? reject(copyInProgress()) : queries.push(query) }) + trace && poolAcquired(trace, l.connection.session.processID, !(waiter && waiter.fresh), l.behind) return l } function own(c, l) { l.connection = c + l.behind = c.inflight l.state = Lease.Active c.owner = l l.next = () => { @@ -316,10 +343,12 @@ function Postgres(a, b) { if (l.state === Lease.Released || l.state === Lease.Ended) return null + const lost = l.state === Lease.Closed l.state = state while (l.queue.length) l.queue.shift().reject(leaseError(l)) + poolReleased(l.connection, lost ? l.error : l.outcome) const c = l.connection if (c.owner !== l) return null @@ -346,9 +375,18 @@ function Postgres(a, b) { } } - async function begin(options, fn) { - !fn && (fn = options, options = '') + function begin(mode, fn) { + !fn && (fn = mode, mode = '') + if (!watched(transactionChannel)) + return transact(mode, fn) + + const trace = transactionContext(options, undefined, mode) + return traceAsync(transactionChannel, trace, () => transact(mode, fn, trace)) + } + + async function transact(options, fn, trace) { const l = await lease(true) + trace && transactionSession(trace, l.connection.session) let savepoints = 0 , prepare = null @@ -358,6 +396,7 @@ function Postgres(a, b) { async function transaction() { await Sql(handler).unsafe('begin ' + options.replace(/[^a-z ]/ig, ''), []).execute().catch(e => { + l.outcome = e e instanceof PostgresError && release() throw e }) @@ -370,6 +409,7 @@ function Postgres(a, b) { try { result = await statement(Sql(q => handler(q, Lease.Settling))) } catch (e) { + l.outcome = l.outcome || e terminate() throw e } @@ -395,6 +435,7 @@ function Postgres(a, b) { throw Errors.generic('INVALID_TRANSACTION_NAME', 'sql.prepare requires a non-empty string without null bytes') prepare = 'E\'' + x.replace(/\\/g, '\\\\').replace(/'/g, '\'\'') + '\'' + trace && (trace.prepared = x) } let uncaughtError , result @@ -409,6 +450,7 @@ function Postgres(a, b) { if (uncaughtError) throw uncaughtError } catch (e) { + name || (l.outcome = e) l.state === Lease.Active && await (name ? sql.unsafe('rollback to ' + escapeIdentifier(name)) : finish(final => final.unsafe('rollback')) @@ -430,6 +472,8 @@ function Postgres(a, b) { uncaughtError && { cause: uncaughtError } ) } + + trace && transactionCommitted(trace, prepare === null) } return result @@ -439,7 +483,10 @@ function Postgres(a, b) { return savepoint(sql => sql.apply(sql, arguments)) arguments.length === 1 && (fn = name, name = null) - return scope(fn, 's' + savepoints++ + (name ? '_' + name : '')) + const id = 's' + savepoints++ + (name ? '_' + name : '') + return watched(transactionChannel) + ? traceAsync(transactionChannel, transactionContext(options, l.connection.session, undefined, id), () => scope(fn, id)) + : scope(fn, id) } function track(q) { @@ -476,22 +523,32 @@ function Postgres(a, b) { } function handler(query) { + const refusal = query.trace && watched(poolConnectChannel) + ? traceRoute(query, poolContext(stats()), route) + : route(query) + refusal && query.reject(refusal) + } + + function route(query) { if (ending) - return query.reject(Errors.connection('CONNECTION_ENDED', options, options)) + return Errors.connection('CONNECTION_ENDED', options, options) const idle = takeOpen() if (idle) - return go(idle, query) + return (go(idle, query), null) if (closed.length) - return connect(closed.shift(), query) + return (connect(closed.shift(), query), null) const c = takeBusy() - c - ? go(c, query) - : allHeldByCopy() - ? query.reject(copyInProgress()) - : queries.push(query) + if (c) + return (go(c, query), null) + + if (allHeldByCopy()) + return copyInProgress() + + queries.push(query) + return null } function allHeldByCopy() { @@ -585,6 +642,7 @@ function Postgres(a, b) { } function connect(c, query) { + query.reserve ? query.fresh = true : query.poolWait && (query.poolWait.fresh = true) move(c, connecting) c.connect(query) return c diff --git a/src/query.js b/src/query.js index c1777a90..575c6014 100644 --- a/src/query.js +++ b/src/query.js @@ -1,3 +1,5 @@ +import { queryChannel, watched, traceQuery, querySettled, queryFailed } from './tracing.js' + const originCache = new Map() , originStackCache = new Map() , originError = Symbol('OriginError') @@ -24,13 +26,15 @@ export class Query extends Promise { this.statement = null // dd-trace's load-time rewrite only accepts single-expression arrows here - this.resolve = x => (this.active = false, this.settled = true, resolve(x)) - this.reject = x => (this.active = false, this.settled = true, reject(x)) + this.resolve = x => (this.trace && querySettled(this, x), this.active = false, this.settled = true, resolve(x)) + this.reject = x => (this.trace && queryFailed(this, x), this.active = false, this.settled = true, reject(x)) this.active = false this.settled = false this.cancelled = null this.executed = false + this.trace = undefined + this.poolWait = undefined this.signature = '' this[originError] = this.handler.debug @@ -105,8 +109,8 @@ export class Query extends Promise { resolve({ value, done: false }) return new Promise(r => prev = r) } - this.resolve = () => (this.active = false, resolve({ done: true })) - this.reject = x => (this.active = false, reject(x)) + this.resolve = x => (this.trace && querySettled(this, x), this.active = false, resolve({ done: true })) + this.reject = x => (this.trace && queryFailed(this, x), this.active = false, reject(x)) }) this.execute() return promise @@ -146,7 +150,15 @@ export class Query extends Promise { } async handle() { - !this.executed && (this.executed = true) && await 1 && this.handler(this) + if (this.executed) + return + + this.executed = true + if (!this.initialization && !this.settled && watched(queryChannel)) + return traceQuery(this, dispatch) + + await 1 + this.handler(this) } execute() { @@ -170,6 +182,11 @@ export class Query extends Promise { } } +async function dispatch(q) { + await 1 + q.handler(q) +} + function cachedError(xs) { if (originCache.has(xs)) return originCache.get(xs) diff --git a/src/tracing.js b/src/tracing.js new file mode 100644 index 00000000..42cefb34 --- /dev/null +++ b/src/tracing.js @@ -0,0 +1,186 @@ +import dc from 'diagnostics_channel' +import { performance } from 'perf_hooks' + +const supported = Boolean(dc) && typeof dc.tracingChannel === 'function' + +export const queryChannel = supported ? dc.tracingChannel('postgres:query') : undefined +export const connectChannel = supported ? dc.tracingChannel('postgres:connect') : undefined +export const transactionChannel = supported ? dc.tracingChannel('postgres:transaction') : undefined +export const poolConnectChannel = supported ? dc.tracingChannel('postgres:pool:connect') : undefined +export const poolReleaseChannel = supported ? dc.channel('postgres:pool:release') : undefined +export const poolRemoveChannel = supported ? dc.channel('postgres:pool:remove') : undefined + +export function watched(channel) { + return channel !== undefined && (channel.hasSubscribers ?? channel.start.hasSubscribers) +} + +export function run(channel, context, fn, ...args) { + try { + return channel.start.runStores(context, fn, undefined, ...args) + } finally { + channel.end.publish(context) + } +} + +export function resolved(channel, context) { + channel.asyncStart.publish(context) + channel.asyncEnd.publish(context) +} + +export function rejected(channel, context, error) { + context.error = error + channel.error.publish(context) + resolved(channel, context) +} + +export function traceAsync(channel, context, fn) { + return run(channel, context, fn).then( + result => (resolved(channel, context), result), + error => { + rejected(channel, context, error) + throw error + } + ) +} + +function serverOf({ host, port, path }) { + return path + ? { host: path, port: undefined } + : host && host.length === 1 + ? { host: host[0], port: port[0] } + : { host: undefined, port: undefined } +} + +function clientOf(config) { + return { database: config.database, ...serverOf(config), user: config.user, ssl: Boolean(config.ssl), processID: undefined } +} + +export function connectContext(config) { + const { database, host, port, user, ssl } = clientOf(config) + return { connection: { database, host, port, user, ssl } } +} + +export function connectAttempt(context, host, port) { + context.connection.host = host + context.connection.port = port +} + +export function connectSecured(context, ssl) { + context.connection.ssl = ssl +} + +export function transactionContext(config, session, options, savepoint) { + return { client: { ...clientOf(config), ...session }, options, savepoint, prepared: undefined } +} + +export function transactionSession(context, session) { + Object.assign(context.client, session) +} + +export function transactionCommitted(context, committed) { + context.result = { committed } +} + +export function poolContext({ totalCount, idleCount, waitingCount, max }) { + return { pool: { totalCount, idleCount, waitingCount, maxSize: max } } +} + +export function poolAcquired(context, processID, reused, queuedBehind) { + context.handoverTime = performance.timeOrigin + performance.now() + context.queuedBehind = queuedBehind + context.client = { processID, reused } +} + +export function poolHandover(q, queuedBehind, session) { + const wait = q.poolWait + wait.context.handoverTime = performance.timeOrigin + performance.now() + wait.context.queuedBehind = queuedBehind + wait.context.client = { processID: session.processID, reused: !wait.fresh } +} + +export function traceRoute(q, context, fn) { + const wait = q.poolWait = { context, fresh: false, routing: true, activated: false, failure: undefined } + try { + return run(poolConnectChannel, context, fn, q) + } finally { + wait.routing = false + settleRouted(q, wait) + } +} + +function settleRouted(q, wait) { + if (q.poolWait !== wait || !(wait.activated || wait.failure)) + return + + q.poolWait = undefined + if (wait.activated || wait.context.handoverTime !== undefined) + resolved(poolConnectChannel, wait.context) + else + rejected(poolConnectChannel, wait.context, wait.failure.error) + + wait.failure && queryFailed(q, wait.failure.error) +} + +export function poolStarted(q) { + const wait = q.poolWait + if (wait.routing) { + wait.activated = true + return + } + + q.poolWait = undefined + resolved(poolConnectChannel, wait.context) +} + +function poolFailed(q, error) { + const wait = q.poolWait + q.poolWait = undefined + rejected(poolConnectChannel, wait.context, error) +} + +export function poolReleased(connection, error) { + if (poolReleaseChannel && poolReleaseChannel.hasSubscribers) + poolReleaseChannel.publish({ client: { processID: connection.session.processID }, error }) +} + +export function poolRemoved(processID, error) { + poolRemoveChannel && poolRemoveChannel.hasSubscribers && poolRemoveChannel.publish({ client: { processID }, error }) +} + +export function traceQuery(q, fn) { + q.trace = { query: { text: undefined, name: undefined }, client: clientOf(q.handler.config || {}) } + return run(queryChannel, q.trace, fn, q) +} + +export function queryBuilt(q, session) { + const context = q.trace + context.query.text = q.string + context.query.name = q.statement.name || undefined + Object.assign(context.client, session) +} + +export function querySettled(q, result) { + if (q.streaming && result && typeof result.pipe === 'function') + return + + q.poolWait && poolStarted(q) + const context = q.trace + q.trace = undefined + context.result = result + resolved(queryChannel, context) +} + +export function queryFailed(q, error) { + if (q.poolWait) { + if (q.poolWait.routing) { + q.poolWait.failure = q.poolWait.failure || { error } + return + } + + poolFailed(q, error) + } + + const context = q.trace + q.trace = undefined + rejected(queryChannel, context, error) +} diff --git a/tests-startup/workerd-edge.mjs b/tests-startup/workerd-edge.mjs index d00446fe..b264c19b 100644 --- a/tests-startup/workerd-edge.mjs +++ b/tests-startup/workerd-edge.mjs @@ -6,7 +6,7 @@ import path from 'node:path' const scenarios = [ 'arrays-false', 'arrays-true', 'timers', 'end-timeout', 'end-plain', 'end-while-connecting', 'end-tls', - 'copy-out', 'subscribe', 'diagnostics-channel', 'ssl-verify', 'tls-private-ca', 'pooled', 'no-end', + 'copy-out', 'subscribe', 'diagnostics-channel', 'tracing-channels', 'ssl-verify', 'tls-private-ca', 'pooled', 'no-end', 'abort-neon', 'abort-neon-pooled', 'abort-supabase-pooler', 'abort-hyperdrive', 'abort-hyperdrive-pooled', 'abort-hyperdrive-supabase' ] const hyperdrive = [ @@ -24,9 +24,15 @@ const cpuVariants = [ const gate = { median: 1, p95: 1, exceeded: 0.02, p: 0.01, failures: 0.02 } async function prepare(out, compatibilityDate, release, base) { + const subscribe = ['query', 'all'].includes(process.env.AB_SUBSCRIBE) ? process.env.AB_SUBSCRIBE : '' await mkdir(out, { recursive: true }) await cp(path.join(repo, 'cf'), path.join(out, 'cf'), { recursive: true }) base && await cp(path.join(base, 'cf'), path.join(out, 'base/cf'), { recursive: true }) + if (subscribe && base) { + const tracing = path.join(out, 'base/cf/src/tracing.js') + await writeFile(tracing, (await readFile(tracing, 'utf8')).replaceAll('\'postgres:', '\'postgres-base:')) + } + await cp(path.join(repo, 'src'), path.join(out, 'src'), { recursive: true }) await cp(path.join(repo, 'tests-startup/workerd-worker.js'), path.join(out, 'worker.js')) await writeFile(path.join(out, 'driver.js'), [ @@ -35,6 +41,7 @@ async function prepare(out, compatibilityDate, release, base) { 'export { default as src } from \'./src/index.js\'', ...(release ? ['export { default as release } from \'@steadybrew/postgresjs\''] : []), ...(base ? ['export { default as base } from \'./base/cf/src/index.js\''] : []), + ...(subscribe ? ['export const subscribe = \'' + subscribe + '\''] : []), '' ].join('\n')) release && execFileSync('npm', ['install', '--prefix', out, '--no-save', '--no-audit', '--no-fund', release], { stdio: 'inherit' }) diff --git a/tests-startup/workerd-run.mjs b/tests-startup/workerd-run.mjs index da5db830..b1b3502e 100644 --- a/tests-startup/workerd-run.mjs +++ b/tests-startup/workerd-run.mjs @@ -86,7 +86,7 @@ async function main() { : [ 'cold-reserve-false', 'arrays-false', 'arrays-true', 'catalog-error', 'timers', 'connect-timeout', 'end-timeout', 'end-plain', 'end-while-connecting', - 'copy-out', 'ssl-prefer-declined', 'subscribe', 'diagnostics-channel', 'no-end', + 'copy-out', 'ssl-prefer-declined', 'subscribe', 'diagnostics-channel', 'tracing-channels', 'no-end', ...(tlsPort && process.env.WORKERD_TLS_CA ? tlsScenarios : []) ] for (const name of scenarios) { diff --git a/tests-startup/workerd-worker.js b/tests-startup/workerd-worker.js index 147f60f3..d239ba53 100644 --- a/tests-startup/workerd-worker.js +++ b/tests-startup/workerd-worker.js @@ -1,12 +1,28 @@ /* eslint no-console: 0 */ /* global addEventListener */ import * as drivers from './driver.js' +import dc from 'node:diagnostics_channel' +import { AsyncLocalStorage } from 'node:async_hooks' let unhandled = [] addEventListener('unhandledrejection', (event) => { const reason = event.reason?.stack || String(event.reason) unhandled.push(reason) console.error('WORKER_UNHANDLED_REJECTION', reason) }) +if (drivers.subscribe) { + const noop = () => undefined + const storage = new AsyncLocalStorage() + const names = drivers.subscribe === 'query' ? ['query'] : ['query', 'connect', 'pool:connect', 'transaction'] + for (const name of names) { + const channel = dc.tracingChannel('postgres:' + name) + channel.start.bindStore(storage, () => name) + channel.subscribe({ start: noop, end: noop, asyncStart: noop, asyncEnd: noop, error: noop }) + } + if (drivers.subscribe === 'all') { + dc.subscribe('postgres:pool:release', noop) + dc.subscribe('postgres:pool:remove', noop) + } +} const nativeSetTimeout = globalThis.setTimeout globalThis.setTimeout = function(callback, delay, ...extra) { if (extra.length) throw new TypeError('Failed to execute \'setTimeout\': parameter 3 is not of type \'Array\'.') @@ -181,6 +197,39 @@ const scenarios = { } return { tracingChannel: Boolean(tracing), events } }, + async 'tracing-channels'(port) { + const dc = await import('node:diagnostics_channel') + const { AsyncLocalStorage } = await import('node:async_hooks') + const storage = new AsyncLocalStorage() + const query = dc.tracingChannel('postgres:query') + const connection = dc.tracingChannel('postgres:connect') + equal([typeof dc.tracingChannel, typeof query.start.runStores, typeof query.start.bindStore], ['function', 'function', 'function']) + const seen = [] + const handlers = (name) => ({ + start: () => seen.push(name + ':start:' + storage.getStore()), + end: () => seen.push(name + ':end'), + asyncStart: () => seen.push(name + ':asyncStart'), + asyncEnd: (context) => seen.push(name + ':asyncEnd:' + (context.query ? context.query.text : context.connection.host)), + error: (context) => seen.push(name + ':error:' + context.error.message) + }) + const subscribed = [[query, handlers('query')], [connection, handlers('connect')]] + query.start.bindStore(storage, () => 'span') + subscribed.forEach(([channel, handler]) => channel.subscribe(handler)) + const sql = client(port, { max: 1 }) + try { + await within(sql`select 1 as x`, 5000, 'traced query') + } finally { + subscribed.forEach(([channel, handler]) => channel.unsubscribe(handler)) + query.start.unbindStore(storage) + await within(sql.end(), 3000, 'end') + } + const address = sql.options.host[0] + equal(seen, [ + 'query:start:span', 'query:end', 'connect:start:span', 'connect:end', 'connect:asyncStart', 'connect:asyncEnd:' + address, + 'query:asyncStart', 'query:asyncEnd:select 1 as x' + ]) + return { seen } + }, async 'tls-private-ca'() { const results = [] for (const ssl of ['require', 'prefer']) { diff --git a/tests-types/index.ts b/tests-types/index.ts index d1d7c719..b6dd60f2 100644 --- a/tests-types/index.ts +++ b/tests-types/index.ts @@ -27,3 +27,12 @@ new postgres.PostgresError({ code: '23505' }) new postgres.PostgresError('boom') // @ts-expect-error new postgres.PostgresError(42) + +const client: postgres.TraceClient = { database: 'a', host: 'localhost', port: 5432, user: 'b', ssl: false, processID: undefined } +const queryTrace: postgres.QueryTraceContext = { query: { text: undefined, name: undefined }, client } +const connectTrace: postgres.ConnectTraceContext = { connection: { database: 'a', host: 'localhost', port: 5432, user: 'b', ssl: false } } +const poolConnectTrace: postgres.PoolConnectTraceContext = { pool: { totalCount: 1, idleCount: 0, waitingCount: 0, maxSize: 10 }, handoverTime: 1, queuedBehind: 0, client: { processID: 1, reused: true } } +const transactionTrace: postgres.TransactionTraceContext = { client, options: 'read write', savepoint: undefined, prepared: undefined, result: { committed: true } } +const releaseMessage: postgres.PoolReleaseMessage = { client: { processID: 1 }, error: undefined } +const removeMessage: postgres.PoolRemoveMessage = { client: { processID: 1 }, error: undefined } +void [queryTrace, connectTrace, poolConnectTrace, transactionTrace, releaseMessage, removeMessage] diff --git a/tests/index.js b/tests/index.js index 311330b8..631a4b8f 100644 --- a/tests/index.js +++ b/tests/index.js @@ -7,6 +7,7 @@ import crypto from 'crypto' import postgres from '../src/index.js' import { vercelPool } from '../src/vercel.js' +import './tracing.js' const delay = ms => new Promise(r => setTimeout(r, ms)) const rel = x => new URL(x, import.meta.url) diff --git a/tests/tracing.js b/tests/tracing.js new file mode 100644 index 00000000..23d5cd5b --- /dev/null +++ b/tests/tracing.js @@ -0,0 +1,819 @@ +import dc from 'diagnostics_channel' +import { AsyncLocalStorage } from 'async_hooks' + +import { t } from './test.js' +import postgres from '../src/index.js' + +const snapshot = context => Object.fromEntries(Object.entries(context)) +const settle = () => new Promise(setImmediate) +const now = () => performance.timeOrigin + performance.now() +const rel = x => new URL(x, import.meta.url) +const stages = ['start', 'end', 'asyncStart', 'asyncEnd', 'error'] + +const options = { + db: 'postgres_js_test', + user: 'postgres_js_test', + idle_timeout: 1, + connect_timeout: 1, + max: 1 +} + +function collect(...names) { + const events = [] + const undo = [] + const ids = new Map() + const id = context => ids.get(context) || (ids.set(context, ids.size + 1), ids.size) + for (const name of names.filter(x => x.startsWith('postgres:'))) { + const handler = context => events.push({ name, stage: 'publish', id: id(context), time: now(), context: snapshot(context) }) + dc.subscribe(name, handler) + undo.push(() => dc.unsubscribe(name, handler)) + } + + for (const name of names.filter(x => !x.startsWith('postgres:'))) { + const channel = dc.tracingChannel('postgres:' + name) + const handlers = Object.fromEntries(stages.map(stage => + [stage, context => events.push({ name, stage, id: id(context), time: now(), context: snapshot(context) })] + )) + channel.subscribe(handlers) + undo.push(() => channel.unsubscribe(handlers)) + } + + return { events, stop: () => undo.forEach(x => x()) } +} + +const of = (events, name) => events.filter(x => x.name === name) +const sequence = (events, name) => of(events, name).map(x => x.stage).join(' ') +const ended = (events, name = 'query') => of(events, name).filter(x => x.stage === 'asyncEnd').map(x => x.context) +const plain = 'start end asyncStart asyncEnd' +const failed = 'start end error asyncStart asyncEnd' +const count = (events, stage, name = 'query') => of(events, name).filter(x => x.stage === stage).length +const settled = (events, name = 'query') => JSON.stringify([count(events, 'start', name), count(events, 'asyncEnd', name)]) +const server = o => ({ host: o.path || o.host[0], port: o.path ? undefined : o.port[0] }) + +async function traced(names, fn) { + const sql = postgres(options) + const { events, stop } = collect(...names) + try { + await fn(sql) + } finally { + stop() + await sql.end() + } + return events +} + +t('Tracing publishes a query in order with the context filled in as it is built', async() => { + const sql = postgres(options) + const { events, stop } = collect('query', 'postgres:pool:remove') + await sql`select ${ 5 }::int as x` + await sql.end() + stop() + const [done] = ended(events) + const [remove] = of(events, 'postgres:pool:remove') + const { host, port } = server(sql.options) + return [ + JSON.stringify([ + plain, 'select $1::int as x', true, 'SELECT', 1, + sql.options.database, sql.options.user, host, port, false, true, false, true + ]), + JSON.stringify([ + sequence(events, 'query'), done.query.text, typeof done.query.name === 'string', + done.result.command, done.result.count, done.client.database, done.client.user, done.client.host, done.client.port, done.client.ssl, + typeof done.client.processID === 'number', 'values' in done, + remove.context.client.processID === done.client.processID + ]) + ] +}) + +t('Tracing names a prepared statement and leaves an unnamed one without a name', async() => { + const events = await traced(['query'], async sql => { + await sql`select 1 as x` + await sql`select 2 as x`.simple() + }) + const [prepared, simple] = ended(events) + return [JSON.stringify([true, null]), JSON.stringify([typeof prepared.query.name === 'string', simple.query.name || null])] +}) + +t('Tracing publishes error before asyncStart and keeps the error on the context', async() => { + const events = await traced(['query'], sql => sql`select 1/0`.catch(() => undefined)) + const [done] = ended(events) + return [failed + '|22012', sequence(events, 'query') + '|' + done.error.code] +}) + +t('Tracing a cursor with a callback settles once', async() => { + const events = await traced(['query'], sql => sql`select * from generate_series(1, 5)`.cursor(2, () => undefined)) + return [JSON.stringify([1, 1]), settled(events)] +}) + +t('Tracing a cursor read with for await settles once, to the end and on break', async() => { + const events = await traced(['query'], async sql => { + for await (const rows of sql`select * from generate_series(1, 5)`.cursor(2)) + rows.length + for await (const rows of sql`select * from generate_series(1, 5)`.cursor(2)) { + rows.length + break + } + await sql`select 1` + }) + return [JSON.stringify([3, 3]), settled(events)] +}) + +t('Tracing forEach settles once', async() => { + const events = await traced(['query'], sql => sql`select * from generate_series(1, 5)`.forEach(() => undefined)) + return [JSON.stringify([1, 1]), settled(events)] +}) + +t('Tracing COPY streams settles each query once, when the copy is over', async() => { + const events = await traced(['query'], async sql => { + await sql`create temp table tracing_copy (x int)` + const writable = await sql`copy tracing_copy from stdin`.writable() + writable.end('1\n2\n') + await new Promise(resolve => writable.on('finish', resolve)) + const readable = await sql`copy (select * from tracing_copy order by x) to stdout`.readable() + readable.resume() + await new Promise(resolve => readable.on('end', resolve)) + await sql`select 1` + }) + const copies = ended(events).filter(x => /^copy/.test(x.query.text)) + return [ + JSON.stringify([4, 4, ['COPY', 'COPY']]), + JSON.stringify([count(events, 'start'), count(events, 'asyncEnd'), copies.map(x => x.result.command)]) + ] +}) + +t('Tracing describe, simple, unsafe and file settle once each', async() => { + const events = await traced(['query'], async sql => { + await sql`select 1 as x`.describe() + await sql`select 2 as x`.simple() + await sql.unsafe('select 3 as x') + await sql.unsafe('select $1::int as x', [4]) + await sql.file(rel('select.sql')) + }) + return [JSON.stringify([5, 5]), settled(events)] +}) + +t('Tracing pipelined queries settle once each with their own context', async() => { + const events = await traced(['query'], sql => Promise.all([sql`select 1 as x`, sql`select 2 as x`, sql`select 3 as x`])) + return [ + JSON.stringify(['select 1 as x', 'select 2 as x', 'select 3 as x']), + JSON.stringify(ended(events).map(x => x.query.text)) + ] +}) + +t('Tracing a query cancelled before it is sent settles with the cancellation and is never built', async() => { + const sql = postgres(options) + const { events, stop } = collect('query') + const slow = sql`select pg_sleep(0.2)`.execute() + const waiting = sql`select 2 as x` + const code = waiting.then(() => 'resolved', e => e.code) + await waiting.cancel() + await slow + const result = await code + stop() + await sql.end() + const never = of(events, 'query').find(x => x.stage === 'asyncEnd' && x.context.query.text === undefined) + const own = events.filter(x => x.id === never.id) + return [ + JSON.stringify(['57014', failed]), + JSON.stringify([result, own.map(x => x.stage).join(' ')]) + ] +}) + +t('Tracing a reserved connection settles each query once', async() => { + const sql = postgres(options) + const { events, stop } = collect('query') + const reserved = await sql.reserve() + await reserved`select 1 as x` + reserved.release() + stop() + await sql.end() + return [JSON.stringify([1, 1]), settled(events)] +}) + +t('Tracing a transaction publishes begin, its queries and a savepoint', async() => { + const events = await traced(['transaction', 'query'], sql => sql.begin('read write', async sql => { + await sql`select 1 as x` + await sql.savepoint('inner', sql => sql`select 2 as x`) + })) + const [savepoint, transaction] = ended(events, 'transaction') + return [ + JSON.stringify([ + 'start end start end asyncStart asyncEnd asyncStart asyncEnd', 'read write', 's0_inner', { committed: true }, null, + ['begin read write', 'select 1 as x', 'savepoint "s0_inner"', 'select 2 as x', 'commit'] + ]), + JSON.stringify([ + sequence(events, 'transaction'), transaction.options, savepoint.savepoint, transaction.result, savepoint.result || null, + ended(events).map(x => x.query.text) + ]) + ] +}) + +t('Tracing a rolled back transaction publishes the error', async() => { + const events = await traced(['transaction'], sql => sql.begin(() => { + throw new Error('stop') + }).catch(() => undefined)) + const [done] = ended(events, 'transaction') + return [failed + '|stop', sequence(events, 'transaction') + '|' + done.error.message] +}) + +t('Tracing a connection publishes its attempt and the removal', async() => { + const sql = postgres(options) + const { events, stop } = collect('connect', 'postgres:pool:remove') + await sql`select 1` + await sql.end() + stop() + const [done] = ended(events, 'connect') + const [remove] = of(events, 'postgres:pool:remove') + const { host, port } = server(sql.options) + return [ + JSON.stringify([plain, [sql.options.database, host, port, sql.options.user, false], true, null]), + JSON.stringify([ + sequence(events, 'connect'), + [done.connection.database, done.connection.host, done.connection.port, done.connection.user, done.connection.ssl], + typeof remove.context.client.processID === 'number', remove.context.error || null + ]) + ] +}) + +t('Tracing a connection that fails to authenticate or to connect publishes the error', async() => { + const { events, stop } = collect('connect') + const codes = [] + const wrong = postgres({ ...options, user: 'postgres_js_test_md5', pass: 'wrong' }) + codes.push(await wrong`select 1`.catch(e => e.code)) + const refused = postgres({ ...options, host: '127.0.0.1', port: 33333 }) + codes.push(await refused`select 1`.catch(e => e.code)) + stop() + return [ + JSON.stringify([['28P01', 'ECONNREFUSED'], failed + ' ' + failed, ['28P01', 'ECONNREFUSED']]), + JSON.stringify([codes, sequence(events, 'connect'), ended(events, 'connect').map(x => x.error.code)]) + ] +}) + +t('Tracing start runs in the AsyncLocalStorage context of the caller', async() => { + const storage = new AsyncLocalStorage() + const seen = [] + const transforms = [] + const sql = postgres({ ...options, debug: () => seen.push(storage.getStore()) }) + const channel = dc.tracingChannel('postgres:query') + const handlers = { start: () => seen.push('start:' + storage.getStore()) } + channel.start.bindStore(storage, () => (transforms.push(storage.getStore()), 'span')) + channel.subscribe(handlers) + try { + await sql`select 1` + seen.length = 0 + transforms.length = 0 + await storage.run('caller', async() => { + await sql`select 2` + }) + } finally { + channel.unsubscribe(handlers) + channel.start.unbindStore(storage) + await sql.end() + } + return [JSON.stringify([['caller'], ['start:span', 'span']]), JSON.stringify([transforms, seen])] +}) + +t('Tracing a transaction runs its function inside the transaction context', async() => { + const storage = new AsyncLocalStorage() + const sql = postgres(options) + const channel = dc.tracingChannel('postgres:transaction') + const handlers = { start: () => undefined } + channel.start.bindStore(storage, () => 'transaction') + channel.subscribe(handlers) + let seen + try { + await sql.begin(async() => { + seen = storage.getStore() + }) + } finally { + channel.unsubscribe(handlers) + channel.start.unbindStore(storage) + await sql.end() + } + return ['transaction', seen] +}) + +t('Nothing is published when nothing subscribes', async() => { + const published = [] + const restore = [] + const names = ['query', 'connect', 'transaction', 'pool:connect'].flatMap(name => stages.map(stage => 'tracing:postgres:' + name + ':' + stage)) + .concat('postgres:pool:release', 'postgres:pool:remove') + for (const name of names) { + const channel = dc.channel(name) + for (const method of ['publish', 'runStores']) { + const original = channel[method] + channel[method] = function() { + published.push(name + '.' + method) + return original.apply(this, arguments) + } + restore.push(() => delete channel[method]) + } + } + + const sql = postgres(options) + try { + await sql`select 1 as x` + await sql.begin(async sql => { + await sql`select 2 as x` + await sql.savepoint(sql => sql`select 3 as x`) + }) + const reserved = await sql.reserve() + await reserved`select 4 as x` + reserved.release() + } finally { + await sql.end() + restore.forEach(x => x()) + } + + return [JSON.stringify([]), JSON.stringify(published)] +}) + +t('Tracing contexts have no result or error key until one is set', async() => { + const events = await traced(['query', 'connect', 'transaction', 'pool:connect'], sql => sql.begin(async sql => { + await sql.savepoint(sql => sql`select 1 as x`) + })) + const early = events.filter(x => x.stage === 'start' || x.stage === 'end') + return [ + JSON.stringify([['connect', 'pool:connect', 'query', 'transaction'], 0]), + JSON.stringify([[...new Set(early.map(x => x.name))].sort(), early.filter(x => 'result' in x.context || 'error' in x.context).length]) + ] +}) + +const cancelledBeforeAwait = async(extra, warm) => { + const sql = postgres({ ...options, ...extra }) + warm && await sql`select 1` + const { events, stop } = collect('query') + const query = sql`select 2 as x` + query.cancel() + const code = await query.catch(e => e.code) + await sql`select 3 as x` + stop() + await sql.end() + return JSON.stringify([code, count(events, 'start'), count(events, 'asyncEnd')]) +} + +t('Tracing a query cancelled before it is awaited opens no span, with several connections', async() => + [JSON.stringify(['57014', 1, 1]), await cancelledBeforeAwait({ max: 2 }, true)] +) + +t('Tracing a query cancelled before it is awaited opens no span, with one connection', async() => + [JSON.stringify(['57014', 1, 1]), await cancelledBeforeAwait({ max: 1 }, true)] +) + +t('Tracing a query cancelled before it is awaited opens no span, on a pool that has to connect', async() => + [JSON.stringify(['57014', 1, 1]), await cancelledBeforeAwait({ max: 2 }, false)] +) + +t('Tracing a retried query settles once', async() => { + const sql = postgres(options) + const { events, stop } = collect('query') + await sql`drop table if exists tracing_retry` + await sql`create table tracing_retry (name text)` + await sql`insert into tracing_retry values ('1')` + await sql`select name from tracing_retry` + await sql`alter table tracing_retry alter column name type int using name::integer` + const [row] = await sql`select name from tracing_retry` + await sql`drop table tracing_retry` + stop() + await sql.end() + return [JSON.stringify([1, 7, 7]), JSON.stringify([row.name, count(events, 'start'), count(events, 'asyncEnd')])] +}) + +t('Tracing a server error in the middle of a cursor settles once with the error', async() => { + const events = await traced(['query'], async sql => { + await sql`select 1 / (x - 3) from generate_series(1, 5) x`.cursor(2, () => undefined).catch(() => undefined) + await sql`select 1` + }) + const [first] = ended(events) + return [JSON.stringify([2, 2, '22012']), JSON.stringify([count(events, 'start'), count(events, 'asyncEnd'), first.error && first.error.code])] +}) + +t('Tracing a cursor callback that throws settles once with the error', async() => { + const events = await traced(['query'], async sql => { + await sql`select * from generate_series(1, 5)`.cursor(2, () => { + throw new Error('boom') + }).catch(() => undefined) + await sql`select 1` + }) + const [first] = ended(events) + return [JSON.stringify([2, 2, 'boom']), JSON.stringify([count(events, 'start'), count(events, 'asyncEnd'), first.error && first.error.message])] +}) + +t('Tracing a query cancelled in the queue of a reserved connection settles once', async() => { + const sql = postgres({ ...options, max_pipeline: 1 }) + const { events, stop } = collect('query') + const reserved = await sql.reserve() + const first = reserved`select pg_sleep(0.1)`.execute() + const second = reserved`select 2 as x`.execute() + const third = reserved`select 3 as x`.execute() + const code = await third.cancel().then(() => third.catch(e => e.code)) + await Promise.allSettled([first, second]) + reserved.release() + stop() + await sql.end() + return [JSON.stringify(['57014', 3, 3]), JSON.stringify([code, count(events, 'start'), count(events, 'asyncEnd')])] +}) + +t('Tracing a query cancelled in the queue of a transaction settles once', async() => { + const sql = postgres({ ...options, max_pipeline: 1 }) + const { events, stop } = collect('query') + await sql.begin(async sql => { + const queries = [sql`select pg_sleep(0.1)`.execute(), sql`select 2 as x`.execute(), sql`select 3 as x`.execute()] + await queries[2].cancel() + await Promise.allSettled(queries) + }).catch(() => undefined) + stop() + await sql.end() + return [ + JSON.stringify([true, 1]), + JSON.stringify([count(events, 'start') === count(events, 'asyncEnd'), ended(events).filter(x => x.error && x.error.code === '57014').length]) + ] +}) + +t('Tracing queries rejected because the pool is ending settle once', async() => { + const sql = postgres({ ...options, max: 1 }) + const { events, stop } = collect('query') + const running = sql`select pg_sleep(0.1)`.execute() + const queued = sql`select 2 as x`.execute() + await Promise.allSettled([running, queued, sql.end()]) + stop() + return [JSON.stringify([2, 2]), settled(events)] +}) + +t('Tracing a pool checkout reports the pool at start and whether the connection was reused', async() => { + const sql = postgres(options) + const { events, stop } = collect('pool:connect') + const first = await sql.reserve() + first.release() + const second = await sql.reserve() + second.release() + stop() + await sql.end() + const starts = of(events, 'pool:connect').filter(x => x.stage === 'start').map(x => x.context.pool) + const [one, two] = ended(events, 'pool:connect') + return [ + JSON.stringify([ + plain + ' ' + plain, + [{ totalCount: 0, idleCount: 0, waitingCount: 0, maxSize: 1 }, { totalCount: 1, idleCount: 1, waitingCount: 0, maxSize: 1 }], + [false, true], true + ]), + JSON.stringify([ + sequence(events, 'pool:connect'), starts, [one.client.reused, two.client.reused], + typeof one.client.processID === 'number' && one.client.processID === two.client.processID + ]) + ] +}) + +t('Tracing a pool checkout that fails publishes the error and no client', async() => { + const { events, stop } = collect('pool:connect') + const refused = postgres({ ...options, host: '127.0.0.1', port: 33333 }) + const codes = [await refused.reserve().catch(e => e.code)] + const ended_ = postgres(options) + await ended_.end() + codes.push(await ended_.reserve().catch(e => e.code)) + stop() + const done = ended(events, 'pool:connect') + return [ + JSON.stringify([['ECONNREFUSED', 'CONNECTION_ENDED'], failed + ' ' + failed, [false, false], ['ECONNREFUSED', 'CONNECTION_ENDED']]), + JSON.stringify([codes, sequence(events, 'pool:connect'), done.map(x => 'client' in x), done.map(x => x.error.code)]) + ] +}) + +t('Tracing a release publishes once, with the connection of the checkout', async() => { + const sql = postgres(options) + const { events, stop } = collect('pool:connect', 'postgres:pool:release') + const reserved = await sql.reserve() + reserved.release() + reserved.release() + stop() + await sql.end() + const releases = of(events, 'postgres:pool:release') + const [checkout] = ended(events, 'pool:connect') + return [ + JSON.stringify([1, true, null]), + JSON.stringify([releases.length, releases[0].context.client.processID === checkout.client.processID, releases[0].context.error || null]) + ] +}) + +t('Tracing a transaction publishes its release with the error that ended it', async() => { + const sql = postgres(options) + const { events, stop } = collect('pool:connect', 'postgres:pool:release', 'transaction') + await sql.begin(async sql => { + await sql`select 1` + }) + await sql.begin(() => { + throw new Error('stop') + }).catch(() => undefined) + stop() + await sql.end() + const releases = of(events, 'postgres:pool:release').map(x => x.context) + const checkouts = ended(events, 'pool:connect') + return [ + JSON.stringify([2, [null, 'stop'], [true, true], 2]), + JSON.stringify([ + releases.length, releases.map(x => x.error ? x.error.message : null), + releases.map((x, i) => x.client.processID === checkouts[i].client.processID), checkouts.length + ]) + ] +}) + +const spans = (events, name = 'pool:connect') => ended(events, name) +const at = (events, name, stage, nth = 0) => events.indexOf(of(events, name).filter(x => x.stage === stage)[nth]) +const asyncEndOf = (events, text) => events.indexOf(of(events, 'query').find(x => x.stage === 'asyncEnd' && x.context.query.text === text)) + +t('Tracing a query on an idle connection has a pool span that starts and ends at once', async() => { + const sql = postgres(options) + await sql`select 1` + const { events, stop } = collect('query', 'pool:connect') + await sql`select 2 as x` + stop() + await sql.end() + const [span] = spans(events) + const start = of(events, 'pool:connect').find(x => x.stage === 'start') + return [ + JSON.stringify([plain, 0, true, true, true, true]), + JSON.stringify([ + sequence(events, 'pool:connect'), span.queuedBehind, span.handoverTime >= start.time, span.client.reused, + typeof span.client.processID === 'number', at(events, 'query', 'start') < at(events, 'pool:connect', 'start') + ]) + ] +}) + +t('Tracing a query that opens its connection nests the connect span inside the pool span', async() => { + const storage = new AsyncLocalStorage() + const sql = postgres(options) + const pool = dc.tracingChannel('postgres:pool:connect') + const connect = dc.tracingChannel('postgres:connect') + const seen = [] + const probe = { start: () => seen.push(storage.getStore()) } + pool.start.bindStore(storage, () => 'pool') + connect.subscribe(probe) + const { events, stop } = collect('query', 'pool:connect', 'connect') + try { + await sql`select 1` + } finally { + stop() + connect.unsubscribe(probe) + pool.start.unbindStore(storage) + await sql.end() + } + const [span] = spans(events) + return [ + JSON.stringify([['pool'], false, plain, true, true]), + JSON.stringify([ + seen, span.client.reused, sequence(events, 'pool:connect'), + at(events, 'pool:connect', 'start') < at(events, 'connect', 'start'), + at(events, 'connect', 'asyncEnd') < at(events, 'pool:connect', 'asyncEnd') + ]) + ] +}) + +t('Tracing a query sent ahead to a busy connection ends its pool span when the earlier query is done', async() => { + const sql = postgres(options) + await sql`select 1` + const { events, stop } = collect('query', 'pool:connect') + const slow = sql`select pg_sleep(0.2)`.execute() + await settle() + const waiting = sql`select 2 as x` + await Promise.all([slow, waiting]) + stop() + await sql.end() + const [first, second] = spans(events) + return [ + JSON.stringify([0, true, true, true]), + JSON.stringify([ + first.queuedBehind, second.queuedBehind >= 1, second.client.reused, + asyncEndOf(events, 'select pg_sleep(0.2)') < at(events, 'pool:connect', 'asyncEnd', 1) + ]) + ] +}) + +t('Tracing queries drained from the pool queue when the connection opens wait behind the first', async() => { + const sql = postgres(options) + const { events, stop } = collect('query', 'pool:connect') + await Promise.all([sql`select 1 as x`, sql`select 2 as x`, sql`select 3 as x`]) + stop() + await sql.end() + const found = spans(events) + const pairs = [0, 1, 2].map(i => at(events, 'pool:connect', 'asyncEnd', i) < at(events, 'query', 'asyncEnd', i)) + return [ + JSON.stringify([3, 3, [false, true, true], true, [true, true, true]]), + JSON.stringify([ + count(events, 'asyncEnd', 'pool:connect'), count(events, 'start', 'pool:connect'), found.map(x => x.client.reused), + found[1].queuedBehind >= 1, pairs + ]) + ] +}) + +t('Tracing fails the pool span before the query when the pool is ending', async() => { + const sql = postgres(options) + await sql`select 1` + const { events, stop } = collect('query', 'pool:connect') + const ending = sql.end() + const code = await sql`select 2 as x`.catch(e => e.code) + await ending + stop() + const [span] = spans(events) + return [ + JSON.stringify(['CONNECTION_ENDED', failed, 'CONNECTION_ENDED', true]), + JSON.stringify([ + code, sequence(events, 'pool:connect'), span.error.code, + at(events, 'pool:connect', 'asyncEnd') < at(events, 'query', 'asyncEnd') + ]) + ] +}) + +t('Tracing fails the pool span of a query refused during a COPY', async() => { + const sql = postgres(options) + await sql`create temp table tracing_pool_copy (x int)` + const writable = await sql`copy tracing_pool_copy from stdin`.writable() + const { events, stop } = collect('query', 'pool:connect') + const code = await sql`select 1`.catch(e => e.code) + stop() + writable.end('1\n') + await new Promise(resolve => writable.on('finish', resolve)) + await sql.end() + const [span] = spans(events) + return [ + JSON.stringify(['COPY_IN_PROGRESS', failed, 'COPY_IN_PROGRESS']), + JSON.stringify([code, sequence(events, 'pool:connect'), span.error.code]) + ] +}) + +t('Tracing fails the pool span of a query cancelled in the pool queue', async() => { + const sql = postgres(options) + await sql`select 1` + const { events, stop } = collect('query', 'pool:connect') + const first = sql`select pg_sleep(${ 0.1 })`.execute() + const waiting = sql`select 2 as x` + const code = waiting.then(() => 'resolved', e => e.code) + await settle() + await waiting.cancel() + await first + const result = await code + stop() + await sql.end() + const found = spans(events) + return [ + JSON.stringify(['57014', ['57014', 'ok'], 2, 2]), + JSON.stringify([ + result, found.map(x => x.error ? x.error.code : 'ok').sort(), count(events, 'asyncEnd', 'pool:connect'), count(events, 'start', 'pool:connect') + ]) + ] +}) + +t('Tracing fails the pool span of a query whose connection attempt fails', async() => { + const { events, stop } = collect('query', 'pool:connect', 'connect') + const refused = postgres({ ...options, host: '127.0.0.1', port: 33333 }) + const code = await refused`select 1`.catch(e => e.code) + stop() + return [ + JSON.stringify(['ECONNREFUSED', failed, failed, true]), + JSON.stringify([ + code, sequence(events, 'pool:connect'), sequence(events, 'connect'), + at(events, 'pool:connect', 'asyncEnd') < at(events, 'query', 'asyncEnd') + ]) + ] +}) + +t('Tracing fails the pool span of a query that waited behind others on a connection that is lost', async() => { + const sql = postgres(options) + const killer = postgres({ ...options, max: 1 }) + const [{ pid }] = await sql`select pg_backend_pid() as pid` + const { events, stop } = collect('query', 'pool:connect') + const running = sql`select pg_sleep(5)`.execute() + await settle() + const waiting = sql`select 2 as x`.execute() + await killer`select pg_terminate_backend(${ pid })` + const outcomes = await Promise.allSettled([running, waiting]) + stop() + await Promise.all([sql.end(), killer.end()]) + const failures = of(events, 'pool:connect').filter(x => x.stage === 'error') + const failed_ = events.find(x => x.name === 'pool:connect' && x.stage === 'asyncEnd' && x.context.error) + return [ + JSON.stringify(['rejected', 'rejected', 1, true, true]), + JSON.stringify([ + outcomes[0].status, outcomes[1].status, failures.length, + count(events, 'start', 'pool:connect') === count(events, 'asyncEnd', 'pool:connect'), + events.indexOf(failed_) < asyncEndOf(events, 'select 2 as x') + ]) + ] +}) + +t('Tracing opens no pool span when only one of the query and pool channels has a subscriber', async() => { + const sql = postgres(options) + const poolOnly = collect('pool:connect') + await sql`select 1` + poolOnly.stop() + const published = [] + const channel = dc.channel('tracing:postgres:pool:connect:start') + const original = channel.runStores + channel.runStores = function() { + published.push('runStores') + return original.apply(this, arguments) + } + const queryOnly = collect('query') + try { + await sql`select 2` + } finally { + queryOnly.stop() + delete channel.runStores + await sql.end() + } + return [JSON.stringify([0, []]), JSON.stringify([poolOnly.events.length, published])] +}) + +t('Tracing a query that fails to build while it is routed settles its pool span once, after end', async() => { + const sql = postgres(options) + await sql`select 1` + const idle = collect('query', 'pool:connect') + await sql`select ${ undefined }`.catch(() => undefined) + idle.stop() + const slow = sql`select pg_sleep(0.2)`.execute() + await settle() + const busy = collect('query', 'pool:connect') + await sql`select ${ undefined }`.catch(() => undefined) + busy.stop() + await slow + await sql.end() + const summary = events => [ + sequence(events, 'pool:connect'), sequence(events, 'query'), count(events, 'asyncEnd', 'pool:connect'), count(events, 'asyncEnd'), + at(events, 'pool:connect', 'asyncEnd') < at(events, 'query', 'asyncEnd') + ] + return [ + JSON.stringify([[plain, failed, 1, 1, true], [plain, failed, 1, 1, true]]), + JSON.stringify([summary(idle.events), summary(busy.events)]) + ] +}) + +t('Tracing statements inside begin and reserve publish no plain pool span', async() => { + const sql = postgres(options) + const { events, stop } = collect('query', 'pool:connect') + await sql.begin(async sql => { + await sql`select 1` + }) + const reserved = await sql.reserve() + await reserved`select 2` + reserved.release() + stop() + await sql.end() + return [ + JSON.stringify([2, 2, 4]), + JSON.stringify([count(events, 'start', 'pool:connect'), count(events, 'asyncEnd', 'pool:connect'), count(events, 'start')]) + ] +}) + +t('Tracing a lease taken on a busy connection reports the queries ahead of it', async() => { + const sql = postgres(options) + await sql`select 1` + const slow = sql`select pg_sleep(0.2)`.execute() + await settle() + const { events, stop } = collect('pool:connect') + await sql.begin(async sql => { + await sql`select 1` + }) + stop() + await slow + await sql.end() + const [span] = spans(events) + return [JSON.stringify([1, true]), JSON.stringify([span.queuedBehind, span.client.reused])] +}) + +const reconnected = async(queued) => { + const sql = postgres(options) + const killer = postgres({ ...options, max: 1 }) + const [{ pid }] = await sql`select pg_backend_pid() as pid` + await killer`select 1` + const { events, stop } = collect('query', 'pool:connect') + const running = sql`select pg_sleep(${ 5 })`.execute() + await settle() + const waiting = queued === 'lease' ? sql.reserve() : sql`select 2 as x`.execute() + await killer`select pg_terminate_backend(${ pid })` + const [, outcome] = await Promise.allSettled([running, waiting]) + stop() + outcome.status === 'fulfilled' && queued === 'lease' && outcome.value.release() + await Promise.all([sql.end(), killer.end()]) + return JSON.stringify([outcome.status, spans(events).filter(x => x.client && x.client.reused === false).length]) +} + +t('Tracing a query served by the reconnect of a closed connection reports that it opened the connection', async() => + [JSON.stringify(['fulfilled', 1]), await reconnected('query')] +) + +t('Tracing a lease served by the reconnect of a closed connection reports that it opened the connection', async() => + [JSON.stringify(['fulfilled', 1]), await reconnected('lease')] +) + +t('Tracing a lease served after the initial query of a connection counts that query as ahead of it', async() => { + const sql = postgres(options) + const { events, stop } = collect('pool:connect') + const first = sql`select pg_sleep(0.1)`.execute() + await settle() + const reserved = await sql.reserve() + reserved.release() + stop() + await first + await sql.end() + const [span] = spans(events) + return [JSON.stringify([1, true]), JSON.stringify([span.queuedBehind, span.client.reused])] +}) diff --git a/transpile.deno.js b/transpile.deno.js index 25ea9dac..2eea9e45 100644 --- a/transpile.deno.js +++ b/transpile.deno.js @@ -25,7 +25,7 @@ fs.readdirSync('src').forEach(name => ) ) -fs.readdirSync('tests').forEach(name => +fs.readdirSync('tests').filter(name => name !== 'tracing.js').forEach(name => fs.writeFileSync( path.join(tests, name), name.endsWith('.js') @@ -44,8 +44,10 @@ function transpile(x, name, folder) { .replace(/\nexec\(/g, '\nawait exec(') .replace('{ spawnSync }', '{ spawn }') } - if (name === 'index.js') + if (name === 'index.js') { + x = x.replace('import \'./tracing.js\'\n', '') x += '\n;globalThis.addEventListener("unload", () => Deno.exit(process.exitCode))' + } } const buffer = x.includes('Buffer') @@ -74,6 +76,7 @@ function transpile(x, name, folder) { '(query.writable.push({ chunk }), callback())' ) .replace('socket.setKeepAlive(true, 1000 * keep_alive)', 'socket.setKeepAlive(true)') + .replace('from \'diagnostics_channel\'', 'from \'node:diagnostics_channel\'') .replace('node:stream', std + 'node/stream.ts') .replace('import net from \'net\'', 'import { net } from \'../polyfills.js\'') .replace('import tls from \'tls\'', 'import { tls } from \'../polyfills.js\'') diff --git a/types/index.d.ts b/types/index.d.ts index 7574d4b0..09ebd712 100644 --- a/types/index.d.ts +++ b/types/index.d.ts @@ -798,6 +798,87 @@ declare namespace postgres { prepare(name: string): Promise>; } + /** The connection a traced call runs on. Host, port and ssl are those of the connection once it is known. */ + interface TraceClient { + database: string; + host: string | undefined; + port: number | undefined; + user: string; + ssl: boolean; + /** Backend process id; set once the call is written to a connection */ + processID: number | undefined; + } + + /** Context of the `postgres:query` tracing channel. `query` and the connection fields of `client` are set when the query is written to a connection, after `start`. */ + interface QueryTraceContext { + query: { + /** SQL text with `$n` placeholders as sent to the server */ + text: string | undefined; + /** Prepared statement name; undefined for an unnamed statement */ + name: string | undefined; + }; + client: TraceClient; + /** Set before `asyncStart` when the query succeeded */ + result?: ResultMeta | Statement | Readable | Writable; + /** Set before `error` when the query failed */ + error?: Error; + } + + /** Context of the `postgres:connect` tracing channel, one span for each connection acquisition. */ + interface ConnectTraceContext { + connection: Omit; + error?: Error; + } + + /** Context of the `postgres:pool:connect` tracing channel: every wait for a connection, for `sql.reserve()`, `sql.begin()` and plain pooled queries. */ + interface PoolConnectTraceContext { + pool: { + totalCount: number; + idleCount: number; + waitingCount: number; + maxSize: number; + }; + /** Epoch milliseconds at which the connection took the call over; set before `asyncStart` once a connection took the call over. `handoverTime` minus the `start` time is the wait for a connection. */ + handoverTime?: number; + /** Queries the connection already had running or waiting when it took this one over. For a plain query, the time from `handoverTime` to `asyncStart` is spent behind them; a lease span ends a microtask after the hand-over, so that time falls inside its first statement. */ + queuedBehind?: number; + /** Set with `handoverTime` */ + client?: { + processID: number | undefined; + /** False when the acquisition had to open the connection */ + reused: boolean; + }; + error?: Error; + } + + /** Context of the `postgres:transaction` tracing channel, for `sql.begin()` and `sql.savepoint()`. */ + interface TransactionTraceContext { + client: TraceClient; + /** The mode string passed to `sql.begin()`; undefined for a savepoint */ + options: string | undefined; + /** The savepoint name as sent to the server; undefined for `sql.begin()` */ + savepoint: string | undefined; + /** The name given to `sql.prepare()` */ + prepared: string | undefined; + /** Set by `sql.begin()` on success: `committed` is false when the transaction was prepared */ + result?: { committed: boolean }; + error?: Error; + } + + /** Message of the `postgres:pool:release` channel. */ + interface PoolReleaseMessage { + client: { processID: number | undefined }; + /** The error the lease ended with */ + error: Error | undefined; + } + + /** Message of the `postgres:pool:remove` channel. */ + interface PoolRemoveMessage { + client: { processID: number | undefined }; + /** The error that closed the connection; undefined for a normal close */ + error: Error | undefined; + } + interface ReservedSql = {}> extends Sql { release(): void;