fix: surface unhandled native result statuses on the query callback

packages/pg-native/index.js

@@ -188,23 +188,23 @@ Client.prototype._emitResult = function (pq) {
     case 'PGRES_TUPLES_OK':
     case 'PGRES_COMMAND_OK':
     case 'PGRES_EMPTY_QUERY':
+    case 'PGRES_COPY_OUT':
+    case 'PGRES_COPY_IN':
+    case 'PGRES_COPY_BOTH':
       {
         const result = this._consumeQueryResults(this.pq)
         this.emit('result', result)
       }
       break
 
-    case 'PGRES_COPY_OUT':
-    case 'PGRES_COPY_BOTH': {
-      break
-    }
-
     case 'PGRES_PIPELINE_SYNC':
     case 'PGRES_PIPELINE_ABORTED':
       break
 
     default:
-      this._readError('unrecognized command status: ' + status)
+      this._queryError = new Error(
+        this.pq.resultErrorMessage() || this.pq.errorMessage() || 'unrecognized command status: ' + status
+      )
       break
   }
   return status
@@ -233,8 +233,10 @@ Client.prototype._read = function () {
     const resultStatus = this._emitResult(this.pq)
 
     // if the command initiated copy mode we need to break out of the read loop
-    // so a substream can begin to read copy data
-    if (resultStatus === 'PGRES_COPY_BOTH' || resultStatus === 'PGRES_COPY_OUT') {
+    // so a substream can begin to read or write copy data. Without the
+    // PGRES_COPY_IN case the loop keeps polling PQgetResult while the server
+    // waits for copy data that only the substream can send.
+    if (resultStatus === 'PGRES_COPY_BOTH' || resultStatus === 'PGRES_COPY_OUT' || resultStatus === 'PGRES_COPY_IN') {
       break
     }
 

packages/pg-native/test/unhandled-result-status-integration.js

@@ -0,0 +1,79 @@
+'use strict'
+const assert = require('assert')
+const Client = require('../')
+const concat = require('concat-stream')
+
+// Integration test for the libpq result statuses that used to finish a
+// callback query without building a Result object, run against a real
+// Postgres backend. The unit tests cover _emitResult with stubs; these
+// tests make sure a real COPY round trip leaves the query callback with
+// a usable Result object.
+//
+// Note: the query callbacks below intentionally avoid throwing. They run
+// inside the libpq readable event, and an exception thrown there corrupts
+// the connection, so the assertions run once the copy stream has finished.
+//
+// Each test uses its own client: after a COPY TO STDOUT has been drained
+// through the copy stream, libpq still has the trailing result pending, so
+// the connection is not reusable for the next callback query (this matches
+// the behavior on master and is independent of this fix).
+describe('unhandled libpq result statuses (integration)', function () {
+  beforeEach(function (done) {
+    this.client = new Client()
+    this.client.connect(done)
+  })
+
+  afterEach(function () {
+    this.client.end()
+  })
+
+  it('gives the query callback a Result for COPY TO STDOUT', function (done) {
+    const client = this.client
+    client.query('COPY (SELECT 1 AS n) TO STDOUT', function (err, rows, result) {
+      // consume the copy data so the copy completes
+      const stream = client.getCopyStream()
+      stream.read()
+      stream.pipe(
+        concat(function (buf) {
+          assert.ifError(err)
+          // PGRES_COPY_OUT used to finish without building a Result,
+          // leaving the callback with an undefined result.
+          assert(result, 'expected a result object for PGRES_COPY_OUT')
+          assert.deepStrictEqual(rows, [])
+          assert.deepStrictEqual(result.rows, [])
+          assert.strictEqual(buf.toString('utf8'), '1\n')
+          done()
+        })
+      )
+    })
+  })
+
+  it('gives the query callback a Result for COPY FROM STDIN', function (done) {
+    const client = this.client
+    client.querySync('CREATE TEMP TABLE unhandled_status_copy (name text)')
+    client.query('COPY unhandled_status_copy FROM STDIN', function (err, rows, result) {
+      // feed the copy data and let the copy finish
+      const stream = client.getCopyStream()
+      stream.write(Buffer.from('brian\n', 'utf8'))
+      stream.end()
+      stream.once('finish', function () {
+        assert.ifError(err)
+        // PGRES_COPY_IN used to finish without building a Result,
+        // leaving the callback with an undefined result.
+        assert(result, 'expected a result object for PGRES_COPY_IN')
+        assert.deepStrictEqual(rows, [])
+        const countRows = client.querySync('SELECT count(*) AS count FROM unhandled_status_copy')
+        // count is a bigint, which the driver reports as a string
+        assert.strictEqual(countRows[0].count, '1')
+        // the copy substream consumed the trailing result, so the
+        // connection accepts another callback query afterwards
+        client.query('SELECT 41 AS n', function (err2, rows2, result2) {
+          assert.ifError(err2)
+          assert.strictEqual(result2.command, 'SELECT')
+          assert.strictEqual(rows2.length, 1)
+          done()
+        })
+      })
+    })
+  })
+})

packages/pg-native/test/unhandled-result-status.js

@@ -0,0 +1,82 @@
+const assert = require('assert')
+const EventEmitter = require('events').EventEmitter
+const types = require('pg-types')
+const Client = require('../')
+
+function stubClient(pq) {
+  const client = Object.create(Client.prototype)
+  EventEmitter.call(client)
+  client.pq = pq
+  client._types = types
+  client.arrayMode = false
+  client._resultCount = 0
+  client._queryError = undefined
+  client._results = undefined
+  client._rows = undefined
+  client.on('result', client._onResult.bind(client))
+  return client
+}
+
+function emptyResultPq(status, errorMessage) {
+  return {
+    resultStatus: () => status,
+    resultErrorMessage: () => errorMessage || '',
+    errorMessage: () => errorMessage || '',
+    cmdStatus: () => 'COPY 0',
+    cmdTuples: () => '0',
+    nfields: () => 0,
+    ntuples: () => 0,
+  }
+}
+
+describe('unhandled libpq result statuses', () => {
+  it('emits a Result for COPY_OUT so the query callback is not left with undefined', (done) => {
+    const client = stubClient(emptyResultPq('PGRES_COPY_OUT'))
+    client._queryCallback = (err, rows, results) => {
+      assert.ifError(err)
+      assert.deepEqual(rows, [])
+      assert(results)
+      assert.deepEqual(results.rows, [])
+      done()
+    }
+    client._emitResult(client.pq)
+    client._onReadyForQuery()
+  })
+
+  it('emits a Result for COPY_IN', (done) => {
+    const client = stubClient(emptyResultPq('PGRES_COPY_IN'))
+    client._queryCallback = (err, rows, results) => {
+      assert.ifError(err)
+      assert(results)
+      assert.deepEqual(results.rows, [])
+      done()
+    }
+    client._emitResult(client.pq)
+    client._onReadyForQuery()
+  })
+
+  it('passes unrecognized statuses to the query callback instead of the client error event', (done) => {
+    const client = stubClient(emptyResultPq('PGRES_WEIRD', 'server said no'))
+    client.on('error', () => {
+      done(new Error('should not emit error on the client'))
+    })
+    client._queryCallback = (err) => {
+      assert(err instanceof Error)
+      assert.match(err.message, /server said no/)
+      done()
+    }
+    client._emitResult(client.pq)
+    client._onReadyForQuery()
+  })
+
+  it('falls back to the status name when libpq has no error message', (done) => {
+    const client = stubClient(emptyResultPq('PGRES_WEIRD'))
+    client._queryCallback = (err) => {
+      assert(err instanceof Error)
+      assert.match(err.message, /unrecognized command status: PGRES_WEIRD/)
+      done()
+    }
+    client._emitResult(client.pq)
+    client._onReadyForQuery()
+  })
+})

packages/pg/lib/native/query.js

@@ -99,6 +99,12 @@ NativeQuery.prototype.submit = function (client) {
       return self.handleError(err)
     }
 
+    // COPY and other unhandled libpq statuses used to complete with no result
+    // object. Treat that as an error so callers never see success plus undefined.
+    if (results == null) {
+      return self.handleError(new Error('Native query completed without a result'))
+    }
+
     // emit row events for each row in the result
     if (self._emitRowEvents) {
       if (results.length > 1) {

packages/pg/test/unit/native-query-missing-result-tests.js

@@ -0,0 +1,48 @@
+'use strict'
+const helper = require('./test-helper')
+const assert = require('assert')
+const NativeQuery = require('../../lib/native/query')
+const suite = new helper.Suite()
+const test = suite.test.bind(suite)
+
+function submitWithNativeResult(nativeCbArgs, queryCb) {
+  const query = new NativeQuery({ text: 'SELECT 1', callback: queryCb })
+  const client = {
+    namedQueries: {},
+    native: {
+      arrayMode: false,
+      pq: { resultErrorFields: () => null },
+      query: function (text, valuesOrCb, maybeCb) {
+        const cb = typeof valuesOrCb === 'function' ? valuesOrCb : maybeCb
+        setImmediate(() => cb.apply(null, nativeCbArgs))
+      },
+    },
+  }
+  query.submit(client)
+}
+
+test('missing native result is an error, not success with undefined', (done) => {
+  submitWithNativeResult([null, [], undefined], (err, res) => {
+    assert(err instanceof Error)
+    assert.strictEqual(res, undefined)
+    assert.match(err.message, /without a result/)
+    done()
+  })
+})
+
+test('normal native result still succeeds', (done) => {
+  const result = { rows: [{ n: 1 }], fields: [], command: 'SELECT', rowCount: 1 }
+  submitWithNativeResult([null, result.rows, result], (err, res) => {
+    assert.ifError(err)
+    assert.strictEqual(res, result)
+    done()
+  })
+})
+
+test('native query error is still forwarded', (done) => {
+  const boom = new Error('boom')
+  submitWithNativeResult([boom], (err) => {
+    assert.strictEqual(err, boom)
+    done()
+  })
+})