feat(pg): add abortable query cancellation

docs/pages/apis/client.mdx

@@ -92,6 +92,9 @@ type QueryConfig {
   // custom type parsers just for this query result
   types?: Types;
 
+  // cancel queued or active work when this signal aborts
+  signal?: AbortSignal;
+
   // TODO: document
   queryMode?: string;
 }
@@ -151,6 +154,53 @@ console.log(result.rows) // ['brianc']
 await client.end()
 ```
 
+### Query cancellation
+
+Pass an `AbortSignal` in a query config object to cancel queued or active work:
+
+```js
+const controller = new AbortController()
+const query = client.query({
+  text: 'SELECT pg_sleep(30)',
+  signal: controller.signal,
+})
+
+controller.abort()
+await query // rejects with PostgreSQL error code 57014 if cancellation wins
+```
+
+A query aborted before submission rejects with `signal.reason` and sends nothing. Once submitted, cancellation races with normal completion: PostgreSQL may return error `57014`, or the query may finish normally first. An aborted transaction remains failed until you issue `ROLLBACK`.
+
+For explicit cancellation, use the exported helper with the client that owns the query:
+
+```js
+import { cancelQuery } from 'pg'
+
+const query = client.query('SELECT pg_sleep(30)')
+const cancellation = cancelQuery(client)
+
+await Promise.allSettled([query, cancellation])
+```
+
+`cancelQuery(client)` resolves `true` after the cancellation connection finishes and the target client reaches `ReadyForQuery`. It resolves `false` when there is no active query and rejects if the cancel request cannot be completed. A `true` result does not guarantee that cancellation won the race.
+
+See [Query cancellation](/features/query-cancellation) for the separate-connection design, completion barrier, and race outcomes.
+
+Cancellation uses a separate connection to the selected PostgreSQL endpoint. A custom `stream` must therefore be a factory; a concrete custom stream instance cannot support cancellation. AbortSignal and explicit cancellation are not supported in pipeline mode.
+
+With a pool, check out a client so the query and cancellation target are unambiguous. Do not use a `PoolClient` after `release()`:
+
+```js
+const client = await pool.connect()
+try {
+  const query = client.query('SELECT pg_sleep(30)')
+  const cancellation = cancelQuery(client)
+  await Promise.allSettled([query, cancellation])
+} finally {
+  client.release()
+}
+```
+
 **client.query with a `Submittable`**
 
 If you pass an object to `client.query` and the object has a `.submit` function on it, the client will pass it's PostgreSQL server connection to the object and delegate query dispatching to the supplied object. This is an advanced feature mostly intended for library authors. It is incidentally also currently how the callback and promise based queries above are handled internally, but this is subject to change. It is also how [pg-cursor](https://github.com/brianc/node-pg-cursor) and [pg-query-stream](https://github.com/brianc/node-pg-query-stream) work.

docs/pages/features/_meta.js

@@ -1,6 +1,7 @@
 export default {
   connecting: 'Connecting',
   queries: 'Queries',
+  'query-cancellation': 'Query cancellation',
   pipelining: 'Pipelining',
   pooling: 'Pooling',
   transactions: 'Transactions',

docs/pages/features/queries.mdx

@@ -79,6 +79,8 @@ const res = await client.query(query)
 console.log(res.rows[0])
 ```
 
