check: main counts for the restructure's pinned rows (do not merge)

.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

tests-startup/run.js

@@ -35,7 +35,7 @@ const phaseNames = ['stale-ending-lifetime', 'stale-ending-close', 'stale-ending
                     'cancel-pipelined', 'cancel-pipelined-lost', 'cancel-settled', 'cancel-request-tls-error',
                     'cancel-request-refused', 'cancel-unawaited', 'cancel-query-unawaited',
                     'first-query-pipeline', 'release-inflight', 'draining-release-busy', 'factory-undefined', 'factory-empty',
-                    'ending-queued-reconnect', 'ending-queued-down', 'ending-other-open', 'error-close-reconnect',
+                    'ending-queued-reconnect', 'ending-queued-down', 'ending-queued-down-string', 'ending-other-open', 'error-close-reconnect',
                     'fatal-catalog', 'end-queued-cold',
                     'delay-clamp', 'idle-clamp', 'idle-throw', 'closing-bounded', 'closing-clean', 'closing-no-error',
                     'handout-idle', 'handout-lifetime',

tests-startup/startup.js

@@ -410,7 +410,7 @@ export async function phases(name, postgres, onEvent) {
   const base = { host: '127.0.0.1', user: 'fixture', database: 'fixture', ssl: false, max: 1, fetch_types: true, onnotice: () => { /* Quiet. */ } }
   const hang = ['forced-queued', 'cancel-errors', 'gap-query', 'gap-query-pool', 'stale-ending-lifetime', 'stale-ending-close',
                 'stale-ending-rst', 'fin-inflight', 'rst-inflight', 'begin-socket-cause',
-                'cancel-request-refused', 'cancel-unawaited', 'ending-queued-reconnect', 'ending-queued-down',
+                'cancel-request-refused', 'cancel-unawaited', 'ending-queued-reconnect', 'ending-queued-down', 'ending-queued-down-string',
                 'ending-other-open', 'terminate-ready', 'terminate-draining', 'terminate-pipeline'].includes(name)
   const cause = name.startsWith('cause-') ? name.split('-')[2] : ''
   const holding = ['reserve-end', 'cancel-initial', 'cancel-initial-untyped', 'failover-timeout',
@@ -523,6 +523,7 @@ export async function phases(name, postgres, onEvent) {
       await sleep(500)
       assert(calls - before < 20, 'Factory called ' + (calls - before) + ' times in 500 ms')
       await sql.end({ timeout: 0 })
+      await until(() => sql.stats().listenCount === 0)
     } else if (name === 'fin-inflight' || name === 'rst-inflight') {
       const sql = make({ connect_timeout: 2 })
       await marker(sql)
@@ -944,27 +945,36 @@ export async function phases(name, postgres, onEvent) {
       }
       assert.deepStrictEqual(unhandled, [])
       await sql.end({ timeout: 0 })
-    } else if (name === 'ending-queued-reconnect' || name === 'ending-queued-down') {
+    } else if (name === 'ending-queued-reconnect' || name.startsWith('ending-queued-down')) {
       let calls = 0
-      const down = name === 'ending-queued-down'
+      const down = name.startsWith('ending-queued-down')
+      const thrown = name.endsWith('-string') ? 'factory denied' : new Error('factory denied')
       const options = { connect_timeout: 2, max_pipeline: 0 }
       if (down) {
         const factory = tracked(server)
         options.socket = () => {
           if (calls++)
-            throw new Error('factory denied')
+            throw thrown
           return factory.socket()
         }
       }
       const sql = make(options)
       const hung = settle(sql.unsafe('select hang', [], { simple: true }))
       await until(() => server.events.some(x => x.sql === 'select hang'))
-      const queued = [1, 2, 3].map(x => settle(sql.unsafe('select ' + x, [], { simple: true }).execute()))
+      const queries = [1, 2, 3].map(x => sql.unsafe('select ' + x, [], { simple: true }).execute())
+      const queued = queries.map(x => settle(x))
+      const raw = queries.map(x => x.then(() => 'resolved', error => error))
       const ending = settle(sql.end())
       server.reset()
       assert.strictEqual(await hung, 'rejected:ECONNRESET')
-      const expected = down ? 'rejected:factory denied' : 'resolved'
-      assert.deepStrictEqual(await Promise.all(queued), [expected, expected, expected])
+      if (typeof thrown === 'string') {
+        const [opener, ...rest] = await Promise.all(raw)
+        assert(opener instanceof Error && opener.message === thrown)
+        assert.deepStrictEqual(rest, [thrown, thrown])
+      } else {
+        const expected = down ? 'rejected:factory denied' : 'resolved'
+        assert.deepStrictEqual(await Promise.all(queued), [expected, expected, expected])
+      }
       assert.strictEqual(await ending, 'resolved')
       if (down) {
         assert.strictEqual(calls, 2)

tests-tracing/harness.mjs

@@ -23,7 +23,26 @@ export const scenarios = {
     await tx`select 1 as x`
     throw new Error('rollback')
   }).catch(() => {}),
-  savepoint: sql => sql.begin(tx => tx.savepoint(sp => sp`select 1 as x`))
+  savepoint: sql => sql.begin(tx => tx.savepoint(sp => sp`select 1 as x`)),
+  copyOut: async sql => {
+    const readable = await sql`copy (select generate_series(1, 3)) to stdout`.readable()
+    await new Promise((resolve, reject) => readable.on('data', () => {}).on('end', resolve).on('error', reject))
+  },
+  copyIn: async sql => {
+    await sql`drop table if exists tracing_copy`
+    await sql`create table tracing_copy (x int)`
+    const writable = await sql`copy tracing_copy from stdin`.writable()
+    writable.write('1\n')
+    writable.end('2\n')
+    await new Promise((resolve, reject) => writable.on('finish', resolve).on('error', reject))
+    await sql`drop table tracing_copy`
+  },
+  cancelBeforeSend: async sql => {
+    const query = sql`select 1 as x`
+    await query.cancel()
+    await query.catch(() => {})
+  },
+  buildFailure: sql => sql`select ${ sql.json(1n) } as x`.catch(() => {})
 }
 
 export async function tracer(mode, events) {

tests/index.js

@@ -5828,3 +5828,208 @@ for (const [name, thrown] of unextensibleThrows) {
     ]
   })
 }
+
+const pinned = {
+  copyOut: { resolve: 2, reject: 0 },
+  copyIn: { resolve: 2, reject: 0 },
+  unsafeTransaction: { resolve: 1, reject: 1 },
+  cancelledBeforeAwaitedAtMax1: { resolve: 0, reject: 3 },
+  cancelledBeforeAwaitedAtMax2: { resolve: 0, reject: 1 },
+  cancelledInReservedQueue: { resolve: 0, reject: 3 },
+  cancelledInTransactionQueue: { resolve: 0, reject: 3 }
+}
+
+function counted(query) {
+  const calls = { resolve: 0, reject: 0 }
+  const { resolve, reject } = query
+  query.resolve = x => (calls.resolve++, resolve(x))
+  query.reject = x => (calls.reject++, reject(x))
+  return calls
+}
+
+t('a copy to stdout resolves its query twice, with the stream and then with the result', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1 })
+  const query = sql`copy (select generate_series(1, 10)) to stdout`
+  const calls = counted(query)
+  const readable = await query.readable()
+  let lines = 0
+  for await (const chunk of readable)
+    lines += chunk.toString().split('\n').length - 1
+  await sql`select 1`
+  await sql.end()
+
+  return [
+    JSON.stringify([10, pinned.copyOut]),
+    JSON.stringify([lines, calls])
+  ]
+})
+
+t('a copy from stdin resolves its query twice, with the stream and then with the result', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1 })
+  await sql`create table test (x int)`
+  const query = sql`copy test from stdin`
+  const calls = counted(query)
+  const writable = await query.writable()
+  const handedOut = { ...calls }
+  const finished = new Promise(r => writable.on('finish', r))
+  writable.end('1\n2\n')
+  await finished
+  const rows = (await sql`select count(*)::int as n from test`)[0].n
+  await sql`drop table test`
+  await sql.end()
+
+  return [
+    JSON.stringify([{ resolve: 1, reject: 0 }, 2, pinned.copyIn]),
+    JSON.stringify([handedOut, rows, calls])
+  ]
+})
+
+t('a transaction started with sql`begin` on a pool rejects with UNSAFE_TRANSACTION after its command completed', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 2, fetch_types: false })
+  const query = sql`begin`
+  const calls = counted(query)
+  const error = await query.catch(e => e)
+  await sql.end()
+
+  return [
+    JSON.stringify(['UNSAFE_TRANSACTION', pinned.unsafeTransaction]),
+    JSON.stringify([error.code, calls])
+  ]
+})
+
+t('a query cancelled before it is awaited is never built and the pool keeps working with max: 1', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1 })
+  await sql`select 1`
+  const query = sql`select 2 as x`
+  const calls = counted(query)
+  await query.cancel()
+  const error = await query.catch(e => e)
+  const next = (await sql`select 3 as x`)[0].x
+  await sql.end()
+
+  return [
+    JSON.stringify(['57014', undefined, pinned.cancelledBeforeAwaitedAtMax1, 3]),
+    JSON.stringify([error.code, query.string, calls, next])
+  ]
+})
+
+t('a query cancelled before it is awaited is never built and files an idle connection busy with max: 2 warm', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 2 })
+  await sql`select 1`
+  const query = sql`select 2 as x`
+  const calls = counted(query)
+  await query.cancel()
+  const error = await query.catch(e => e)
+  await new Promise(setImmediate)
+  const stats = sql.stats()
+  const next = (await sql`select 3 as x`)[0].x
+  await sql.end()
+
+  return [
+    JSON.stringify(['57014', undefined, pinned.cancelledBeforeAwaitedAtMax2, { busyCount: 1, idleCount: 0 }, 3]),
+    JSON.stringify([error.code, query.string, calls, { busyCount: stats.busyCount, idleCount: stats.idleCount }, next])
+  ]
+})
+
+t('a query cancelled before it is awaited is never built and opens its connection idle with max: 2 cold', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 2 })
+  const query = sql`select 2 as x`
+  const calls = counted(query)
+  await query.cancel()
+  const error = await query.catch(e => e)
+  await until(() => sql.stats().connectingCount === 0)
+  const stats = sql.stats()
+  const next = (await sql`select 3 as x`)[0].x
+  await sql.end()
+
+  return [
+    JSON.stringify(['57014', undefined, pinned.cancelledBeforeAwaitedAtMax2, { busyCount: 0, idleCount: 1 }, 3]),
+    JSON.stringify([error.code, query.string, calls, { busyCount: stats.busyCount, idleCount: stats.idleCount }, next])
+  ]
+})
+
+t('a reservation queued behind plain queries on a connection that is opening gets it after them', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1 })
+  const order = []
+  const first = sql`select 1 as x`.then(() => order.push('first'))
+  const second = sql`select 2 as x`.then(() => order.push('second'))
+  await 1
+  const reserved = await sql.reserve()
+  const held = sql.stats().reservedCount
+  const third = await reserved`select 3 as x`.then(x => (order.push('reserved'), x[0].x))
+  await Promise.all([first, second])
+  reserved.release()
+  const after = (await sql`select 4 as x`)[0].x
+  await sql.end()
+
+  return [
+    JSON.stringify([['first', 'second', 'reserved'], 1, 3, 4]),
+    JSON.stringify([order, held, third, after])
+  ]
+})
+
+t('a pool closed while its first query is connecting runs that query and reopens for the next one', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1 })
+  const first = sql`select 1 as x`.execute()
+  await 1
+  const closing = sql.close()
+  const ran = (await first)[0].x
+  await closing
+  const next = (await sql`select 2 as x`)[0].x
+  await sql.end()
+
+  return [
+    JSON.stringify([1, 2]),
+    JSON.stringify([ran, next])
+  ]
+})
+
+t('a query cancelled in the queue of a reserved connection settles once', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1, max_pipeline: 0 })
+  const reserved = await sql.reserve()
+  const running = reserved`select pg_sleep(0.1)`.execute()
+  const doomed = reserved`select 2 as x`
+  const calls = counted(doomed)
+  doomed.execute()
+  await new Promise(setImmediate)
+  await doomed.cancel()
+  const code = await doomed.then(() => 'resolved', e => e.code)
+  await running
+  const next = (await reserved`select 3 as x`)[0].x
+  reserved.release()
+  await sql.end()
+
+  return [
+    JSON.stringify([pinned.cancelledInReservedQueue, '57014', 3]),
+    JSON.stringify([calls, code, next])
+  ]
+})
+
+t('a query cancelled in the queue of a transaction settles once', { timeout: 10 }, async() => {
+  const sql = postgres({ ...options, max: 1, max_pipeline: 0 })
+  let calls
+  const code = await sql.begin(async tx => {
+    const running = tx`select pg_sleep(0.1)`.execute()
+    const doomed = tx`select 2 as x`
+    calls = counted(doomed)
+    doomed.execute()
+    await new Promise(setImmediate)
+    await doomed.cancel()
+    await running
+  }).then(() => 'resolved', e => e.code)
+  await sql.end()
+
+  return [
+    JSON.stringify([pinned.cancelledInTransactionQueue, '57014']),
+    JSON.stringify([calls, code])
+  ]
+})
+
+t('a cursor whose callback throws a string rejects with that string', async() => {
+  const sql = postgres({ ...options, max: 1 })
+  const thrown = 'boom'
+  const rejected = await sql`select generate_series(1, 3) as x`.cursor(1, () => { throw thrown }).then(() => 'resolved', e => e)
+  await sql.end()
+
+  return [thrown, rejected]
+})