From d99e946b9c0a5338cb2379fcb276194ad56358bb Mon Sep 17 00:00:00 2001 From: Matthijs Wolting Date: Mon, 14 Sep 2026 17:45:20 +0200 Subject: [PATCH] fix(pg-cursor): preserve cursor state after late responses When a cursor closes before its first backend response arrives, late responses can reset its state and send a second Close/Sync pair. The extra ReadyForQuery can advance the client's query queue too early and corrupt the next query's result. Preserve closed and failed states when handling late descriptions, and keep closed cursors closed when delivering rows. Fields and the active read callback still get delivered. Add focused late-message tests and a PostgreSQL regression that closes before responses arrive, checks for a single Sync, and reuses the connection. Settlement of reads left in the cursor's queue remains outside this fix. --- packages/pg-cursor/index.js | 16 +++- packages/pg-cursor/test/close.js | 32 +++++++ packages/pg-cursor/test/late-messages.js | 111 +++++++++++++++++++++++ 3 files changed, 154 insertions(+), 5 deletions(-) create mode 100644 packages/pg-cursor/test/late-messages.js diff --git a/packages/pg-cursor/index.js b/packages/pg-cursor/index.js index f1553cc9c..5d566882b 100644 --- a/packages/pg-cursor/index.js +++ b/packages/pg-cursor/index.js @@ -27,8 +27,10 @@ class Cursor extends EventEmitter { } _ifNoData() { - this.state = 'idle' - this._shiftQueue() + if (this.state !== 'done' && this.state !== 'error') { + this.state = 'idle' + this._shiftQueue() + } if (this.connection) { this.connection.removeListener('rowDescription', this._rowDescription) } @@ -106,8 +108,10 @@ class Cursor extends EventEmitter { handleRowDescription(msg) { this._result.addFields(msg.fields) - this.state = 'idle' - this._shiftQueue() + if (this.state !== 'done' && this.state !== 'error') { + this.state = 'idle' + this._shiftQueue() + } } handleDataRow(msg) { @@ -117,7 +121,9 @@ class Cursor extends EventEmitter { } _sendRows() { - this.state = 'idle' + if (this.state !== 'done') { + this.state = 'idle' + } setImmediate(() => { const cb = this._cb // remove callback before calling it diff --git a/packages/pg-cursor/test/close.js b/packages/pg-cursor/test/close.js index 4b4c913a3..17c584204 100644 --- a/packages/pg-cursor/test/close.js +++ b/packages/pg-cursor/test/close.js @@ -62,4 +62,36 @@ describe('close', function () { const cursor = new Cursor(text) cursor.close(done) }) + + it('keeps the client usable after closing before the first response', async function () { + // Let the connect callback return before submitting the cursor. + await Promise.resolve() + const client = this.client + const cursor = new Cursor(text) + const connection = client.connection + const sync = connection.sync + let syncCount = 0 + connection.sync = function () { + syncCount++ + return sync.apply(this, arguments) + } + + connection.stream.pause() + try { + client.query(cursor) + assert.strictEqual(cursor.connection, connection) + // Exhaust the portal so CommandComplete arrives after close. + const read = cursor.read(100) + const closed = cursor.close() + connection.stream.resume() + + await Promise.all([read, closed]) + assert.strictEqual(syncCount, 1) + const result = await client.query('SELECT 1 AS value') + assert.deepStrictEqual(result.rows, [{ value: 1 }]) + } finally { + connection.stream.resume() + connection.sync = sync + } + }) }) diff --git a/packages/pg-cursor/test/late-messages.js b/packages/pg-cursor/test/late-messages.js new file mode 100644 index 000000000..b459a4352 --- /dev/null +++ b/packages/pg-cursor/test/late-messages.js @@ -0,0 +1,111 @@ +const assert = require('assert') +const EventEmitter = require('events') +const Cursor = require('../') + +class TestConnection extends EventEmitter { + constructor() { + super() + this.calls = { + close: 0, + execute: 0, + sync: 0, + } + } + + parse() {} + bind() {} + describe() {} + flush() {} + + execute() { + this.calls.execute++ + } + + close() { + this.calls.close++ + } + + sync() { + this.calls.sync++ + } +} + +const submitAndCloseWithQueuedRead = () => { + const cursor = new Cursor('select 1') + const connection = new TestConnection() + + cursor.read(1, () => {}) + cursor.submit(connection) + cursor.close(() => {}) + + assert.strictEqual(cursor.state, 'done') + assert.deepStrictEqual(connection.calls, { close: 1, execute: 0, sync: 1 }) + + return { connection, cursor } +} + +describe('messages received after close', function () { + it('does not execute a queued read after no data', function () { + const { connection, cursor } = submitAndCloseWithQueuedRead() + + connection.emit('noData') + cursor.handleCommandComplete({ text: 'SELECT 0' }) + + assert.strictEqual(cursor.state, 'done') + assert.deepStrictEqual(connection.calls, { close: 1, execute: 0, sync: 1 }) + }) + + it('preserves fields without executing a queued read after row description', function () { + const { connection, cursor } = submitAndCloseWithQueuedRead() + const fields = [{ name: 'value', dataTypeID: 23 }] + + cursor.handleRowDescription({ fields }) + cursor.handleCommandComplete({ text: 'SELECT 0' }) + + assert.strictEqual(cursor.state, 'done') + assert.strictEqual(cursor._result.fields, fields) + assert.deepStrictEqual(connection.calls, { close: 1, execute: 0, sync: 1 }) + }) + + it('delivers the active callback without reopening after portal suspended', function (done) { + const cursor = new Cursor('select 1') + const connection = new TestConnection() + cursor.submit(connection) + cursor.read(1, (err, rows, result) => { + assert.ifError(err) + assert.deepStrictEqual(rows, []) + assert.strictEqual(result.rows, rows) + assert.strictEqual(cursor.state, 'done') + assert.deepStrictEqual(connection.calls, { close: 1, execute: 1, sync: 1 }) + done() + }) + + cursor.close(() => {}) + cursor.handlePortalSuspended() + + assert.strictEqual(cursor.state, 'done') + }) +}) + +describe('messages received after error', function () { + for (const message of ['noData', 'rowDescription']) { + it(`preserves the read error after ${message}`, async function () { + const cursor = new Cursor('select 1') + const connection = new TestConnection() + const error = new Error('query failed') + cursor.submit(connection) + cursor.handleError(error) + + if (message === 'noData') { + // handleError removes the listener; exercise the defensive guard directly. + cursor._ifNoData() + } else { + cursor.handleRowDescription({ fields: [] }) + } + + assert.strictEqual(cursor.state, 'error') + await assert.rejects(cursor.read(1), (err) => err === error) + assert.deepStrictEqual(connection.calls, { close: 0, execute: 0, sync: 1 }) + }) + } +})