fix(pg): keep the connection usable after an error in a rows-limited query

packages/pg/lib/query.js

@@ -30,6 +30,9 @@ class Query extends EventEmitter {
     // potential for multiple results
     this._results = this._result
     this._canceledDueToError = false
+    // set when _getRows pipelines Flush instead of Sync, cleared by whichever
+    // of handleCommandComplete / handleEmptyQuery / handleError sends that Sync
+    this._syncOwed = false
   }
 
   requiresPreparation() {
@@ -104,23 +107,25 @@ class Query extends EventEmitter {
     this._result.addCommandComplete(msg)
     // need to sync after each command complete of a prepared statement
     // if we were using a row count which results in multiple calls to _getRows
-    if (this.rows) {
-      connection.sync()
-    }
+    this._sendOwedSync(connection)
   }
 
   // if a named prepared statement is created with empty query text
   // the backend will send an emptyQuery message but *not* a command complete message
   // since we pipeline sync immediately after execute we don't need to do anything here
   // unless we have rows specified, in which case we did not pipeline the initial sync call
   handleEmptyQuery(connection) {
-    if (this.rows) {
-      connection.sync()
-    }
+    this._sendOwedSync(connection)
   }
 
   handleError(err, connection) {
-    // need to sync after error during a prepared statement
+    // after an ErrorResponse the backend discards messages until it sees a Sync
+    // and withholds ReadyForQuery. In rows-mode no Sync was pipelined (_getRows
+    // flushed instead), so send the one we owe or the client never leaves the
+    // 'awaiting ReadyForQuery' state and every later query on the connection
+    // queues forever.
+    // http://developer.postgresql.org/pgdocs/postgres/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY
+    this._sendOwedSync(connection)
     if (this._canceledDueToError) {
       err = this._canceledDueToError
       this._canceledDueToError = false
@@ -200,11 +205,28 @@ class Query extends EventEmitter {
     if (!rows) {
       connection.sync()
     } else {
-      // otherwise flush the call out to read more rows
+      // otherwise flush the call out to read more rows; the backend stays in
+      // the middle of an extended-protocol sequence until CommandComplete (or
+      // an ErrorResponse / EmptyQuery — see _sendOwedSync) sends the Sync that
+      // terminates it
+      this._syncOwed = true
       connection.flush()
     }
   }
 
+  // send the terminating Sync for a rows-limited portal, exactly once. In
+  // rows-mode the Sync is not pipelined with Execute, so whoever observes the
+  // end of the extended-protocol sequence owes the backend one Sync before it
+  // will answer ReadyForQuery. An ErrorResponse can follow a CommandComplete
+  // (e.g. a timed-out query on some backends), which is why the flag guards
+  // against a second Sync — and a second ReadyForQuery — being sent.
+  _sendOwedSync(connection) {
+    if (this._syncOwed) {
+      this._syncOwed = false
+      connection.sync()
+    }
+  }
+
   // http://developer.postgresql.org/pgdocs/postgres/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY
   prepare(connection) {
     // TODO refactor this poor encapsulation

packages/pg/test/integration/client/query-error-handling-prepared-statement-tests.js

@@ -124,3 +124,133 @@ suite.test('query killed during query execution of prepared statement', function
     })
   )
 })
+
+// When a query runs with the `rows` option, Execute is pipelined with Flush
+// instead of Sync, so an ErrorResponse leaves the backend in ignore-till-sync:
+// it discards every message until a Sync arrives and withholds ReadyForQuery.
+// The client used to never send that Sync, so the query rejected correctly but
+// the connection was wedged — every later query queued forever.
+// https://github.com/brianc/node-postgres/issues/3707
+suite.test('rows-limited query that errors mid-portal stays usable', function (done) {
+  if (helper.args.native) {
+    return done()
+  }
+  const client = new Client(helper.args)
+  client.connect(
+    assert.success(function () {
+      // one ReadyForQuery per completed extended-protocol sequence — the
+      // errored query and the follow-up, no more (a stray second Sync would
+      // produce a third and desync the queue)
+      let readyForQueryCount = 0
+      client.connection.on('readyForQuery', function () {
+        readyForQueryCount++
+      })
+      // divides by zero on row 5, fetched in pages of 2: the error arrives
+      // between PortalSuspended round trips, with the portal still open
+      client.query(
+        { text: 'select 10 / (5 - i) as v from generate_series(1, 7) g(i)', rows: 2 },
+        assert.calls(function (err) {
+          assert.equal(err.message, 'division by zero')
+          client.query(
+            'select 41 + 1 as v',
+            assert.calls(function (err, res) {
+              assert.ifError(err)
+              assert.equal(res.rows[0].v, 42)
+              client.end(
+                assert.success(function () {
+                  assert.equal(readyForQueryCount, 2)
+                  done()
+                })
+              )
+            })
+          )
+        })
+      )
+    })
+  )
+})
+
+suite.test('rows-limited query that errors before any row stays usable', function (done) {
+  if (helper.args.native) {
+    return done()
+  }
+  const client = new Client(helper.args)
+  client.connect(
+    assert.success(function () {
+      let readyForQueryCount = 0
+      client.connection.on('readyForQuery', function () {
+        readyForQueryCount++
+      })
+      // rejected at Parse, before the portal is ever created
+      client.query(
+        { text: 'select 1; select 2', rows: 2 },
+        assert.calls(function (err) {
+          assert.equal(err.message, 'cannot insert multiple commands into a prepared statement')
+          client.query(
+            'select 41 + 1 as v',
+            assert.calls(function (err, res) {
+              assert.ifError(err)
+              assert.equal(res.rows[0].v, 42)
+              client.end(
+                assert.success(function () {
+                  assert.equal(readyForQueryCount, 2)
+                  done()
+                })
+              )
+            })
+          )
+        })
+      )
+    })
+  )
+})
+
+// A row that throws while being parsed cancels the query client-side: the
+// portal still ends with CommandComplete, which sends the owed Sync, and the
+// error is only surfaced on ReadyForQuery. handleError must not send a second
+// Sync for the same sequence — two ReadyForQuery frames would let the next
+// query settle on the first one, mid-flight.
+suite.test('rows-limited query that fails while parsing a row syncs once', function (done) {
+  if (helper.args.native) {
+    return done()
+  }
+  const client = new Client(helper.args)
+  client.connect(
+    assert.success(function () {
+      let readyForQueryCount = 0
+      client.connection.on('readyForQuery', function () {
+        readyForQueryCount++
+      })
+      const query = {
+        text: 'select 1 as v',
+        rows: 1,
+        types: {
+          getTypeParser: function () {
+            return function () {
+              throw new Error('boom')
+            }
+          },
+        },
+      }
+      client.query(
+        query,
+        assert.calls(function (err) {
+          assert.equal(err.message, 'boom')
+          client.query(
+            'select 41 + 1 as v',
+            assert.calls(function (err, res) {
+              assert.ifError(err)
+              assert.equal(res.rows[0].v, 42)
+              client.end(
+                assert.success(function () {
+                  assert.equal(readyForQueryCount, 2)
+                  done()
+                })
+              )
+            })
+          )
+        })
+      )
+    })
+  )
+})