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-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-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) { diff --git a/tests/index.js b/tests/index.js index 26767ad4..f525f5ae 100644 --- a/tests/index.js +++ b/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] +})