diff --git a/packages/pg-native/index.js b/packages/pg-native/index.js index 7fcc26303..527d5dc06 100644 --- a/packages/pg-native/index.js +++ b/packages/pg-native/index.js @@ -386,8 +386,12 @@ Client.prototype._readPipelineResults = function (queries, cb) { const processResults = function () { if (!pq.consumeInput()) { - pq.exitPipelineMode() - return cb(new Error(pq.errorMessage() || 'Failed to consume input')) + // read the message before anything else touches the connection: libpq appends to a single + // error buffer, and exiting pipeline mode on a connection that is still busy adds its own + // "cannot exit pipeline mode while busy" to the end of the reason the caller actually wants. + // The connection is finished either way, so there is nothing to exit cleanly for. + const message = pq.errorMessage() + return cb(new Error(message || 'Failed to consume input')) } while (!pq.isBusy()) { @@ -395,8 +399,10 @@ Client.prototype._readPipelineResults = function (queries, cb) { // null between result groups in pipeline — try again if (pq.isBusy()) return // more data needed if (!pq.getResult()) { - // truly no more results — should not happen before all syncs - break + // libpq has no result left and is not waiting for one, yet we have not seen a sync for + // every query. Nothing further is owed on this connection, so breaking out would return + // without ever calling cb and strand the caller. Fail the batch instead. + return cb(new Error(pq.errorMessage() || 'Connection ended before the pipeline completed')) } } @@ -436,7 +442,13 @@ Client.prototype._readPipelineResults = function (queries, cb) { } if (status === 'PGRES_PIPELINE_ABORTED') { - // Query skipped due to previous error in same sync group + // The server refused to run this one: an earlier query in the same sync group failed and + // the pipeline is aborted until the next sync. Falling through left currentError and + // currentResult null, so the sync branch below handed back {err: null, rows: []} and a + // statement that never ran looked like one that matched no rows. + if (!currentError) { + currentError = new Error('Query was not executed: an earlier query in the same pipeline failed') + } continue } @@ -465,6 +477,11 @@ Client.prototype._readPipelineResults = function (queries, cb) { } pq.on('readable', onReadable) pq.startReader() + // startReader() has to be recorded, or _stopReading() short-circuits on the flag and leaves the + // poll watcher running after the batch is over. A later finish() then closes the handle with the + // watcher still armed, and the next startReader() on it aborts the process with + // "uv_poll_start: Assertion `!uv__is_closing(handle)' failed". + this._reading = true // Try an initial read in case data is already available processResults() diff --git a/packages/pg-native/test/pipeline-connection-loss.js b/packages/pg-native/test/pipeline-connection-loss.js new file mode 100644 index 000000000..eb5e779a5 --- /dev/null +++ b/packages/pg-native/test/pipeline-connection-loss.js @@ -0,0 +1,76 @@ +const net = require('net') +const assert = require('assert') +const Client = require('../') + +describe('pipeline reader', function () { + let proxy + let proxyPort + let clientSockets + + beforeEach(function (done) { + clientSockets = [] + proxy = net.createServer(function (client) { + const upstream = net.connect(Number(process.env.PGPORT || 5432), process.env.PGHOST || 'localhost') + clientSockets.push(client) + client.pipe(upstream) + upstream.pipe(client) + client.on('error', function () {}) + upstream.on('error', function () {}) + }) + proxy.listen(0, '127.0.0.1', function () { + proxyPort = proxy.address().port + done() + }) + }) + + afterEach(function (done) { + proxy.close(function () { + done() + }) + }) + + // the batch starts the reader, so it has to stop it too. It used to leave the poll watcher armed, + // and a later finish() then closed the handle under it. + it('stops the reader it started once the batch is done', function (done) { + const client = new Client() + client.connect(`host=127.0.0.1 port=${proxyPort}`, function (err) { + assert.ifError(err) + let stopped = 0 + const stopReader = client.pq.stopReader.bind(client.pq) + client.pq.stopReader = function () { + stopped++ + return stopReader() + } + client.pipeline([{ text: 'SELECT 1' }, { text: 'SELECT 2' }], function (err) { + assert.ifError(err) + assert(stopped > 0, 'the reader started for the batch was never stopped') + client.end() + done() + }) + }) + }) + + // exitPipelineMode() on a busy connection appends its own complaint to libpq's error buffer, so + // reading the message afterwards buried the reason the caller wanted. + it('reports why the connection went away', function (done) { + this.timeout(10000) + const client = new Client() + client.connect(`host=127.0.0.1 port=${proxyPort}`, function (err) { + assert.ifError(err) + client.pipeline([{ text: 'SELECT pg_sleep(10)' }], function (err) { + assert(err, 'a batch cut off mid flight must fail') + assert( + !/cannot exit pipeline mode/.test(err.message), + `error should say why the connection ended, got: ${err.message}` + ) + client.end() + done() + }) + setTimeout(function () { + clientSockets.forEach(function (socket) { + socket.end() + }) + }, 100) + }) + }) +}) diff --git a/packages/pg/lib/native/client.js b/packages/pg/lib/native/client.js index d305713d6..94f5a0890 100644 --- a/packages/pg/lib/native/client.js +++ b/packages/pg/lib/native/client.js @@ -351,13 +351,21 @@ Client.prototype._pulsePipelinedQueryQueue = function () { self._pipelineInFlight = false if (err) { - // Total pipeline failure — error all queries + // Total pipeline failure. Per-query errors arrive on results[i].err, so reaching here means + // the connection itself is gone: mark it unusable and say so, the way the JS client and the + // non-pipelined native path both do. Without this the client looks healthy after losing its + // backend, a Pool never discards it, and everything routed to it fails one query at a time + // for the life of the process. + self._connected = false + self._queryable = false for (let i = 0; i < nativeQueries.length; i++) { const q = nativeQueries[i] q.native = self.native q.handleError(err) } - self._pulsePipelinedQueryQueue() + self._errorAllQueries(err) + self.emit('error', err) + self.emit('end') return }