diff --git a/packages/pg-native/index.js b/packages/pg-native/index.js index 451daeb7d..1814eef22 100644 --- a/packages/pg-native/index.js +++ b/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 } diff --git a/packages/pg-native/test/unhandled-result-status-integration.js b/packages/pg-native/test/unhandled-result-status-integration.js new file mode 100644 index 000000000..789e99d8b --- /dev/null +++ b/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() + }) + }) + }) + }) +}) diff --git a/packages/pg-native/test/unhandled-result-status.js b/packages/pg-native/test/unhandled-result-status.js new file mode 100644 index 000000000..910f89f86 --- /dev/null +++ b/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() + }) +}) diff --git a/packages/pg/lib/native/query.js b/packages/pg/lib/native/query.js index 8cb561979..105211f06 100644 --- a/packages/pg/lib/native/query.js +++ b/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) { diff --git a/packages/pg/test/unit/native-query-missing-result-tests.js b/packages/pg/test/unit/native-query-missing-result-tests.js new file mode 100644 index 000000000..320d2731d --- /dev/null +++ b/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() + }) +})