fix: fail pipelined queries when the connection dies instead of hanging

packages/pg-native/index.js

@@ -386,17 +386,23 @@ 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()) {
       if (!pq.getResult()) {
         // 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()

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

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
     }