Fix pipelining activation race and add edge-case tests

Gate _sentQueryQueue activation on readyForQuery=true inside
_pulsePipelinedQueryQueue (and remove the redundant promotion block
from _handleReadyForQuery) to eliminate the microtask/macrotask race
where the next query could be activated as _activeQuery before the
error's ReadyForQuery arrived, causing that RFQ to be handled by the
wrong query.

Also adds the error-listener fix for the query_timeout integration test
so the expected stream-destroy doesn't leak as an unhandled 'error'.

Author: mcollina

packages/pg/lib/client.js

@@ -368,10 +368,6 @@ class Client extends EventEmitter {
     if (activeQuery) {
       activeQuery.handleReadyForQuery(this.connection)
     }
-    if (this.pipelining && this._sentQueryQueue.length > 0) {
-      this._activeQuery = this._sentQueryQueue.shift()
-      this.readyForQuery = false
-    }
     this._pulseQueryQueue()
   }
 
@@ -636,7 +632,7 @@ class Client extends EventEmitter {
       }
       this._sentQueryQueue.push(query)
     }
-    if (!this._activeQuery && this._sentQueryQueue.length > 0) {
+    if (this.readyForQuery && !this._activeQuery && this._sentQueryQueue.length > 0) {
       this._activeQuery = this._sentQueryQueue.shift()
       this.readyForQuery = false
     }
@@ -700,9 +696,12 @@ class Client extends EventEmitter {
         const index = this._queryQueue.indexOf(query)
         if (index > -1) {
           this._queryQueue.splice(index, 1)
+        } else if (this.pipelining) {
+          // Query already sent — the pipeline is blocked until it completes.
+          // Destroy the connection to unblock all remaining pipelined queries.
+          this.connection.stream.destroy()
+          return
         }
-        // If already sent on the wire in pipelining mode, we can't remove it
-        // without corrupting the pipeline — the callback was already no-op'd above
 
         this._pulseQueryQueue()
       }, readTimeout)
@@ -763,9 +762,15 @@ class Client extends EventEmitter {
       }
     }
 
