fix(pg-cursor): preserve cursor state after late responses

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

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

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