Fix pipelining edge cases and add benchmark

- Clean up submittedNamedStatements on error in _handleErrorMessage to
  prevent stale entries from blocking future re-preparation of the same
  named statement after a parse failure
- Guard _pulsePipelinedQueryQueue against non-queryable connections
- Fix cancel() and readTimeout for sent queries: removing an already-sent
  query from _sentQueryQueue corrupts the pipeline response mapping since
  the server will still respond to it; no-op the callback instead
- Add bench-pipelining.js comparing serial vs pipelined throughput

Author: mcollina

packages/pg/bench-pipelining.js

@@ -0,0 +1,102 @@
+'use strict'
+const pg = require('./lib')
+
+// Queries to benchmark
+const SIMPLE = { text: 'SELECT 1' }
+const PARAM = {
+  text: 'SELECT $1::int AS n',
+  values: [42],
+}
+const NAMED = {
+  name: 'bench-named',
+  text: 'SELECT $1::int AS n',
+  values: [42],
+}
+
+async function bench(label, fn, seconds) {
+  // warmup
+  for (let i = 0; i < 100; i++) await fn()
+
+  const deadline = Date.now() + seconds * 1000
+  let count = 0
+  while (Date.now() < deadline) {
+    await fn()
+    count++
+  }
+  const qps = (count / seconds).toFixed(0)
+  console.log(`  ${label}: ${qps} qps (${count} queries in ${seconds}s)`)
+  return count / seconds
+}
+
+async function benchPipelined(label, makeQueries, batchSize, seconds) {
+  const client = new pg.Client()
+  await client.connect()
+  client.pipelining = true
+
+  // warmup
+  for (let i = 0; i < 10; i++) {
+    await Promise.all(makeQueries(batchSize).map((q) => client.query(q)))
+  }
+
+  const deadline = Date.now() + seconds * 1000
+  let count = 0
+  while (Date.now() < deadline) {
+    const queries = makeQueries(batchSize)
+    await Promise.all(queries.map((q) => client.query(q)))
+    count += batchSize
+  }
+  const qps = (count / seconds).toFixed(0)
+  console.log(`  ${label} (batch=${batchSize}): ${qps} qps`)
+
+  await client.end()
+  return count / seconds
+}
+
+async function runSerial(label, query, seconds) {
+  const client = new pg.Client()
+  await client.connect()
+
+  const qps = await bench(label, () => client.query(query), seconds)
+  await client.end()
+  return qps
+}
+
+async function run() {
+  const SECONDS = 5
+  const BATCH = 10
+
+  console.log('\n=== Serial (no pipelining) ===')
+  const serialSimple = await runSerial('simple SELECT 1', SIMPLE, SECONDS)
+  const serialParam = await runSerial('parameterized', PARAM, SECONDS)
+  const serialNamed = await runSerial('named prepared', NAMED, SECONDS)
+
+  console.log('\n=== Pipelined ===')
+  const pipedSimple = await benchPipelined(
+    'simple SELECT 1',
+    (n) => Array.from({ length: n }, () => ({ text: 'SELECT 1' })),
+    BATCH,
+    SECONDS
+  )
+  const pipedParam = await benchPipelined(
+    'parameterized',
+    (n) => Array.from({ length: n }, () => ({ text: 'SELECT $1::int AS n', values: [42] })),
+    BATCH,
+    SECONDS
+  )
+  const pipedNamed = await benchPipelined(
+    'named prepared',
+    (n) => Array.from({ length: n }, (_, i) => ({ name: `bench-named-${i}`, text: 'SELECT $1::int AS n', values: [42] })),
+    BATCH,
+    SECONDS
+  )
+
+  console.log('\n=== Speedup ===')
+  console.log(`  simple:       ${(pipedSimple / serialSimple).toFixed(2)}x`)
+  console.log(`  parameterized: ${(pipedParam / serialParam).toFixed(2)}x`)
+  console.log(`  named:        ${(pipedNamed / serialNamed).toFixed(2)}x`)
+}
+
+run().catch((e) => {
+  console.error(e)
+  process.exit(1)
+})

packages/pg/lib/client.js

@@ -415,6 +415,9 @@ class Client extends EventEmitter {
     }
 
     this._activeQuery = null
+    if (activeQuery.name) {
+      delete this.connection.submittedNamedStatements[activeQuery.name]
+    }
     activeQuery.handleError(msg, this.connection)
   }
 
@@ -565,7 +568,9 @@ class Client extends EventEmitter {
     } else if (client._queryQueue.indexOf(query) !== -1) {
       client._queryQueue.splice(client._queryQueue.indexOf(query), 1)
     } else if (client._sentQueryQueue.indexOf(query) !== -1) {
-      client._sentQueryQueue.splice(client._sentQueryQueue.indexOf(query), 1)
+      // Query already sent on wire — can't remove it without corrupting the
+      // pipeline. No-op the callback so the result is silently discarded.
+      query.callback = () => {}
     }
   }
 
@@ -616,7 +621,7 @@ class Client extends EventEmitter {
   }
 
   _pulsePipelinedQueryQueue() {
-    if (!this._connected) {
+    if (!this._connected || !this._queryable) {
       return
     }
     while (this._queryQueue.length > 0) {
@@ -691,16 +696,13 @@ class Client extends EventEmitter {
         // just do nothing if query completes
         query.callback = () => {}
 
-        // Remove from queue
+        // Remove from queue (only safe if not yet sent)
         const index = this._queryQueue.indexOf(query)
         if (index > -1) {
           this._queryQueue.splice(index, 1)
-        } else if (this.pipelining) {
-          const sentIndex = this._sentQueryQueue.indexOf(query)
-          if (sentIndex > -1) {
-            this._sentQueryQueue.splice(sentIndex, 1)
-          }
         }
+        // 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)