From 9015a0e777a6666233375742cd77f73932884ded Mon Sep 17 00:00:00 2001 From: Oleg Komarov Date: Sun, 25 Oct 2026 11:46:14 +0000 Subject: [PATCH 1/3] ci: run the tracing comparison on every pull request and cover COPY, cancel and build failure in it dd-trace and Sentry rewrite and wrap the driver by shape, and npm run test:tracing, which packs the fork and compares the spans against Postgres.js 3.4.9, was only ever run by hand. It now runs as a job on every pull request, and a release waits for it like it waits for the test jobs. The comparison had no COPY scenario, but dd-trace ends a COPY span on the second resolve of the query, so a change that settles a query once would leave every COPY span open and nothing would notice. The harness gains COPY in and out, a query cancelled before it is sent and a query that fails to build, the paths the request lifecycle restructure (#73) touches. --- .github/workflows/test.yml | 17 ++++++++++++++++- tests-tracing/harness.mjs | 21 ++++++++++++++++++++- 2 files changed, 36 insertions(+), 2 deletions(-) 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/tests-tracing/harness.mjs b/tests-tracing/harness.mjs index 9ac9ac49..5cff7ebc 100644 --- a/tests-tracing/harness.mjs +++ b/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) { From 9a254d03013d6029c7ebaedb412beb9852ae5801 Mon Sep 17 00:00:00 2001 From: Oleg Komarov Date: Sun, 25 Oct 2026 11:48:17 +0000 Subject: [PATCH 2/3] test: pin request routing and settling before the restructure (#73) The restructure of the pool and query side must not change behaviour, so these tests assert what main does today on real PostgreSQL, including paths that look wrong. They pass on main and are the reference the refactor commits are held to. The query's resolve and reject are counted with a spy. The counts pin current behaviour: COPY in and out resolve twice (the stream, then the result, which dd-trace relies on), sql`begin` on a pool resolves after it rejected with UNSAFE_TRANSACTION, and a query cancelled before it is awaited rejects once at max: 2 and three times at max: 1, where the refused statement is written and the server's reply to it also trips BindComplete on the unbuilt statement. The restructure settles each query once except COPY, so the later commit changes those entries in one table at the top of the group. A query cancelled before it is awaited is still routed when it is awaited. It is never built and the pool keeps working. An idle connection is filed busy afterwards, while a connection that was still closed opens idle. This is pinned for max: 1 and max: 2, warm and cold, so a later fix shows up as a deliberate change. Also covered: a reservation queued behind plain queries on a connection that is still opening gets it after them, and close() during the first query's connection start runs that query and lets the next one reopen. --- tests/index.js | 153 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) diff --git a/tests/index.js b/tests/index.js index 26767ad4..3b687520 100644 --- a/tests/index.js +++ b/tests/index.js @@ -5828,3 +5828,156 @@ 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 } +} + +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]) + ] +}) From 0fa9f2709b4f6bd150ea3eb95a99e3146a9f156a Mon Sep 17 00:00:00 2001 From: Oleg Komarov Date: Sun, 25 Oct 2026 21:30:53 +0000 Subject: [PATCH 3/3] test: temporary check of the refactor's pinned counts against main --- tests-startup/run.js | 2 +- tests-startup/startup.js | 24 ++++++++++++------ tests/index.js | 54 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 71 insertions(+), 9 deletions(-) diff --git a/tests-startup/run.js b/tests-startup/run.js index a64a9742..08e0ce84 100644 --- a/tests-startup/run.js +++ b/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', diff --git a/tests-startup/startup.js b/tests-startup/startup.js index 8ef4e759..3c028b97 100644 --- a/tests-startup/startup.js +++ b/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) diff --git a/tests/index.js b/tests/index.js index 3b687520..f525f5ae 100644 --- a/tests/index.js +++ b/tests/index.js @@ -5834,7 +5834,9 @@ const pinned = { copyIn: { resolve: 2, reject: 0 }, unsafeTransaction: { resolve: 1, reject: 1 }, cancelledBeforeAwaitedAtMax1: { resolve: 0, reject: 3 }, - cancelledBeforeAwaitedAtMax2: { resolve: 0, reject: 1 } + cancelledBeforeAwaitedAtMax2: { resolve: 0, reject: 1 }, + cancelledInReservedQueue: { resolve: 0, reject: 3 }, + cancelledInTransactionQueue: { resolve: 0, reject: 3 } } function counted(query) { @@ -5981,3 +5983,53 @@ t('a pool closed while its first query is connecting runs that query and reopens 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] +})