fix(query): send Sync after an error in a rows-limited query

packages/pg/lib/query.js

@@ -30,6 +30,7 @@ class Query extends EventEmitter {
     // potential for multiple results
     this._results = this._result
     this._canceledDueToError = false
+    this._hasSentSync = false
   }
 
   requiresPreparation() {
@@ -104,23 +105,20 @@ 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._syncRows(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._syncRows(connection)
   }
 
   handleError(err, connection) {
     // need to sync after error during a prepared statement
+    this._syncRows(connection)
     if (this._canceledDueToError) {
       err = this._canceledDueToError
       this._canceledDueToError = false
@@ -149,6 +147,13 @@ class Query extends EventEmitter {
     this.emit('end', this._results)
   }
 
+  _syncRows(connection) {
+    if (this.rows && !this._hasSentSync) {
+      this._hasSentSync = true
+      connection.sync()
+    }
+  }
+
   submit(connection) {
     if (typeof this.text !== 'string' && typeof this.name !== 'string') {
       return new Error('A query must have either text or a name. Supplying neither is unsupported.')
@@ -231,6 +236,7 @@ class Query extends EventEmitter {
       // we should close parse to avoid leaking connections
       connection.close({ type: 'S', name: this.name })
       connection.sync()
+      this._hasSentSync = true
 
       this.handleError(err, connection)
       return

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

@@ -0,0 +1,38 @@
+'use strict'
+const helper = require('./../test-helper')
+const assert = require('assert')
+
+const suite = new helper.Suite()
+
+suite.test('recovers after an error in a rows-limited query', async function () {
+  const client = new helper.pg.Client()
+
+  await client.connect()
+
+  try {
+    const error = await client
+      .query({
+        text: 'select 10 / (5 - i) as v from generate_series(1, 7) g(i)',
+        rows: 2,
+      })
+      .catch((err) => err)
+
+    assert.strictEqual(error.code, '22012')
+
+    const followUp = client.query('select 1').then(
+      () => 'follow-up resolved',
+      (err) => `follow-up rejected: ${err.message}`
+    )
+    let timeout
+    const hung = new Promise((resolve) => {
+      timeout = setTimeout(resolve, 4000, 'follow-up hung')
+    })
+
+    const result = await Promise.race([followUp, hung])
+    clearTimeout(timeout)
+
+    assert.strictEqual(result, 'follow-up resolved')
+  } finally {
+    await client.end()
+  }
+})

packages/pg/test/unit/client/throw-in-bind-tests.js

@@ -69,6 +69,23 @@ suite.test('emits error event when bind throws (no callback)', function (done) {
   client.query(query)
 })
 
+suite.test('calls sync once when bind throws for rows-limited query', function (done) {
+  const { client, con, calls } = setupClient()
+  con.emit('readyForQuery')
+  client.query(
+    new Query({
+      text: 'select $1',
+      values: ['x'],
+      rows: 1,
+      callback: function (err) {
+        assert.equal(err, bindError)
+        assert.equal(calls.sync, 1, 'sync should be called once')
+        done()
+      },
+    })
+  )
+})
+
 suite.test('send close when bind throws', function (done) {
   const { client, con, calls } = setupClient()
   con.emit('readyForQuery')