fix: sync connection after error during rows-limited query

packages/pg/lib/query.js

@@ -124,6 +124,24 @@ class Query extends EventEmitter {
     if (this._canceledDueToError) {
       err = this._canceledDueToError
       this._canceledDueToError = false
+    } else if (this.rows) {
+      // An ErrorResponse from the backend terminates the current Execute/Sync
+      // pipeline without sending ReadyForQuery. For a rows-limited (portal
+      // suspension) query we pipeline Flush instead of Sync between page
+      // requests (see _getRows), so unlike a plain prepared statement, the
+      // backend has not already been sent a Sync for this cycle. Per the
+      // extended query protocol, the backend discards messages until it
+      // receives Sync, so without this call the connection would wedge:
+      // ReadyForQuery never arrives and every later query queues forever.
+      //
+      // This branch only runs for a "real" ErrorResponse reaching handleError
+      // directly - not for the _canceledDueToError path above, which is
+      // replayed from handleReadyForQuery after ReadyForQuery has already
+      // been received for this query, i.e. after a Sync has already round
+      // tripped. Syncing again there would be a protocol-correct but
+      // redundant second Sync/ReadyForQuery pair, so it's intentionally
+      // excluded here.
+      connection.sync()
     }
     // if callback supplied do not emit error event as uncaught error
     // events will bubble up to node process

packages/pg/test/integration/gh-issues/3707-tests.js

@@ -0,0 +1,74 @@
+'use strict'
+const helper = require('../test-helper')
+const assert = require('assert')
+
+// An ErrorResponse during a `rows`-limited (portal suspension) query used to
+// permanently wedge the connection: `_getRows` pipelines Flush rather than
+// Sync between page requests, and `handleError` never sent a Sync to make up
+// for it. Per the extended query protocol, the backend discards all
+// subsequent messages until it receives a Sync, so every later query on the
+// same connection would queue forever.
+// https://github.com/brianc/node-postgres/issues/3707
+const suite = new helper.Suite()
+
+suite.test('recovers after an error during a rows-limited query', async () => {
+  const client = new helper.Client()
+  await client.connect()
+
+  let caught = null
+  try {
+    // division by zero fails partway through the result set, after at
+    // least one page (rows: 2) has already been returned by the portal.
+    await client.query({
+      text: 'SELECT 1 / (x - 5) AS v FROM generate_series(1, 10) AS x',
+      rows: 2,
+    })
+  } catch (err) {
+    caught = err
+  }
+
+  assert(caught, 'expected the query to reject with a division-by-zero error')
+  assert.equal(caught.code, '22012') // division_by_zero
+
+  // Before the fix this would hang until the test's 5s timeout fired,
+  // because the connection never received ReadyForQuery.
+  const res = await client.query('SELECT 1 AS ok')
+  assert.equal(res.rows[0].ok, 1)
+
+  await client.end()
+})
+
+suite.test('does not double-sync when a row fails to parse after an error-free portal suspension', async () => {
+  const client = new helper.Client()
+  await client.connect()
+
+  // A parse failure in handleDataRow sets _canceledDueToError and is
+  // replayed from handleReadyForQuery, i.e. after ReadyForQuery has already
+  // been received for this query. handleError must not sync a second time
+  // on that path, or it would desync the client from the server's reply
+  // stream for the next query.
+  const throwingTypes = {
+    getTypeParser: () => () => {
+      throw new Error('forced row parse failure')
+    },
+  }
+
+  let caught = null
+  try {
+    await client.query({
+      text: 'SELECT x FROM generate_series(1, 10) AS x',
+      rows: 2,
+      types: throwingTypes,
+    })
+  } catch (err) {
+    caught = err
+  }
+
+  assert(caught, 'expected the query to reject with the forced parse error')
+  assert.equal(caught.message, 'forced row parse failure')
+
+  const res = await client.query('SELECT 1 AS ok')
+  assert.equal(res.rows[0].ok, 1)
+
+  await client.end()
+})