From 920a8aa505d0410b5057eff3ab11cf134bbbb839 Mon Sep 17 00:00:00 2001 From: Ibrahim Hafez Date: Thu, 29 Oct 2026 15:52:17 +0300 Subject: [PATCH 1/3] fix(pg-cursor): handle exceptions from row type parsers --- packages/pg-cursor/index.js | 28 ++++++- packages/pg-cursor/test/type-parser-errors.js | 80 +++++++++++++++++++ .../test/type-parser-errors.ts | 80 +++++++++++++++++++ 3 files changed, 185 insertions(+), 3 deletions(-) create mode 100644 packages/pg-cursor/test/type-parser-errors.js create mode 100644 packages/pg-query-stream/test/type-parser-errors.ts diff --git a/packages/pg-cursor/index.js b/packages/pg-cursor/index.js index 5d566882b..e77a64b48 100644 --- a/packages/pg-cursor/index.js +++ b/packages/pg-cursor/index.js @@ -115,7 +115,16 @@ class Cursor extends EventEmitter { } handleDataRow(msg) { - const row = this._result.parseRow(msg.fields) + if (this._error) return + + let row + try { + row = this._result.parseRow(msg.fields) + } catch (err) { + // Drain the current batch before closing the portal and reporting the error. + this._error = err + return + } this.emit('row', row, this._result) this._rows.push(row) } @@ -144,10 +153,17 @@ class Cursor extends EventEmitter { } handlePortalSuspended() { + if (this._error) { + return this._closePortal() + } this._sendRows() } handleReadyForQuery() { + if (this._error) { + this.state = 'done' + return this._reportError(this._error) + } this._sendRows() this.state = 'done' this.emit('end', this._result) @@ -174,10 +190,16 @@ class Cursor extends EventEmitter { } this.state = 'error' + this._reportError(this._error || msg) + } + + _reportError(msg) { this._error = msg // satisfy any waiting callback if (this._cb) { - this._cb(msg) + const cb = this._cb + this._cb = null + cb(msg) } // dispatch error to all waiting callbacks for (let i = 0; i < this._queue.length; i++) { @@ -250,7 +272,7 @@ class Cursor extends EventEmitter { this._getRows(rows, cb) } else if (this.state === 'busy' || this.state === 'initialized') { this._queue.push([rows, cb]) - } else if (this.state === 'error') { + } else if (this.state === 'error' || (this.state === 'done' && this._error)) { setImmediate(() => cb(this._error)) } else if (this.state === 'done') { setImmediate(() => cb(null, [])) diff --git a/packages/pg-cursor/test/type-parser-errors.js b/packages/pg-cursor/test/type-parser-errors.js new file mode 100644 index 000000000..f5000e812 --- /dev/null +++ b/packages/pg-cursor/test/type-parser-errors.js @@ -0,0 +1,80 @@ +'use strict' +const assert = require('assert') +const Cursor = require('../') +const pg = require('pg') + +describe('type parser errors', () => { + let client + + beforeEach(async () => { + client = new pg.Client() + await client.connect() + }) + + afterEach(async () => { + await client.end() + }) + + it('preserves the parser error if a later row causes a server error', async () => { + const parserError = new Error('cannot decode value') + const types = { + getTypeParser: () => (value) => { + if (value === '3') throw parserError + return Number(value) + }, + } + const text = 'SELECT value, 10 / (5 - value) AS quotient FROM generate_series(1, 6) value' + await assert.rejects(client.query({ text, types }), (error) => error === parserError) + + const cursor = client.query(new Cursor(text, [], { types })) + await assert.rejects(cursor.read(20), (error) => error === parserError) + const result = await client.query('SELECT 42 AS answer') + assert.deepStrictEqual(result.rows, [{ answer: 42 }]) + }) + + for (const rowMode of [undefined, 'array']) { + for (const batchSize of [2, 20]) { + it(`reports a parser error once with rowMode=${rowMode} and batchSize=${batchSize}`, async () => { + const parserError = new Error('cannot decode value') + const parsedValues = [] + const types = { + getTypeParser: () => (value) => { + parsedValues.push(value) + if (value === '3') throw parserError + return Number(value) + }, + } + const cursor = client.query(new Cursor('SELECT value FROM generate_series(1, 6) value', [], { types, rowMode })) + const emittedErrors = [] + let endCount = 0 + cursor.on('error', (error) => emittedErrors.push(error)) + cursor.on('end', () => endCount++) + + assert.deepStrictEqual(await cursor.read(1), rowMode === 'array' ? [[1]] : [{ value: 1 }]) + + let callbackCount = 0 + const failedRead = new Promise((resolve, reject) => { + cursor.read(batchSize, (error) => { + callbackCount++ + error ? reject(error) : resolve() + }) + }) + const queuedRead = cursor.read(1) + const results = await Promise.allSettled([failedRead, queuedRead]) + for (const result of results) { + assert.strictEqual(result.status, 'rejected') + assert.strictEqual(result.reason, parserError) + } + + await cursor.close() + await assert.rejects(cursor.read(1), (error) => error === parserError) + const result = await client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors') + assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }]) + assert.strictEqual(callbackCount, 1) + assert.deepStrictEqual(emittedErrors, [parserError]) + assert.strictEqual(endCount, 0) + assert.deepStrictEqual(parsedValues, ['1', '2', '3']) + }) + } + } +}) diff --git a/packages/pg-query-stream/test/type-parser-errors.ts b/packages/pg-query-stream/test/type-parser-errors.ts new file mode 100644 index 000000000..a24bd9ffd --- /dev/null +++ b/packages/pg-query-stream/test/type-parser-errors.ts @@ -0,0 +1,80 @@ +import assert from 'assert' +import pg from 'pg' +import QueryStream from '../src' + +describe('type parser errors', () => { + it('closes a flowing stream once and allows the queued query to finish', async () => { + const client = new pg.Client() + await client.connect() + try { + const parserError = new Error('cannot decode value') + const stream = new QueryStream('SELECT value FROM generate_series(1, 6) value', [], { + batchSize: 2, + types: { + getTypeParser: () => (value) => { + if (value === '3') throw parserError + return Number(value) + }, + }, + }) + const emittedErrors = [] + const callbackErrors = [] + let closeCount = 0 + stream.on('data', (row) => assert(row.value < 3)) + stream.on('error', (error) => emittedErrors.push(error)) + stream.on('close', () => closeCount++) + const closed = new Promise((resolve) => stream.once('close', resolve)) + client.query(stream, (error) => callbackErrors.push(error)) + const queued = client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors') + const [result] = await Promise.all([queued, closed]) + assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }]) + assert.deepStrictEqual(callbackErrors, [parserError]) + assert.deepStrictEqual(emittedErrors, [parserError]) + assert.strictEqual(closeCount, 1) + } finally { + await client.end() + } + }) + + for (const batchSize of [2, 20]) { + it(`rejects iteration and releases the portal with batchSize=${batchSize}`, async () => { + const client = new pg.Client() + await client.connect() + try { + const parserError = new Error('cannot decode value') + const stream = new QueryStream('SELECT value FROM generate_series(1, 6) value', [], { + batchSize, + types: { + getTypeParser: () => (value) => { + if (value === '3') throw parserError + return Number(value) + }, + }, + }) + const callbackErrors = [] + const emittedErrors = [] + let closeCount = 0 + stream.on('error', (error) => emittedErrors.push(error)) + stream.on('close', () => closeCount++) + client.query(stream, (error) => callbackErrors.push(error)) + + await assert.rejects( + async () => { + for await (const row of stream) { + assert(row.value < 3) + } + }, + (error) => error === parserError + ) + + const result = await client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors') + assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }]) + assert.deepStrictEqual(callbackErrors, [parserError]) + assert.deepStrictEqual(emittedErrors, [parserError]) + assert.strictEqual(closeCount, 1) + } finally { + await client.end() + } + }) + } +}) From fc3689bfbb549fb45cc0e4b293b1d2fad3537ed5 Mon Sep 17 00:00:00 2001 From: Ibrahim Hafez Date: Thu, 29 Oct 2026 15:59:18 +0300 Subject: [PATCH 2/3] test(pg-query-stream): use the typed stream callback property --- packages/pg-query-stream/test/type-parser-errors.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/pg-query-stream/test/type-parser-errors.ts b/packages/pg-query-stream/test/type-parser-errors.ts index a24bd9ffd..e61410c0b 100644 --- a/packages/pg-query-stream/test/type-parser-errors.ts +++ b/packages/pg-query-stream/test/type-parser-errors.ts @@ -24,7 +24,8 @@ describe('type parser errors', () => { stream.on('error', (error) => emittedErrors.push(error)) stream.on('close', () => closeCount++) const closed = new Promise((resolve) => stream.once('close', resolve)) - client.query(stream, (error) => callbackErrors.push(error)) + stream.callback = (error) => callbackErrors.push(error) + client.query(stream) const queued = client.query('SELECT 42 AS answer, (SELECT count(*)::int FROM pg_cursors) AS cursors') const [result] = await Promise.all([queued, closed]) assert.deepStrictEqual(result.rows, [{ answer: 42, cursors: 0 }]) @@ -56,7 +57,8 @@ describe('type parser errors', () => { let closeCount = 0 stream.on('error', (error) => emittedErrors.push(error)) stream.on('close', () => closeCount++) - client.query(stream, (error) => callbackErrors.push(error)) + stream.callback = (error) => callbackErrors.push(error) + client.query(stream) await assert.rejects( async () => { From ca68aece3f1857bc265be033170654a7e4e081a6 Mon Sep 17 00:00:00 2001 From: Ibrahim Hafez Date: Thu, 29 Oct 2026 16:34:20 +0300 Subject: [PATCH 3/3] fix(pg-cursor): report timeout errors only once --- packages/pg-cursor/index.js | 3 ++ packages/pg-cursor/test/error-handling.js | 37 +++++++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/packages/pg-cursor/index.js b/packages/pg-cursor/index.js index e77a64b48..2032cb328 100644 --- a/packages/pg-cursor/index.js +++ b/packages/pg-cursor/index.js @@ -194,6 +194,9 @@ class Cursor extends EventEmitter { } _reportError(msg) { + // A timeout can report the error before ReadyForQuery. + if (this._errorReported) return + this._errorReported = true this._error = msg // satisfy any waiting callback if (this._cb) { diff --git a/packages/pg-cursor/test/error-handling.js b/packages/pg-cursor/test/error-handling.js index 9234870d5..425d9c67b 100644 --- a/packages/pg-cursor/test/error-handling.js +++ b/packages/pg-cursor/test/error-handling.js @@ -51,6 +51,43 @@ describe('error handling', function () { client.end() }) + + it('reports a read timeout once after the server finishes the query', async () => { + const client = new pg.Client() + await client.connect() + try { + await client.query('BEGIN') + const cursor = new Cursor('SELECT 1 AS value') + const queryTimeout = 20 + cursor.query_timeout = queryTimeout + const errors = [] + const callbackErrors = [] + cursor.on('error', (error) => errors.push(error)) + + // Hold server messages so the timeout precedes protocol completion. + client.connection.stream.pause() + client.query(cursor) + const timeoutError = await new Promise((resolve) => { + cursor.read(1, (error) => { + callbackErrors.push(error) + resolve(error) + }) + }) + assert.strictEqual(timeoutError.message, 'Query read timeout') + const completed = new Promise((resolve) => client.connection.once('readyForQuery', resolve)) + client.connection.stream.resume() + await completed + + const result = await client.query('SELECT 42 AS answer') + assert.deepStrictEqual(result.rows, [{ answer: 42 }]) + assert.deepStrictEqual(callbackErrors, [timeoutError]) + assert.deepStrictEqual(errors, [timeoutError]) + await client.query('ROLLBACK') + } finally { + client.connection.stream.resume() + await client.end() + } + }) }) describe('read callback does not fire sync', () => {