+The config object can also include an [`AbortSignal`](/apis/client#query-cancellation). This is available for `client.query` and `pool.query`; use a checked-out client when you need the explicit `cancelQuery(client)` helper. See [Query cancellation](/features/query-cancellation) for the lifecycle and race semantics.
+
 The query config object allows for a few more advanced scenarios:
 
 ### Prepared statements

docs/pages/features/query-cancellation.mdx

@@ -0,0 +1,46 @@
+---
+title: Query cancellation
+---
+
+## Why cancellation uses another connection
+
+PostgreSQL cannot receive a cancellation command on a connection while that connection is busy running a query. node-postgres therefore opens a short-lived connection to the same selected endpoint and sends a `CancelRequest` containing the active backend's process ID and secret key.
+
+This does not check out another `pg.Pool` client or create an authenticated database session. Concurrent cancellation calls for one client share the same operation, but different clients can still open short-lived cancel sockets at the same time and put pressure on PostgreSQL or proxy connection acceptance.
+
+The cancellation connection reuses the selected TCP address or Unix socket, SSL mode, and logical TLS server name. This matters when DNS returns multiple addresses or a connection uses failover: sending the request to a different server would not cancel the active backend.
+
+## Why cancellation has two completion conditions
+
+A cancellation request and the original query complete independently:
+
+1. The cancellation connection must finish writing the `CancelRequest` and reach EOF.
+2. The original query connection must reach `ReadyForQuery`.
+
+node-postgres pauses that client's query queue until both conditions are met. Without this barrier, a delayed cancellation request could arrive after the next query starts and cancel the wrong work.
+
+`cancelQuery(client)` resolving `true` means both sides completed safely. It does not mean cancellation won the race.
+
+## Race outcomes
+
+PostgreSQL decides the outcome once a query has been submitted:
+
+- If the query finishes first, its normal result is preserved.
+- If cancellation wins, the query rejects with PostgreSQL error code `57014`.
+- If an `AbortSignal` fires before submission, the query rejects with `signal.reason` and sends nothing.
+
+When delivery may have started but cannot be confirmed, node-postgres closes the original connection. This fail-closed behavior prevents an uncertain late cancellation request from affecting later work. A manual cancellation that fails before any write leaves the original query running and rejects only the `cancelQuery` promise.
+
+## Pools and transactions
+
+Explicit cancellation needs the client that owns the active query, so use `pool.connect()` rather than `pool.query()`. Keep the client checked out until the query and cancellation operation settle, then release it. A released `PoolClient` reference must not be reused.
+
+Cancellation does not roll back a transaction. When PostgreSQL cancels a statement inside a transaction, the transaction remains failed until the application issues `ROLLBACK`.
+
+## Limitations
+
+- Query cancellation is not supported in pipeline mode because multiple queries may already be in flight.
+- A concrete custom `stream` instance cannot create the second connection. Supply a stream factory when cancellation is required.
+- Successful dispatch cannot guarantee server-side cancellation because normal query completion may win the race.
+
+See the [`client.query` cancellation API](/apis/client#query-cancellation) for signatures and examples.

packages/pg/esm/index.mjs

@@ -10,6 +10,7 @@ export const Query = pg.Query
 export const DatabaseError = pg.DatabaseError
 export const escapeIdentifier = pg.escapeIdentifier
 export const escapeLiteral = pg.escapeLiteral
+export const cancelQuery = pg.cancelQuery
 export const Result = pg.Result
 export const TypeOverrides = pg.TypeOverrides
 

packages/pg/lib/abort.js

@@ -0,0 +1,23 @@
+'use strict'
+
+function isAbortSignal(signal) {
+  return (
+    signal &&
+    typeof signal === 'object' &&
+    typeof signal.aborted === 'boolean' &&
+    typeof signal.addEventListener === 'function' &&
+    typeof signal.removeEventListener === 'function'
+  )
+}
+
+function getAbortReason(signal) {
+  if (signal.reason !== undefined) {
+    return signal.reason
+  }
+  const error = new Error('This operation was aborted')
+  error.name = 'AbortError'
+  error.code = 'ABORT_ERR'
+  return error
+}
+
+module.exports = { getAbortReason, isAbortSignal }

packages/pg/lib/cancel-connection.js

@@ -0,0 +1,129 @@
+'use strict'
+
+module.exports = function cancelConnection(Connection, connection, processID, secretKey, timeoutMillis, signal) {
+  const rejectBeforeConnect = (message) => {
+    const error = new Error(message)
+    Object.defineProperty(error, 'cancelDispatchMayHaveStarted', { value: false })
+    return Promise.reject(error)
+  }
+
+  if (connection._hasCustomStream && !connection._streamFactory) {
+    return rejectBeforeConnect('Cannot cancel a query using a concrete custom stream instance')
+  }
+
+  const port = connection._cancelPort || connection._connectPort
+  const host = connection._cancelHost || connection._connectHost
+  if (port === undefined || port === null) {
+    return rejectBeforeConnect('Cannot cancel a query before the connection endpoint is known')
+  }
+
+  const transport = new Connection({
+    stream: connection._streamFactory || undefined,
+    ssl: connection.ssl,
+    sslNegotiation: connection.sslNegotiation,
+    sslServername: connection._sslServername || connection._connectHost,
+  })
+  const timeout = timeoutMillis > 0 ? timeoutMillis : 5000
+
+  return new Promise((resolve, reject) => {
+    let settled = false
+    let writeAttempted = false
+    let writeCompleted = false
+
+    const cleanup = (keepErrorListener) => {
+      clearTimeout(timer)
+      transport.removeListener('connect', onConnect)
+      transport.removeListener('sslconnect', sendCancel)
+      if (!keepErrorListener) {
+        transport.removeListener('error', fail)
+      }
+      transport.removeListener('end', onEnd)
+      signal?.removeEventListener?.('abort', onAbort)
+    }
+
+    const finish = (error) => {
+      if (settled) {
+        return
+      }
+      settled = true
+      if (error) {
+        cleanup(true)
+        Object.defineProperty(error, 'cancelDispatchMayHaveStarted', { value: writeAttempted })
+        transport.stream.destroy?.()
+        reject(error)
+      } else {
+        cleanup(false)
+        resolve()
+      }
+    }
+
+    const fail = (error) => finish(error instanceof Error ? error : new Error(String(error)))
+
+    const onAbort = () => {
+      const reason = signal.reason
+      fail(reason instanceof Error ? reason : new Error('Cancel request aborted'))
+    }
+
+    const onEnd = () => {
+      if (!writeCompleted) {
+        fail(new Error('Cancel connection ended before the request was written'))
+        return
+      }
+      finish()
+    }
+
+    const sendCancel = () => {
+      if (settled || writeAttempted) {
+        return
+      }
+      writeAttempted = true
+      try {
+        const accepted = transport.cancel(processID, secretKey, (error) => {
+          if (error) {
+            fail(error)
+            return
+          }
+          writeCompleted = true
+        })
+        if (accepted === false) {
+          writeAttempted = false
+          fail(new Error('Cancel connection is not writable'))
+        }
+      } catch (error) {
+        fail(error)
+      }
+    }
+
+    const onConnect = () => {
+      if (!transport.ssl) {
+        sendCancel()
+      } else if (transport.sslNegotiation !== 'direct') {
+        transport.requestSsl()
+      }
+    }
+
+    const timer = setTimeout(() => {
+      const error = new Error('Cancel request timeout')
+      error.code = 'PG_CANCEL_TIMEOUT'
+      fail(error)
+    }, timeout)
+    timer.unref?.()
+
+    transport.on('connect', onConnect)
+    transport.on('sslconnect', sendCancel)
+    transport.on('error', fail)
+    transport.on('end', onEnd)
+    signal?.addEventListener?.('abort', onAbort, { once: true })
+
+    if (signal?.aborted) {
+      onAbort()
+      return
+    }
+
+    try {
+      transport.connect(port, host)
+    } catch (error) {
+      fail(error)
+    }
+  })
+}

packages/pg/lib/cancel-query.js

@@ -0,0 +1,8 @@
+'use strict'
+
+module.exports = function cancelQuery(client) {
+  if (!client || typeof client._cancelQuery !== 'function') {
+    return Promise.reject(new TypeError('cancelQuery requires a connected pg Client'))
+  }
+  return client._cancelQuery(false)
+}