fix(pg): settle a query once when a value fails to serialize

packages/pg/lib/query.js

@@ -239,7 +239,10 @@ class Query extends EventEmitter {
       connection.close({ type: 'S', name: this.name })
       connection.sync()
 
-      this.handleError(err, connection)
+      // Settle once the server answers that Sync with ReadyForQuery, the same way a
+      // failing type parser does. Reporting the error here would leave this query as
+      // the active one, and the ReadyForQuery would then also fire the success path.
+      this._canceledDueToError = err
       return
     }
 

packages/pg/test/integration/client/bind-error-settles-once-tests.js

@@ -0,0 +1,62 @@
+'use strict'
+const helper = require('./test-helper')
+const assert = require('assert')
+const { Query } = helper.pg
+
+const suite = new helper.Suite()
+
+const badValue = {
+  toPostgres() {
+    throw new Error('bad value')
+  },
+}
+
+const settle = (ms) => new Promise((resolve) => setTimeout(resolve, ms))
+
+for (const pipeline of [false, true]) {
+  const label = pipeline ? ' (pipeline)' : ''
+
+  suite.test(
+    'callback runs once when a value fails to serialize' + label,
+    !helper.args.native &&
+      async function () {
+        const client = helper.client(undefined, { pipeline })
+        try {
+          const results = []
+          client.query('SELECT $1::text AS v', [badValue], (err, res) => results.push(err ? err.message : res.rows))
+          if (!pipeline) {
+            await settle(50)
+          }
+          const next = await client.query('SELECT 42 AS answer')
+          await settle(50)
+          assert.deepStrictEqual(results, ['bad value'])
+          assert.strictEqual(next.rows[0].answer, 42)
+        } finally {
+          await client.end()
+        }
+      }
+  )
+
+  suite.test(
+    'no end event after the error when a value fails to serialize' + label,
+    !helper.args.native &&
+      async function () {
+        const client = helper.client(undefined, { pipeline })
+        try {
+          const events = []
+          const query = new Query('SELECT $1::text AS v', [badValue])
+          query.on('error', (err) => events.push(err.message))
+          query.on('end', () => events.push('end'))
+          client.query(query)
+          if (!pipeline) {
+            await settle(50)
+          }
+          await client.query('SELECT 1')
+          await settle(50)
+          assert.deepStrictEqual(events, ['bad value'])
+        } finally {
+          await client.end()
+        }
+      }
+  )
+}

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

@@ -31,6 +31,8 @@ const setupClient = function () {
   }
   con.sync = function () {
     calls.sync++
+    // the server answers the Sync with ReadyForQuery
+    setImmediate(() => con.emit('readyForQuery'))
   }
 
   return { client, con, calls }
@@ -84,3 +86,69 @@ suite.test('send close when bind throws', function (done) {
     })
   )
 })
+
+suite.test('calls callback exactly once when bind throws', function (done) {
+  const { client, con } = setupClient()
+  con.emit('readyForQuery')
+  const results = []
+  client.query(
+    new Query({
+      text: 'select $1',
+      values: ['x'],
+      callback: function (err, res) {
+        results.push(err || res)
+      },
+    })
+  )
+  // the ReadyForQuery that answers the Sync must not call back a second time
+  setTimeout(() => {
+    assert.deepEqual(results, [bindError])
+    done()
+  }, 20)
+})
+
+suite.test('does not emit end after error when bind throws', function (done) {
+  const { client, con } = setupClient()
+  con.emit('readyForQuery')
+  const events = []
+  const query = new Query({
+    text: 'select $1',
+    values: ['x'],
+  })
+  query.on('error', (err) => events.push(err))
+  query.on('end', () => events.push('end'))
+  client.query(query)
+  setTimeout(() => {
+    assert.deepEqual(events, [bindError])
+    done()
+  }, 20)
+})
+
+suite.test('runs the next query after bind throws', function (done) {
+  const { client, con } = setupClient()
+  con.emit('readyForQuery')
+  const order = []
+  client.query(
+    new Query({
+      text: 'select $1',
+      values: ['x'],
+      callback: function (err) {
+        order.push(err)
+      },
+    })
+  )
+  client.query(
+    new Query({
+      text: 'select 1',
+      callback: function (err) {
+        order.push(err || 'ok')
+      },
+    })
+  )
+  // ReadyForQuery for the second (simple) query
+  setImmediate(() => setImmediate(() => con.emit('readyForQuery')))
+  setTimeout(() => {
+    assert.deepEqual(order, [bindError, 'ok'])
+    done()
+  }, 20)
+})