feat(pipeline-mode): add backpressure support with configurable query limits

Author: Simone Nigro

docs/pages/features/queries.mdx

@@ -211,6 +211,37 @@ await client.query('INSERT INTO t VALUES (2)') // also fails: transaction aborte
 await client.query('COMMIT') // returns ROLLBACK command
 ```
 
+### Backpressure
+
+When submitting a large number of queries in pipeline mode, you may want to limit how many queries are pending at once to prevent memory issues. The client supports configurable backpressure:
+
+```js
+const client = new Client({
+  pipelineMode: true,
+  pipelineMaxQueries: 100, // default is 1000
+})
+```
+
+When the number of pending queries reaches `pipelineMaxQueries`, new queries are queued internally and will be sent once space becomes available. The client emits events to help you monitor backpressure:
+
+```js
+client.on('pipelineFull', () => {
+  console.log('Pipeline is full, new queries will wait')
+})
+
+client.on('pipelineDrain', () => {
+  console.log('Pipeline has drained, accepting new queries')
+})
+```
+
+You can also check the current pipeline depth:
+
+```js
+console.log(client.pendingQueryCount) // number of queries waiting for results
+```
+
+The `pipelineDrain` event is emitted when the pending query count drops below 75% of `pipelineMaxQueries` (the "low water mark").
+
 ### When to Use Pipeline Mode
 
 Pipeline mode is most beneficial when:

packages/pg/lib/client.js

@@ -84,6 +84,12 @@ class Client extends EventEmitter {
     // Pipeline mode configuration
     this._pipelineMode = Boolean(c.pipelineMode) || false
 
+    // Backpressure configuration for pipeline mode
+    this._pipelineMaxQueries = c.pipelineMaxQueries || 1000
+    this._lowWaterMark = Math.floor(this._pipelineMaxQueries * 0.75)
+    this._pipelinePaused = false
+    this._waitingForDrain = []
+
     // Queue for tracking pending query results in pipeline mode
     this._pendingQueries = []
 
@@ -132,6 +138,11 @@ class Client extends EventEmitter {
     return this._pipelineMode
   }
 
+  // Property for current pipeline depth (number of pending queries)
+  get pendingQueryCount() {
+    return this._pendingQueries.length + this._queryQueue.length
+  }
+
   _errorAllQueries(err) {
     const enqueueError = (query) => {
       process.nextTick(() => {
@@ -394,6 +405,9 @@ class Client extends EventEmitter {
       ) {
         this._pendingQueries.shift()
         currentQuery.handleReadyForQuery(this.connection)
+
+        // Check if backpressure should be released after query completion
+        this._checkBackpressureResume()
       }
 
       // Check if more queries to send
@@ -774,6 +788,20 @@ class Client extends EventEmitter {
     return this._pendingQueries[0]
   }
 
+  // Called when queries complete to check if backpressure should be released
+  // Resumes accepting new queries when pipeline depth drops below low water mark
+  _checkBackpressureResume() {
+    if (this._pipelinePaused && this.pendingQueryCount < this._lowWaterMark) {
+      this._pipelinePaused = false
+      this.emit('pipelineDrain')
+
+      // Resume all waiting queries
+      const waiting = this._waitingForDrain
+      this._waitingForDrain = []
+      waiting.forEach((resolve) => resolve())
+    }
+  }
+
   query(config, values, callback) {
     // can take in strings, config object or query object
     let query
@@ -875,11 +903,37 @@ class Client extends EventEmitter {
       return result
     }
 
+    // Backpressure handling for pipeline mode
+    // we queue the query and delay its execution
+    // the query is added to the queue immediately, but _pulseQueryQueue respects backpressure
+    if (this._pipelineMode && this.pendingQueryCount >= this._pipelineMaxQueries) {
+      // Emit pipelineFull event if not already paused
+      if (!this._pipelinePaused) {
+        this._pipelinePaused = true
+        this.emit('pipelineFull')
+      }
+      // Queue the query with backpressure - it will be processed when space is available
+      this._queueQueryWithBackpressure(query)
+      return result
+    }
+
     this._queryQueue.push(query)
     this._pulseQueryQueue()
     return result
   }
 
+  // Internal method to queue a query when backpressure is active
+  // The query waits for drain before being added to the main queue
+  _queueQueryWithBackpressure(query) {
+    const waitForDrain = new this._Promise((resolve) => {
+      this._waitingForDrain.push(resolve)
+    })
+    waitForDrain.then(() => {
+      this._queryQueue.push(query)
+      this._pulseQueryQueue()
+    })
+  }
+
   ref() {
     this.connection.ref()
   }

packages/pg/test/integration/client/pipeline-mode-tests.js

@@ -1655,3 +1655,207 @@ suite.test('pipeline mode - Pool with size 1 maintains transaction isolation', (
       pool.end().then(() => done(err))
     })
 })
+
+suite.test('pipeline mode - pipelineFull event is emitted when pipelineMaxQueries is reached', (done) => {
+  // Use a small pipelineMaxQueries to trigger backpressure quickly
+  const client = new Client({ pipelineMode: true, pipelineMaxQueries: 10 })
+
+  let pipelineFullEmitted = false
+  client.on('pipelineFull', () => {
+    pipelineFullEmitted = true
+  })
+
+  client.connect((err) => {
+    if (err) return done(err)
+
+    // Submit more queries than pipelineMaxQueries to trigger backpressure
+    const promises = []
+    for (let i = 0; i < 15; i++) {
+      promises.push(client.query('SELECT $1::int as num, pg_sleep(0.01)', [i]))
+    }
+
+    Promise.all(promises)
+      .then((results) => {
+        // Verify all queries completed successfully
+        assert.equal(results.length, 15, 'All 15 queries should complete')
+
+        // Verify pipelineFull event was emitted
+        assert.ok(pipelineFullEmitted, 'pipelineFull event should have been emitted')
+
+        client.end(done)
+      })
+      .catch((err) => {
+        client.end(() => done(err))
+      })
+  })
+})
+
+suite.test('pipeline mode - pipelineDrain event is emitted when pipeline depth drops below low water mark', (done) => {
+  // Use a small pipelineMaxQueries to trigger backpressure quickly
+  // lowWaterMark will be 75% of 10 = 7
+  const client = new Client({ pipelineMode: true, pipelineMaxQueries: 10 })
+
+  let pipelineFullEmitted = false
+  let pipelineDrainEmitted = false
+
+  client.on('pipelineFull', () => {
+    pipelineFullEmitted = true
+  })
+
+  client.on('pipelineDrain', () => {
+    pipelineDrainEmitted = true
+  })
+
+  client.connect((err) => {
+    if (err) return done(err)
+
+    // Submit more queries than pipelineMaxQueries to trigger backpressure
+    // Then wait for them to complete and verify pipelineDrain is emitted
+    const promises = []
+    for (let i = 0; i < 15; i++) {
+      promises.push(client.query('SELECT $1::int as num, pg_sleep(0.01)', [i]))
+    }
+
+    Promise.all(promises)
+      .then((results) => {
+        // Verify all queries completed successfully
+        assert.equal(results.length, 15, 'All 15 queries should complete')
+
+        // Verify both events were emitted
+        assert.ok(pipelineFullEmitted, 'pipelineFull event should have been emitted')
+        assert.ok(
+          pipelineDrainEmitted,
+          'pipelineDrain event should have been emitted when pipeline depth dropped below low water mark'
+        )
+
+        client.end(done)
+      })
+      .catch((err) => {
+        client.end(() => done(err))
+      })
+  })
+})
+
+suite.test('pipeline mode - backpressure activation with real PostgreSQL connection', (done) => {
+  // Test that backpressure correctly pauses query submission when pipelineMaxQueries is reached
+  const pipelineMaxQueries = 5
+  const client = new Client({ pipelineMode: true, pipelineMaxQueries: pipelineMaxQueries })
+
+  let pipelineFullCount = 0
+  let maxObservedPendingCount = 0
+
+  client.on('pipelineFull', () => {
+    pipelineFullCount++
+    // Record the pending count when pipelineFull is emitted
+    if (client.pendingQueryCount > maxObservedPendingCount) {
+      maxObservedPendingCount = client.pendingQueryCount
+    }
+  })
+
+  client.connect((err) => {
+    if (err) return done(err)
+
+    // Submit many queries - backpressure should prevent pendingQueryCount from exceeding pipelineMaxQueries
+    const numQueries = 20
+    const promises = []
+
+    for (let i = 0; i < numQueries; i++) {
+      promises.push(
+        client.query('SELECT $1::int as num, pg_sleep(0.02)', [i]).then((r) => ({ success: true, num: r.rows[0].num }))
+      )
+    }
+
+    Promise.all(promises)
+      .then((results) => {
+        // All queries should complete successfully
+        assert.equal(results.length, numQueries, `All ${numQueries} queries should complete`)
+        results.forEach((r, idx) => {
+          assert.equal(r.success, true, `Query ${idx} should succeed`)
+          assert.equal(r.num, idx.toString(), `Query ${idx} should return correct value`)
+        })
+
+        // pipelineFull should have been emitted at least once
+        assert.ok(pipelineFullCount >= 1, 'pipelineFull event should have been emitted at least once')
+
+        // pendingQueryCount should never have exceeded pipelineMaxQueries significantly
+        // (there may be slight timing variations, so we allow a small buffer)
+        assert.ok(
+          maxObservedPendingCount <= pipelineMaxQueries + 2,
+          `maxObservedPendingCount (${maxObservedPendingCount}) should not significantly exceed pipelineMaxQueries (${pipelineMaxQueries})`
+        )
+
+        client.end(done)
+      })
+      .catch((err) => {
+        client.end(() => done(err))
+      })
+  })
+})
+
+suite.test('pipeline mode - backpressure release and query resumption', (done) => {
+  // Test that queries resume correctly after backpressure is released
+  const pipelineMaxQueries = 5
+  const client = new Client({ pipelineMode: true, pipelineMaxQueries: pipelineMaxQueries })
+
+  let pipelineFullEmitted = false
+  let pipelineDrainEmitted = false
+  const eventOrder = []
+
+  client.on('pipelineFull', () => {
+    pipelineFullEmitted = true
+    eventOrder.push('full')
+  })
+
+  client.on('pipelineDrain', () => {
+    pipelineDrainEmitted = true
+    eventOrder.push('drain')
+  })
+
+  client.connect((err) => {
+    if (err) return done(err)
+
+    // Submit queries that will trigger backpressure and then drain
+    const numQueries = 15
+    const promises = []
+    const completionOrder = []
+
+    for (let i = 0; i < numQueries; i++) {
+      promises.push(
+        client.query('SELECT $1::int as num, pg_sleep(0.01)', [i]).then((r) => {
+          completionOrder.push(parseInt(r.rows[0].num))
+          return r.rows[0].num
+        })
+      )
+    }
+
+    Promise.all(promises)
+      .then((results) => {
+        // All queries should complete
+        assert.equal(results.length, numQueries, `All ${numQueries} queries should complete`)
+
+        // Verify pipelineFull was emitted (backpressure activated)
+        assert.ok(pipelineFullEmitted, 'pipelineFull event should have been emitted')
+
+        // Verify pipelineDrain was emitted (backpressure released)
+        assert.ok(pipelineDrainEmitted, 'pipelineDrain event should have been emitted')
+
+        // Verify event order: full should come before drain
+        const fullIndex = eventOrder.indexOf('full')
+        const drainIndex = eventOrder.indexOf('drain')
+        assert.ok(fullIndex < drainIndex, 'pipelineFull should be emitted before pipelineDrain')
+
+        // Verify queries completed in submission order (pipeline preserves order)
+        for (let i = 0; i < completionOrder.length; i++) {
+          assert.equal(completionOrder[i], i, `Query ${i} should complete in order`)
+        }
+
+        // Verify pendingQueryCount is 0 after all queries complete
+        assert.equal(client.pendingQueryCount, 0, 'pendingQueryCount should be 0 after all queries complete')
+
+        client.end(done)
+      })
+      .catch((err) => {
+        client.end(() => done(err))
+      })
+  })
+})