-    if (this._getActiveQuery() || this._sentQueryQueue.length > 0 || !this._queryable) {
-      // if we have an active query we need to force a disconnect
-      // on the socket - otherwise a hung query could block end forever
+    if (!this._queryable) {
+      // socket is dead — force close
+      this.connection.stream.destroy()
+    } else if (this.pipelining && (this._getActiveQuery() || this._sentQueryQueue.length > 0 || this._queryQueue.length > 0)) {
+      // pipelined queries are already on the wire (or queued to send) and will
+      // complete normally; wait for drain then do a graceful goodbye
+      this.once('drain', () => this.connection.end())
+    } else if (this._getActiveQuery()) {
+      // non-pipelining: a hung query could block end forever — force disconnect
       this.connection.stream.destroy()
     } else {
       this.connection.end()

packages/pg/test/integration/client/pipelining-tests.js

@@ -86,3 +86,67 @@ suite.test('pipelining drain event', async function () {
   await drainPromise
   await client.end()
 })
+
+// #12: end() during active pipelining — should drain gracefully, not destroy
+suite.test('end() waits for in-flight pipelined queries to complete', async function () {
+  const client = helper.client()
+  client.pipelining = true
+
+  // Fire queries then call end() immediately without awaiting them
+  const p1 = client.query('SELECT 1 AS num')
+  const p2 = client.query('SELECT 2 AS num')
+  const endPromise = client.end()
+
+  // All queries should resolve (not error) because end() drains gracefully
+  const [r1, r2] = await Promise.all([p1, p2])
+  assert.equal(r1.rows[0].num, 1)
+  assert.equal(r2.rows[0].num, 2)
+  await endPromise
+})
+
+// #13: named statement error cleanup — submittedNamedStatements not left stale
+suite.test('named statement parse error cleans up and allows re-preparation', async function () {
+  const client = helper.client()
+  client.pipelining = true
+
+  // Use an invalid type to force a server-side parse error
+  const err = await client
+    .query({ name: 'bad-stmt', text: 'SELECT $1::nonexistent_type_xyz', values: [1] })
+    .then(() => null)
+    .catch((e) => e)
+
+  assert.ok(err, 'expected parse to fail')
+
+  // The stale submittedNamedStatements entry should be gone.
+  // Re-using the same name with valid SQL should work.
+  const result = await client.query({ name: 'bad-stmt', text: 'SELECT $1::int AS n', values: [42] })
+  assert.equal(result.rows[0].n, 42)
+
+  await client.end()
+})
+
+// #14: query_timeout with pipelining
+// When an already-sent pipelined query times out, the connection is destroyed
+// to unblock the pipeline — subsequent queries error rather than hanging.
+suite.test('query_timeout on sent pipelined query destroys connection to unblock', async function () {
+  const client = helper.client()
+  client.pipelining = true
+  client.on('error', () => {}) // absorb the 'error' event emitted when stream is destroyed
+
+  const results = await Promise.allSettled([
+    client.query('SELECT 1 AS num'),
+    client.query({ text: 'SELECT pg_sleep(30)', query_timeout: 100 }),
+    client.query('SELECT 3 AS num'),
+  ])
+
+  // Query 1 completes before the slow query enters the pipeline
+  assert.equal(results[0].status, 'fulfilled')
+  assert.equal(results[0].value.rows[0].num, 1)
+
+  // Query 2 times out
+  assert.equal(results[1].status, 'rejected')
+  assert.ok(results[1].reason.message.includes('timeout'), `unexpected error: ${results[1].reason.message}`)
+
+  // Query 3 errors because the connection was destroyed to unblock the pipeline
+  assert.equal(results[2].status, 'rejected')
+})

packages/pg/test/unit/client/simple-query-tests.js

@@ -172,6 +172,60 @@ test('executing query', function () {
       con.emit('readyForQuery')
       con.emit('readyForQuery')
     })
+
+    test('extended protocol: sends parse/bind/sync for each pipelined parameterized query', function () {
+      const client = helper.client()
+      client.pipelining = true
+      const con = client.connection
+      con.emit('readyForQuery')
+
+      client.query({ text: 'SELECT $1::int', values: [1] })
+      client.query({ text: 'SELECT $1::int', values: [2] })
+
+      // both parse messages should have been sent immediately
+      assert.lengthIs(con.parseMessages, 2)
+      assert.equal(con.parseMessages[0].text, 'SELECT $1::int')
+      assert.equal(con.parseMessages[1].text, 'SELECT $1::int')
+      // both bind messages too
+      assert.lengthIs(con.bindMessages, 2)
+      // each query sends its own sync
+      assert.equal(con.syncCount, 2)
+    })
+
+    test('named statement: parse sent only once when pipelining the same name', function () {
+      const client = helper.client()
+      client.pipelining = true
+      const con = client.connection
+      con.emit('readyForQuery')
+
+      client.query({ name: 'my-stmt', text: 'SELECT $1::int', values: [1] })
+      client.query({ name: 'my-stmt', text: 'SELECT $1::int', values: [2] })
+
+      // parse sent only once — second query reuses the submitted statement
+      assert.lengthIs(con.parseMessages, 1)
+      // both bind messages sent
+      assert.lengthIs(con.bindMessages, 2)
+    })
+
+    test('enabling pipelining while no queries are in flight', function () {
+      const client = helper.client()
+      const con = client.connection
+      con.emit('readyForQuery')
+
+      // start non-pipelining
+      assert.equal(client.pipelining, false)
+      client.query('before')
+      assert.lengthIs(con.queries, 1)
+
+      // simulate server responds
+      con.emit('readyForQuery')
+
+      // now enable pipelining — should work for subsequent queries
+      client.pipelining = true
+      client.query('after-one')
+      client.query('after-two')
+      assert.lengthIs(con.queries, 3)
+    })
   })
 
   test('handles errors', function () {

packages/pg/test/unit/client/test-helper.js

@@ -10,7 +10,22 @@ const makeClient = function (config) {
   connection.query = function (text) {
     this.queries.push(text)
   }
+  connection.parse = function (msg) {
+    this.parseMessages.push(msg)
+  }
+  connection.bind = function (msg) {
+    this.bindMessages.push(msg)
+  }
+  connection.describe = function (msg) {}
+  connection.execute = function (msg) {}
+  connection.sync = function () {
+    this.syncCount++
+  }
+  connection.flush = function () {}
   connection.queries = []
+  connection.parseMessages = []
+  connection.bindMessages = []
+  connection.syncCount = 0
   const client = new Client({ connection: connection, ...config })
   client.connect()
   client.connection.emit('connect')