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 }) + }) + } +})