From 8ae7f068cbb8879bed502362e4c8d8d8eaf5510f Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Mon, 10 Aug 2026 06:25:44 +0200 Subject: [PATCH 1/7] Let pool.query pipeline on a connection that is already working --- docs/pages/apis/pool.mdx | 13 ++- docs/pages/features/pipelining.mdx | 40 ++++++--- packages/pg-pool/index.js | 133 ++++++++++++++++++++++++++--- packages/pg-pool/test/pipeline.js | 122 ++++++++++++++++++++++++++ 4 files changed, 281 insertions(+), 27 deletions(-) create mode 100644 packages/pg-pool/test/pipeline.js diff --git a/docs/pages/apis/pool.mdx b/docs/pages/apis/pool.mdx index 3627dd1c5..3686da023 100644 --- a/docs/pages/apis/pool.mdx +++ b/docs/pages/apis/pool.mdx @@ -31,6 +31,8 @@ type Config = { // Maximum number of clients the pool should contain. // By default this is set to 10. There is some nuance to setting the maximum size of your pool. // See https://node-postgres.com/guides/pool-sizing for more information. + // With `pipeline: true` this is still the number of connections, but the number of queries + // in flight can reach max * maxPipeline. max?: number // Minimum number of clients the pool should hold on to and _not_ destroy with the idleTimeoutMillis. @@ -71,9 +73,16 @@ type Config = { onConnect?: (client: Client) => void | Promise // When set to true, enables query pipelining on every client the pool creates. - // Pipelined clients send queries to the server without waiting for previous responses. - // Default is false. See /features/pipelining for details. + // Pipelined clients send queries to the server without waiting for previous responses, and + // pool.query() can use a connection that is already working instead of waiting for a free one. + // pool.connect() is not affected, it still checks out a connection nobody else is using. + // Queries can complete in a different order, so the default is false. + // See /features/pipelining for details. pipeline?: boolean + + // Maximum number of queries pool.query() sends on the same connection before waiting. + // Only used with `pipeline: true`. The default is 10. + maxPipeline?: number } ``` diff --git a/docs/pages/features/pipelining.mdx b/docs/pages/features/pipelining.mdx index 7943aa490..9c9e00833 100644 --- a/docs/pages/features/pipelining.mdx +++ b/docs/pages/features/pipelining.mdx @@ -51,24 +51,38 @@ Pass `pipeline: true` in the pool config to enable it on every client the pool c ```js import { Pool } from 'pg' -const pool = new Pool({ pipeline: true }) - -const client = await pool.connect() -// client.pipeline is already true +const pool = new Pool({ max: 10, pipeline: true, maxPipeline: 10 }) const [users, orders] = await Promise.all([ - client.query('SELECT * FROM users WHERE id = $1', [1]), - client.query('SELECT * FROM orders WHERE user_id = $1', [1]), + pool.query('SELECT * FROM users WHERE id = $1', [1]), + pool.query('SELECT * FROM orders WHERE user_id = $1', [1]), ]) - -client.release() ``` - -
- pool.query() checks out a client for a single query and releases it immediately, so pipelining has no effect there. Use pool.connect() to check out a client and send multiple queries on it. -
- +Without pipelining `pool.query()` holds a connection for one query, so a query has to wait for a +free connection. With `pipeline: true` the pool sends it on a connection that is already working, +once every connection is busy. It opens connections up to `max` first, then picks the one with the +fewest queries in flight, up to `maxPipeline` per connection (default 10). + +`max` is still the number of connections. It is not the number of queries in flight anymore, those +can reach `max * maxPipeline`. This does not add load on the server, PostgreSQL runs a pipeline +serially on the same backend, but a query can wait behind the ones already sent on its connection. +Queries can also complete in a different order than without pipelining, which is why this is off by +default. + +`pool.connect()` is not affected: it still checks out a connection nobody else is using, so +transactions, `SET` and cursors keep working as before. + +```js +const client = await pool.connect() +try { + await client.query('BEGIN') + await client.query('INSERT INTO users(name) VALUES($1)', ['brianc']) + await client.query('COMMIT') +} finally { + client.release() +} +``` ## Error isolation diff --git a/packages/pg-pool/index.js b/packages/pg-pool/index.js index ab514fa88..dafe36c8a 100644 --- a/packages/pg-pool/index.js +++ b/packages/pg-pool/index.js @@ -18,11 +18,25 @@ class IdleItem { } class PendingItem { - constructor(callback) { + // exclusive wants a connection nobody else is using, which is what pool.connect() gives + constructor(callback, exclusive = true) { this.callback = callback + this.exclusive = exclusive } } +// queries written on a client whose result has not come back yet +const inFlight = (client) => client._poolPipelined || 0 + +const canPipeline = (client, maxPipeline) => { + if (inFlight(client) >= maxPipeline) { + return false + } + // streams without writableNeedDrain (pg-cloudflare) count as never backed up + const stream = client.connection && client.connection.stream + return !(stream && stream.writableNeedDrain) +} + function throwOnDoubleRelease() { throw new Error('Release called on client which has already been released to the pool.') } @@ -56,7 +70,13 @@ function makeIdleListener(pool, client) { client.on('error', () => { pool.log('additional client error after disconnection due to error', err) }) + const wasPipelining = inFlight(client) > 0 pool._remove(client) + // a pipelined client is not idle: every query on it already gets this error + if (wasPipelining) { + pool.log('client error while pipelining, reported to the queries in flight', err) + return + } // TODO - document that once the pool emits an error // the client has already been closed & purged and is unusable pool.emit('error', err, client) @@ -88,6 +108,9 @@ class Pool extends EventEmitter { this.options.max = this.options.max || this.options.poolSize || 10 this.options.min = this.options.min || 0 + // max stays the number of connections, queries in flight can reach max * maxPipeline + this.options.pipeline = Boolean(this.options.pipeline) + this.options.maxPipeline = this.options.maxPipeline || 10 this.options.maxUses = this.options.maxUses || Infinity this.options.allowExitOnIdle = this.options.allowExitOnIdle || false this.options.maxLifetimeSeconds = this.options.maxLifetimeSeconds || 0 @@ -134,6 +157,10 @@ class Pool extends EventEmitter { this.log('pulse queue on ending') if (this._idle.length) { this._idle.slice().map((item) => { + // still finishing pipelined queries, the last one to come back pulses again + if (inFlight(item.client)) { + return + } this._remove(item.client) }) } @@ -153,20 +180,49 @@ class Pool extends EventEmitter { if (!this._idle.length && this._isFull()) { return } - const pendingItem = this._pendingQueue.shift() - if (this._idle.length) { - const idleItem = this._idle.pop() + const idleItem = this._takeIdleItem(this._pendingQueue[0]) + if (idleItem) { clearTimeout(idleItem.timeoutId) const client = idleItem.client client.ref && client.ref() const idleListener = idleItem.idleListener - return this._acquireClient(client, pendingItem, idleListener, false) + return this._acquireClient(client, this._pendingQueue.shift(), idleListener, false) } if (!this._isFull()) { - return this.newClient(pendingItem) + return this.newClient(this._pendingQueue.shift()) + } + // pipelining: every connection is at maxPipeline or backed up, wait for a result + } + + // the idle client to serve this item, or undefined to open a new connection + _takeIdleItem(pendingItem) { + if (!this._idle.length) { + return undefined } - throw new Error('unexpected condition') + if (!this.options.pipeline) { + return this._idle.pop() + } + + const free = this._idle.findIndex((item) => inFlight(item.client) === 0) + if (free !== -1) { + return this._idle.splice(free, 1)[0] + } + // a transaction cannot share a connection, and a query prefers a new one + if (pendingItem.exclusive || !this._isFull()) { + return undefined + } + + let best + for (const item of this._idle) { + if ( + canPipeline(item.client, this.options.maxPipeline) && + (!best || inFlight(item.client) < inFlight(best.client)) + ) { + best = item + } + } + return best && this._idle.splice(this._idle.indexOf(best), 1)[0] } _remove(client, callback) { @@ -188,6 +244,10 @@ class Pool extends EventEmitter { } connect(cb) { + return this._checkout(true, cb) + } + + _checkout(exclusive, cb) { if (this.ending) { const err = new Error('Cannot use a pool after calling end on the pool') return cb ? cb(err) : this.Promise.reject(err) @@ -204,7 +264,7 @@ class Pool extends EventEmitter { } if (!this.options.connectionTimeoutMillis) { - this._pendingQueue.push(new PendingItem(response.callback)) + this._pendingQueue.push(new PendingItem(response.callback, exclusive)) return result } @@ -213,7 +273,7 @@ class Pool extends EventEmitter { response.callback(err, res, done) } - const pendingItem = new PendingItem(queueCallback) + const pendingItem = new PendingItem(queueCallback, exclusive) // set connection timeout on checking out an existing client const tid = setTimeout(() => { @@ -232,7 +292,7 @@ class Pool extends EventEmitter { return result } - this.newClient(new PendingItem(response.callback)) + this.newClient(new PendingItem(response.callback, exclusive)) return result } @@ -408,7 +468,8 @@ class Pool extends EventEmitter { let tid if (this.options.idleTimeoutMillis && this._isAboveMin()) { tid = setTimeout(() => { - if (this._isAboveMin()) { + // a client with queries in flight is not idle, the next release re-arms this + if (this._isAboveMin() && !inFlight(client)) { this.log('remove idle client') this._remove(client, this._pulseQueue.bind(this)) } @@ -420,7 +481,7 @@ class Pool extends EventEmitter { } } - if (this.options.allowExitOnIdle) { + if (this.options.allowExitOnIdle && !inFlight(client)) { client.unref() } @@ -446,6 +507,11 @@ class Pool extends EventEmitter { const response = promisify(this.Promise, cb) cb = response.callback + if (this.options.pipeline) { + this._pipelineQuery(text, values, cb) + return response.result + } + this.connect((err, client) => { if (err) { return cb(err) @@ -485,6 +551,49 @@ class Pool extends EventEmitter { return response.result } + // the connection goes back to the pool as soon as the query is written, so the + // next one can be sent behind it instead of waiting for the result + _pipelineQuery(text, values, cb) { + this._checkout(false, (err, client, release) => { + if (err) { + return cb(err) + } + + let released = false + const releaseOnce = (err) => { + if (!released) { + released = true + release(err) + } + } + + client._poolPipelined = inFlight(client) + 1 + this.log('dispatching query') + try { + client.query(text, values, (err, res) => { + this.log('query dispatched') + client._poolPipelined-- + // no-op unless the query failed before it was written + releaseOnce() + try { + return err ? cb(err) : cb(undefined, res) + } finally { + // a slot just freed up, and this caller comes before the next one + this._pulseQueue() + } + }) + } catch (err) { + client._poolPipelined-- + releaseOnce(err) + return cb(err) + } + + // on a tick of its own: releasing here would re-enter _pulseQueue and + // recurse once per queued query + process.nextTick(releaseOnce) + }) + } + end(cb) { this.log('ending') if (this.ending) { diff --git a/packages/pg-pool/test/pipeline.js b/packages/pg-pool/test/pipeline.js new file mode 100644 index 000000000..d53a5bd23 --- /dev/null +++ b/packages/pg-pool/test/pipeline.js @@ -0,0 +1,122 @@ +const describe = require('mocha').describe +const it = require('mocha').it +const expect = require('expect.js') + +const Pool = require('..') + +const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms)) + +// the assertions below look at the pool while queries are in flight, so open the +// first connection up front and keep connecting out of the measured window +const warm = async (options) => { + const pool = new Pool(options) + await pool.query('SELECT 1') + return pool +} + +describe('pipeline', () => { + it('sends a query on a connection that is already working', async () => { + const pool = await warm({ max: 1, pipeline: true }) + const slow = pool.query('SELECT pg_sleep(0.3)') + const fast = pool.query('SELECT 1 AS num') + await wait(50) + + expect(pool.waitingCount).to.equal(0) + expect(pool.totalCount).to.equal(1) + expect((await fast).rows[0].num).to.equal(1) + + await slow + await pool.end() + }) + + it('opens connections up to max before pipelining', async () => { + const pool = await warm({ max: 3, pipeline: true }) + const queries = [1, 2, 3, 4, 5, 6].map((num) => pool.query('SELECT pg_sleep(0.2), $1::int AS num', [num])) + await wait(50) + + expect(pool.totalCount).to.equal(3) + expect(pool.waitingCount).to.equal(0) + + const results = await Promise.all(queries) + expect(results.map((res) => res.rows[0].num)).to.eql([1, 2, 3, 4, 5, 6]) + await pool.end() + }) + + it('does not send more than maxPipeline on one connection', async () => { + const pool = await warm({ max: 1, maxPipeline: 2, pipeline: true }) + const queries = [1, 2, 3, 4, 5].map((num) => pool.query('SELECT pg_sleep(0.1), $1::int AS num', [num])) + await wait(50) + + expect(pool.waitingCount).to.equal(3) + + const results = await Promise.all(queries) + expect(results.map((res) => res.rows[0].num)).to.eql([1, 2, 3, 4, 5]) + await pool.end() + }) + + it('gives pool.connect a connection nobody else is using', async () => { + const pool = await warm({ max: 1, pipeline: true }) + const query = pool.query('SELECT pg_sleep(0.2)') + await wait(50) + + let checkedOut = false + const checkout = pool.connect().then((client) => { + checkedOut = true + client.release() + }) + await wait(50) + expect(checkedOut).to.equal(false) + + await query + await checkout + expect(checkedOut).to.equal(true) + await pool.end() + }) + + it('keeps the connection after a query error', async () => { + const pool = await warm({ max: 1, pipeline: true }) + const bad = pool.query('SELECT * FROM table_that_does_not_exist') + const good = pool.query('SELECT 1 AS num') + + await bad.then( + () => { + throw new Error('expected the query to fail') + }, + (err) => expect(err.message).to.contain('table_that_does_not_exist') + ) + expect((await good).rows[0].num).to.equal(1) + expect(pool.totalCount).to.equal(1) + + expect((await pool.query('SELECT 2 AS num')).rows[0].num).to.equal(2) + await pool.end() + }) + + it('reports a broken connection to every query on it', async () => { + const pool = new Pool({ max: 1, pipeline: true }) + let client + pool.once('connect', (c) => (client = c)) + await pool.query('SELECT 1') + + const queries = [1, 2, 3].map(() => pool.query('SELECT pg_sleep(0.2)')) + await wait(50) + client.connection.stream.destroy() + + const results = await Promise.allSettled(queries) + expect(results.map((res) => res.status)).to.eql(['rejected', 'rejected', 'rejected']) + expect(pool.totalCount).to.equal(0) + + expect((await pool.query('SELECT 1 AS num')).rows[0].num).to.equal(1) + await pool.end() + }) + + it('waits for the queries in flight on end', async () => { + const pool = await warm({ max: 1, pipeline: true }) + const queries = [1, 2, 3].map((num) => pool.query('SELECT pg_sleep(0.1), $1::int AS num', [num])) + await wait(50) + expect(pool.waitingCount).to.equal(0) + + await pool.end() + const results = await Promise.all(queries) + expect(results.map((res) => res.rows[0].num)).to.eql([1, 2, 3]) + }) +}) From c545d4c88956992496b6924aad35e775db5a0789 Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Mon, 10 Aug 2026 07:10:22 +0200 Subject: [PATCH 2/7] Keep the pool consistent when a pipelined query cannot be answered --- docs/pages/features/pipelining.mdx | 13 +- packages/pg-pool/index.js | 91 ++++++-- packages/pg-pool/test/idle-timeout-exit.js | 1 + packages/pg-pool/test/pipeline.js | 233 +++++++++++++++++++++ 4 files changed, 315 insertions(+), 23 deletions(-) diff --git a/docs/pages/features/pipelining.mdx b/docs/pages/features/pipelining.mdx index 9c9e00833..6524361bf 100644 --- a/docs/pages/features/pipelining.mdx +++ b/docs/pages/features/pipelining.mdx @@ -71,7 +71,7 @@ Queries can also complete in a different order than without pipelining, which is default. `pool.connect()` is not affected: it still checks out a connection nobody else is using, so -transactions, `SET` and cursors keep working as before. +transactions and `SET` keep working as before. ```js const client = await pool.connect() @@ -84,6 +84,11 @@ try { } ``` +`pool.query()` refuses a submittable (`pg-cursor`, `pg-query-stream`) while pipelining, because +those read their rows over several round trips and cannot share a connection. Check out a client +with `pool.connect()` for them. `connectionTimeoutMillis` also covers the wait for a free pipeline +slot, not only the wait for a connection. + ## Error isolation Each pipelined query gets its own error boundary. A failing query in the middle of a batch does not break the other queries: @@ -116,6 +121,12 @@ const queries = Array.from({ length: 100 }, (_, i) => ({ const results = await Promise.all(queries.map(q => client.query(q))) ``` +## Query timeouts + +`query_timeout` cannot cancel a single pipelined query, because the queries behind it are already on the wire. +When it fires node-postgres closes the connection, so the other queries sharing it fail too. +With a pool the connection is then replaced, and `pool.query()` keeps working, but you should have a `pool.on('error')` listener as usual. + ## Graceful shutdown Calling `client.end()` while pipelined queries are in flight will wait for all of them to complete before closing the connection: diff --git a/packages/pg-pool/index.js b/packages/pg-pool/index.js index dafe36c8a..a0557c38c 100644 --- a/packages/pg-pool/index.js +++ b/packages/pg-pool/index.js @@ -225,6 +225,16 @@ class Pool extends EventEmitter { return best && this._idle.splice(this._idle.indexOf(best), 1)[0] } + // a client with pipelined queries on it still owes results, so it stays in _clients and out of + // _idle until they arrive: dropping it now would let the pool hold more sockets than max + _removeWhenIdle(client) { + if (inFlight(client)) { + client._poolRemoveWhenIdle = true + return + } + return this._remove(client, this._pulseQueue.bind(this)) + } + _remove(client, callback) { const removed = removeWhere(this._idle, (item) => item.client === client) @@ -454,31 +464,43 @@ class Pool extends EventEmitter { this.log('remove expended client') } - return this._remove(client, this._pulseQueue.bind(this)) + return this._removeWhenIdle(client) } const isExpired = this._expired.has(client) if (isExpired) { this.log('remove expired client') this._expired.delete(client) - return this._remove(client, this._pulseQueue.bind(this)) + return this._removeWhenIdle(client) } // idle timeout let tid if (this.options.idleTimeoutMillis && this._isAboveMin()) { - tid = setTimeout(() => { - // a client with queries in flight is not idle, the next release re-arms this - if (this._isAboveMin() && !inFlight(client)) { + const armIdleTimeout = () => { + const timer = setTimeout(() => { + if (!this._isAboveMin()) { + return + } + // pipelined queries are still coming back, this client is not idle yet + if (inFlight(client)) { + const item = this._idle.find((idleItem) => idleItem.client === client) + if (item) { + item.timeoutId = armIdleTimeout() + } + return + } this.log('remove idle client') this._remove(client, this._pulseQueue.bind(this)) - } - }, this.options.idleTimeoutMillis) + }, this.options.idleTimeoutMillis) - if (this.options.allowExitOnIdle) { - // allow Node to exit if this is all that's left - tid.unref() + if (this.options.allowExitOnIdle) { + // allow Node to exit if this is all that's left + timer.unref() + } + return timer } + tid = armIdleTimeout() } if (this.options.allowExitOnIdle && !inFlight(client)) { @@ -554,6 +576,15 @@ class Pool extends EventEmitter { // the connection goes back to the pool as soon as the query is written, so the // next one can be sent behind it instead of waiting for the result _pipelineQuery(text, values, cb) { + // a submittable answers on its own object, not through this callback, + // so the pool would never learn the query is over and would hold the connection forever + if (text != null && typeof text.submit === 'function') { + process.nextTick(() => + cb(new Error('pool.query does not take a submittable in pipeline mode, check out a client with pool.connect()')) + ) + return + } + this._checkout(false, (err, client, release) => { if (err) { return cb(err) @@ -569,20 +600,36 @@ class Pool extends EventEmitter { client._poolPipelined = inFlight(client) + 1 this.log('dispatching query') + let answered = false + const answer = (err, res) => { + // a query that fails while it is being written is answered twice by the client + if (answered) { + return + } + answered = true + this.log('query dispatched') + client._poolPipelined-- + if (this.options.allowExitOnIdle && !inFlight(client)) { + // the release ran while this result was still pending, so it could not unref + client.unref() + } + if (!inFlight(client) && client._poolRemoveWhenIdle) { + this._remove(client, this._pulseQueue.bind(this)) + } + // no-op unless the query failed before it was written + releaseOnce() + try { + return err ? cb(err) : cb(undefined, res) + } finally { + // a slot just freed up, and this caller comes before the next one + this._pulseQueue() + } + } + try { - client.query(text, values, (err, res) => { - this.log('query dispatched') - client._poolPipelined-- - // no-op unless the query failed before it was written - releaseOnce() - try { - return err ? cb(err) : cb(undefined, res) - } finally { - // a slot just freed up, and this caller comes before the next one - this._pulseQueue() - } - }) + client.query(text, values, answer) } catch (err) { + answered = true client._poolPipelined-- releaseOnce(err) return cb(err) diff --git a/packages/pg-pool/test/idle-timeout-exit.js b/packages/pg-pool/test/idle-timeout-exit.js index 7304bcff1..54591cc84 100644 --- a/packages/pg-pool/test/idle-timeout-exit.js +++ b/packages/pg-pool/test/idle-timeout-exit.js @@ -6,6 +6,7 @@ if (module === require.main) { const pool = new Pool({ maxLifetimeSeconds: 2, idleTimeoutMillis: 200, + pipeline: process.env.PIPELINE === '1', ...(allowExitOnIdle ? { allowExitOnIdle: true } : {}), }) pool.query('SELECT NOW()', (err, res) => console.log('completed first')) diff --git a/packages/pg-pool/test/pipeline.js b/packages/pg-pool/test/pipeline.js index d53a5bd23..0c0fc2bc9 100644 --- a/packages/pg-pool/test/pipeline.js +++ b/packages/pg-pool/test/pipeline.js @@ -1,3 +1,6 @@ +const path = require('path') +const { fork } = require('child_process') +const Cursor = require('pg-cursor') const describe = require('mocha').describe const it = require('mocha').it const expect = require('expect.js') @@ -54,6 +57,210 @@ describe('pipeline', () => { await pool.end() }) + it('answers every query of a large burst', async () => { + const pool = await warm({ max: 2, pipeline: true }) + const nums = Array.from({ length: 500 }, (_, i) => i) + const results = await Promise.all(nums.map((num) => pool.query('SELECT $1::int AS num', [num]))) + + expect(results.map((res) => res.rows[0].num)).to.eql(nums) + expect(pool.totalCount).to.equal(2) + await pool.end() + }) + + it('accepts the callback and the values forms', async () => { + const pool = new Pool({ max: 1, pipeline: true }) + const withValues = await new Promise((resolve, reject) => { + pool.query('SELECT $1::int AS num', [7], (err, res) => (err ? reject(err) : resolve(res))) + }) + const withoutValues = await new Promise((resolve, reject) => { + pool.query('SELECT 8 AS num', (err, res) => (err ? reject(err) : resolve(res))) + }) + + expect(withValues.rows[0].num).to.equal(7) + expect(withoutValues.rows[0].num).to.equal(8) + await pool.end() + }) + + it('pipelines named prepared statements', async () => { + const pool = await warm({ max: 1, pipeline: true }) + const nums = Array.from({ length: 20 }, (_, i) => i) + const results = await Promise.all( + nums.map((num) => pool.query({ name: 'pipeline-test', text: 'SELECT $1::int AS num', values: [num] })) + ) + + expect(results.map((res) => res.rows[0].num)).to.eql(nums) + await pool.end() + }) + + it('keeps the connection when a query is rejected before it is written', async () => { + const pool = await warm({ max: 1, pipeline: true }) + await pool.query({ text: 'SELECT $1::int', values: 'not an array' }).then( + () => { + throw new Error('expected the query to fail') + }, + (err) => expect(err.message).to.contain('must be an array') + ) + + expect((await pool.query('SELECT 1 AS num')).rows[0].num).to.equal(1) + expect(pool.totalCount).to.equal(1) + expect(pool.idleCount).to.equal(1) + await pool.end() + }) + + it('stays usable when a query fails while it is being written', async () => { + const circular = {} + circular.self = circular + const pool = await warm({ max: 1, pipeline: true }) + await pool.query('SELECT $1::text', [circular]).then( + () => { + throw new Error('expected the query to fail') + }, + (err) => expect(err.message).to.contain('circular') + ) + + expect((await pool.query('SELECT 1 AS num')).rows[0].num).to.equal(1) + await pool.end() + }) + + it('refuses a submittable instead of holding the connection', async () => { + const pool = await warm({ max: 1, pipeline: true }) + await pool.query(new Cursor('SELECT * FROM generate_series(0, 10)')).then( + () => { + throw new Error('expected the query to be refused') + }, + (err) => expect(err.message).to.contain('pool.connect()') + ) + + expect((await pool.query('SELECT 1 AS num')).rows[0].num).to.equal(1) + expect(pool.idleCount).to.equal(1) + await pool.end() + }) + + it('waits for the queries in flight when maxUses drops the connection', async () => { + const pool = new Pool({ max: 1, maxUses: 1, pipeline: true }) + const query = pool.query('SELECT pg_sleep(0.3), 1 AS num') + await wait(60) + + const started = Date.now() + await pool.end() + expect(Date.now() - started).to.be.greaterThan(150) + expect((await query).rows[0].num).to.equal(1) + }) + + it('does not hold more connections than max while recycling', async () => { + const application_name = 'pipeline-recycle-test' + const spy = new Pool({ max: 1 }) + const pool = new Pool({ max: 2, maxUses: 3, pipeline: true, application_name }) + const count = async () => { + const res = await spy.query('SELECT count(*)::int AS c FROM pg_stat_activity WHERE application_name = $1', [ + application_name, + ]) + return res.rows[0].c + } + + let peak = 0 + const watch = setInterval(() => count().then((c) => (peak = Math.max(peak, c))), 10) + for (let round = 0; round < 5; round++) { + await Promise.all(Array.from({ length: 4 }, () => pool.query('SELECT pg_sleep(0.1)'))) + } + clearInterval(watch) + + expect(peak).to.be.lessThan(3) + await pool.end() + await spy.end() + }) + + it('recycles connections on maxUses while pipelining', async () => { + const pool = await warm({ max: 1, maxUses: 3, pipeline: true }) + const removed = [] + pool.on('remove', () => removed.push(1)) + + const nums = Array.from({ length: 10 }, (_, i) => i) + const results = await Promise.all(nums.map((num) => pool.query('SELECT $1::int AS num', [num]))) + expect(results.map((res) => res.rows[0].num)).to.eql(nums) + + await wait(100) + expect(removed.length).to.be.greaterThan(1) + await pool.end() + }) + + it('answers in order on a single connection', async () => { + const pool = await warm({ max: 1, maxPipeline: 100, pipeline: true }) + const nums = Array.from({ length: 100 }, (_, i) => i) + const order = [] + await Promise.all( + nums.map((num) => pool.query('SELECT $1::int AS num', [num]).then((res) => order.push(res.rows[0].num))) + ) + + expect(order).to.eql(nums) + await pool.end() + }) + + it('runs the connection hooks once per connection, not per query', async () => { + let connects = 0 + let verifies = 0 + const pool = new Pool({ + max: 2, + pipeline: true, + onConnect: (client) => client.query("SET application_name = 'pipeline-test'").then(() => connects++), + verify: (client, cb) => { + verifies++ + cb() + }, + }) + const nums = Array.from({ length: 20 }, (_, i) => i) + const results = await Promise.all(nums.map((num) => pool.query('SELECT $1::int AS num', [num]))) + + expect(results.map((res) => res.rows[0].num)).to.eql(nums) + expect(connects).to.equal(pool.totalCount) + expect(verifies).to.equal(pool.totalCount) + expect((await pool.query('SHOW application_name')).rows[0].application_name).to.equal('pipeline-test') + await pool.end() + }) + + it('times out a query that waits too long for a connection', async () => { + const pool = await warm({ max: 1, maxPipeline: 1, connectionTimeoutMillis: 60, pipeline: true }) + const slow = pool.query('SELECT pg_sleep(0.3)') + await pool.query('SELECT 1 AS num').then( + () => { + throw new Error('expected the query to time out') + }, + (err) => expect(err.message).to.contain('timeout exceeded') + ) + + await slow + expect((await pool.query('SELECT 2 AS num')).rows[0].num).to.equal(2) + expect(pool.waitingCount).to.equal(0) + await pool.end() + }) + + it('rotates a pipelining connection on maxLifetimeSeconds', async () => { + const pool = await warm({ max: 1, maxLifetimeSeconds: 1, pipeline: true }) + const first = await pool.query('SELECT pg_backend_pid() AS pid') + await wait(1400) + const second = await pool.query('SELECT pg_backend_pid() AS pid') + + expect(second.rows[0].pid).to.not.equal(first.rows[0].pid) + await pool.end() + }) + + it('keeps a transaction on a connection of its own', async () => { + const pool = await warm({ max: 2, pipeline: true }) + const client = await pool.connect() + await client.query('BEGIN') + const inside = await client.query('SELECT pg_backend_pid() AS pid') + + const outside = await Promise.all([ + pool.query('SELECT pg_backend_pid() AS pid'), + pool.query('SELECT pg_backend_pid() AS pid'), + ]) + expect(outside.map((res) => res.rows[0].pid)).to.not.contain(inside.rows[0].pid) + + await client.query('COMMIT') + client.release() + await pool.end() + }) + it('gives pool.connect a connection nobody else is using', async () => { const pool = await warm({ max: 1, pipeline: true }) const query = pool.query('SELECT pg_sleep(0.2)') @@ -91,6 +298,16 @@ describe('pipeline', () => { await pool.end() }) + it('removes a connection that goes idle after a long query', async () => { + const pool = await warm({ max: 1, idleTimeoutMillis: 100, pipeline: true }) + await pool.query('SELECT pg_sleep(0.15)') + expect(pool.totalCount).to.equal(1) + + await wait(250) + expect(pool.totalCount).to.equal(0) + await pool.end() + }) + it('reports a broken connection to every query on it', async () => { const pool = new Pool({ max: 1, pipeline: true }) let client @@ -109,6 +326,22 @@ describe('pipeline', () => { await pool.end() }) + it('lets the program exit when allowExitOnIdle is set', function (done) { + const child = fork(path.join(__dirname, 'idle-timeout-exit.js'), [], { + stdio: ['ignore', 'pipe', 'inherit', 'ipc'], + env: { ...process.env, ALLOW_EXIT_ON_IDLE: '1', PIPELINE: '1' }, + }) + let result = '' + child.stdout.setEncoding('utf8') + child.stdout.on('data', (chunk) => (result += chunk)) + child.on('error', done) + child.on('exit', (exitCode) => { + expect(exitCode).to.equal(0) + expect(result).to.equal('completed first\ncompleted second\n') + done() + }) + }) + it('waits for the queries in flight on end', async () => { const pool = await warm({ max: 1, pipeline: true }) const queries = [1, 2, 3].map((num) => pool.query('SELECT pg_sleep(0.1), $1::int AS num', [num])) From 733233a0f6f98527579fd937ccce39a763f627a0 Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Mon, 10 Aug 2026 07:11:57 +0200 Subject: [PATCH 3/7] Document how connectionTimeoutMillis behaves with a pipelining pool --- docs/pages/features/pipelining.mdx | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/pages/features/pipelining.mdx b/docs/pages/features/pipelining.mdx index 6524361bf..4f2997eb7 100644 --- a/docs/pages/features/pipelining.mdx +++ b/docs/pages/features/pipelining.mdx @@ -86,8 +86,9 @@ try { `pool.query()` refuses a submittable (`pg-cursor`, `pg-query-stream`) while pipelining, because those read their rows over several round trips and cannot share a connection. Check out a client -with `pool.connect()` for them. `connectionTimeoutMillis` also covers the wait for a free pipeline -slot, not only the wait for a connection. +with `pool.connect()` for them. `connectionTimeoutMillis` counts the wait for a pipeline slot too, +so a query can time out with a connect error while the pool has a healthy connection sitting at +`maxPipeline`. ## Error isolation From 31452c83d200e37422c3fe5fb7a866c5b7f1c38b Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Mon, 10 Aug 2026 07:22:02 +0200 Subject: [PATCH 4/7] Emit remove only once when a marked client dies before its last result --- packages/pg-pool/README.md | 2 ++ packages/pg-pool/index.js | 2 ++ packages/pg-pool/test/pipeline.js | 20 ++++++++++++++++++++ 3 files changed, 24 insertions(+) diff --git a/packages/pg-pool/README.md b/packages/pg-pool/README.md index 80c644788..5b9948d95 100644 --- a/packages/pg-pool/README.md +++ b/packages/pg-pool/README.md @@ -36,6 +36,8 @@ const pool2 = new Pool({ idleTimeoutMillis: 1000, // close idle clients after 1 second connectionTimeoutMillis: 1000, // return an error after 1 second if connection could not be established maxUses: 7500, // close (and replace) a connection after it has been used 7500 times (see below for discussion) + pipeline: false, // when true, pool.query() can send a query on a connection that is still waiting for results + maxPipeline: 10, // queries pool.query() may send on the same connection before waiting, only used with pipeline: true }) // you can supply a custom client constructor diff --git a/packages/pg-pool/index.js b/packages/pg-pool/index.js index a0557c38c..b7bba9850 100644 --- a/packages/pg-pool/index.js +++ b/packages/pg-pool/index.js @@ -236,6 +236,8 @@ class Pool extends EventEmitter { } _remove(client, callback) { + // the deferred removal is done here, don't repeat it when the last answer lands + client._poolRemoveWhenIdle = false const removed = removeWhere(this._idle, (item) => item.client === client) if (removed !== undefined) { diff --git a/packages/pg-pool/test/pipeline.js b/packages/pg-pool/test/pipeline.js index 0c0fc2bc9..2148511e7 100644 --- a/packages/pg-pool/test/pipeline.js +++ b/packages/pg-pool/test/pipeline.js @@ -326,6 +326,26 @@ describe('pipeline', () => { await pool.end() }) + it('removes a client only once when it dies while waiting for removal', async () => { + const pool = new Pool({ max: 1, maxUses: 1, pipeline: true }) + let client + const removed = [] + pool.once('connect', (c) => (client = c)) + pool.on('remove', (c) => removed.push(c)) + + const queries = [1, 2].map(() => pool.query('SELECT pg_sleep(0.3)')) + await wait(100) + // released once, so maxUses already marked it for removal with the query in flight + client.connection.stream.destroy() + + await Promise.allSettled(queries) + await wait(100) + expect(removed.filter((c) => c === client).length).to.equal(1) + + expect((await pool.query('SELECT 1 AS num')).rows[0].num).to.equal(1) + await pool.end() + }) + it('lets the program exit when allowExitOnIdle is set', function (done) { const child = fork(path.join(__dirname, 'idle-timeout-exit.js'), [], { stdio: ['ignore', 'pipe', 'inherit', 'ipc'], From 7189b680ccdc6f9a3242116475c059cc4f6ce6ef Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Mon, 10 Aug 2026 07:45:22 +0200 Subject: [PATCH 5/7] Fix maxLifetimeSeconds rotation and result routing after a failed write in pipeline mode --- docs/pages/apis/pool.mdx | 4 ++++ packages/pg-pool/index.js | 17 ++++++++++++++- packages/pg-pool/test/pipeline.js | 35 +++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 1 deletion(-) diff --git a/docs/pages/apis/pool.mdx b/docs/pages/apis/pool.mdx index 3686da023..fd086c116 100644 --- a/docs/pages/apis/pool.mdx +++ b/docs/pages/apis/pool.mdx @@ -238,6 +238,10 @@ The number of clients which are not checked out but are currently idle in the po The number of queued requests waiting on a client when all clients are checked out. It can be helpful to monitor this number to see if you need to adjust the size of the pool. +With `pipeline: true` a connection is counted as idle as soon as its queries are written, not when +the results arrive, so `idleCount` can be positive while every connection is working and +`waitingCount` grows once they are all at `maxPipeline`. + ## events `Pool` instances are also instances of [`EventEmitter`](https://nodejs.org/api/events.html). diff --git a/packages/pg-pool/index.js b/packages/pg-pool/index.js index b7bba9850..48cbbf3ba 100644 --- a/packages/pg-pool/index.js +++ b/packages/pg-pool/index.js @@ -461,7 +461,15 @@ class Pool extends EventEmitter { this.emit('release', err, client) // TODO(bmc): expose a proper, public interface _queryable and _ending - if (err || this.ending || !client._queryable || client._ending || client._poolUseCount >= this.options.maxUses) { + if ( + err || + this.ending || + !client._queryable || + client._ending || + client._poolUseCount >= this.options.maxUses || + // already condemned by an earlier release, keep it out of _idle or it takes new queries forever + client._poolRemoveWhenIdle + ) { if (client._poolUseCount >= this.options.maxUses) { this.log('remove expended client') } @@ -602,8 +610,14 @@ class Pool extends EventEmitter { client._poolPipelined = inFlight(client) + 1 this.log('dispatching query') + let written = false let answered = false const answer = (err, res) => { + // an error while the query is being written answers on the submit stack. finish the write + // first: releasing here would dispatch the next query ahead of this one on the client queue + if (!written) { + return process.nextTick(answer, err, res) + } // a query that fails while it is being written is answered twice by the client if (answered) { return @@ -630,6 +644,7 @@ class Pool extends EventEmitter { try { client.query(text, values, answer) + written = true } catch (err) { answered = true client._poolPipelined-- diff --git a/packages/pg-pool/test/pipeline.js b/packages/pg-pool/test/pipeline.js index 2148511e7..4fe5b7c1a 100644 --- a/packages/pg-pool/test/pipeline.js +++ b/packages/pg-pool/test/pipeline.js @@ -107,6 +107,41 @@ describe('pipeline', () => { await pool.end() }) + it('delivers the right rows to the queries behind a failed write', async () => { + const circular = {} + circular.self = circular + const pool = await warm({ max: 1, pipeline: true }) + const slow = pool.query('SELECT pg_sleep(0.2)') + const bad = pool.query('SELECT $1::text AS bad', [circular]) + const third = pool.query('SELECT 333 AS num') + const fourth = pool.query('SELECT 444 AS num') + + await bad.then( + () => { + throw new Error('expected the query to fail') + }, + (err) => expect(err.message).to.contain('circular') + ) + expect((await third).rows).to.eql([{ num: 333 }]) + expect((await fourth).rows).to.eql([{ num: 444 }]) + await slow + await pool.end() + }) + + it('rotates the connection on maxLifetimeSeconds under constant load', async function () { + this.timeout(5000) + const pool = await warm({ max: 1, maxLifetimeSeconds: 1, pipeline: true }) + const pids = new Set() + const until = Date.now() + 2500 + while (Date.now() < until) { + const res = await pool.query('SELECT pg_backend_pid() AS pid, pg_sleep(0.02)') + pids.add(res.rows[0].pid) + } + + expect(pids.size).to.be.greaterThan(1) + await pool.end() + }) + it('stays usable when a query fails while it is being written', async () => { const circular = {} circular.self = circular From 39fce3b0c4d142715a70eaecec0c6c2d8dc9d92c Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Mon, 10 Aug 2026 07:48:28 +0200 Subject: [PATCH 6/7] Cover routing, events and connection failures in the pipeline tests --- packages/pg-pool/test/pipeline.js | 56 +++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/packages/pg-pool/test/pipeline.js b/packages/pg-pool/test/pipeline.js index 4fe5b7c1a..16b3274f1 100644 --- a/packages/pg-pool/test/pipeline.js +++ b/packages/pg-pool/test/pipeline.js @@ -279,6 +279,58 @@ describe('pipeline', () => { await pool.end() }) + it('spreads queries over the connection with the fewest in flight', async () => { + const pool = await warm({ max: 2, pipeline: true }) + // occupy both connections once so they are both open and idle + await Promise.all([pool.query('SELECT pg_sleep(0.05)'), pool.query('SELECT pg_sleep(0.05)')]) + + const results = await Promise.all( + Array.from({ length: 6 }, () => pool.query('SELECT pg_backend_pid() AS pid, pg_sleep(0.15)')) + ) + const perPid = {} + results.forEach((res) => (perPid[res.rows[0].pid] = (perPid[res.rows[0].pid] || 0) + 1)) + + expect(Object.values(perPid)).to.eql([3, 3]) + await pool.end() + }) + + it('emits acquire and release once per query', async () => { + const pool = await warm({ max: 2, pipeline: true }) + let acquires = 0 + let releases = 0 + pool.on('acquire', () => acquires++) + pool.on('release', () => releases++) + + await Promise.all(Array.from({ length: 10 }, (_, i) => pool.query('SELECT $1::int AS num', [i]))) + expect(acquires).to.equal(10) + expect(releases).to.equal(10) + await pool.end() + }) + + it('rejects the query when the connection cannot be made', async () => { + const pool = new Pool({ port: 1, host: 'localhost', pipeline: true, connectionTimeoutMillis: 2000 }) + await pool.query('SELECT 1').then( + () => { + throw new Error('expected the query to fail') + }, + (err) => expect(err).to.be.an(Error) + ) + expect(pool.totalCount).to.equal(0) + await pool.end() + }) + + it('rejects the query when the verify hook fails', async () => { + const pool = new Pool({ pipeline: true, verify: (client, cb) => cb(new Error('verify says no')) }) + await pool.query('SELECT 1').then( + () => { + throw new Error('expected the query to fail') + }, + (err) => expect(err.message).to.equal('verify says no') + ) + expect(pool.totalCount).to.equal(0) + await pool.end() + }) + it('keeps a transaction on a connection of its own', async () => { const pool = await warm({ max: 2, pipeline: true }) const client = await pool.connect() @@ -346,7 +398,9 @@ describe('pipeline', () => { it('reports a broken connection to every query on it', async () => { const pool = new Pool({ max: 1, pipeline: true }) let client + const poolErrors = [] pool.once('connect', (c) => (client = c)) + pool.on('error', (err) => poolErrors.push(err)) await pool.query('SELECT 1') const queries = [1, 2, 3].map(() => pool.query('SELECT pg_sleep(0.2)')) @@ -356,6 +410,8 @@ describe('pipeline', () => { const results = await Promise.allSettled(queries) expect(results.map((res) => res.status)).to.eql(['rejected', 'rejected', 'rejected']) expect(pool.totalCount).to.equal(0) + // the queries got the error on their own callbacks, the pool must not repeat it + expect(poolErrors).to.have.length(0) expect((await pool.query('SELECT 1 AS num')).rows[0].num).to.equal(1) await pool.end() From aedb45ebb47ae6547c4bf13b9bfc3e7d79acd90d Mon Sep 17 00:00:00 2001 From: Nigro Simone Date: Wed, 12 Aug 2026 06:30:48 +0200 Subject: [PATCH 7/7] Drive pool pipelining from maxPipeline instead of overloading the client pipeline option --- docs/pages/apis/pool.mdx | 16 +++--- docs/pages/features/pipelining.mdx | 30 ++++++++--- packages/pg-pool/README.md | 3 +- packages/pg-pool/index.js | 13 +++-- packages/pg-pool/test/idle-timeout-exit.js | 2 +- packages/pg-pool/test/pipeline.js | 60 +++++++++++----------- 6 files changed, 73 insertions(+), 51 deletions(-) diff --git a/docs/pages/apis/pool.mdx b/docs/pages/apis/pool.mdx index fd086c116..ab04f63fb 100644 --- a/docs/pages/apis/pool.mdx +++ b/docs/pages/apis/pool.mdx @@ -31,7 +31,7 @@ type Config = { // Maximum number of clients the pool should contain. // By default this is set to 10. There is some nuance to setting the maximum size of your pool. // See https://node-postgres.com/guides/pool-sizing for more information. - // With `pipeline: true` this is still the number of connections, but the number of queries + // With `maxPipeline` this is still the number of connections, but the number of queries // in flight can reach max * maxPipeline. max?: number @@ -73,15 +73,15 @@ type Config = { onConnect?: (client: Client) => void | Promise // When set to true, enables query pipelining on every client the pool creates. - // Pipelined clients send queries to the server without waiting for previous responses, and - // pool.query() can use a connection that is already working instead of waiting for a free one. - // pool.connect() is not affected, it still checks out a connection nobody else is using. - // Queries can complete in a different order, so the default is false. - // See /features/pipelining for details. + // Pipelined clients send queries to the server without waiting for previous responses. + // Default is false. See /features/pipelining for details. pipeline?: boolean // Maximum number of queries pool.query() sends on the same connection before waiting. - // Only used with `pipeline: true`. The default is 10. + // The default is 1: a connection serves one query at a time. A higher value lets pool.query() + // use a connection that is already working instead of waiting for a free one, and implies + // pipeline: true. pool.connect() is not affected, it still checks out a connection nobody else + // is using. Queries can complete in a different order, so this is off by default. maxPipeline?: number } ``` @@ -238,7 +238,7 @@ The number of clients which are not checked out but are currently idle in the po The number of queued requests waiting on a client when all clients are checked out. It can be helpful to monitor this number to see if you need to adjust the size of the pool. -With `pipeline: true` a connection is counted as idle as soon as its queries are written, not when +With `maxPipeline` a connection is counted as idle as soon as its queries are written, not when the results arrive, so `idleCount` can be positive while every connection is working and `waitingCount` grows once they are all at `maxPipeline`. diff --git a/docs/pages/features/pipelining.mdx b/docs/pages/features/pipelining.mdx index 4f2997eb7..38ed5c6d8 100644 --- a/docs/pages/features/pipelining.mdx +++ b/docs/pages/features/pipelining.mdx @@ -46,12 +46,29 @@ All query types work with pipelining: plain text, parameterized, and named prepa ## Pipelining with a pool -Pass `pipeline: true` in the pool config to enable it on every client the pool creates: +Pass `pipeline: true` in the pool config to enable it on every client the pool creates. Each client +pipelines the queries you send on it after `pool.connect()`, `pool.query()` works as usual: ```js import { Pool } from 'pg' -const pool = new Pool({ max: 10, pipeline: true, maxPipeline: 10 }) +const pool = new Pool({ pipeline: true }) + +const client = await pool.connect() +// client.pipeline is already true +const [users, orders] = await Promise.all([ + client.query('SELECT * FROM users WHERE id = $1', [1]), + client.query('SELECT * FROM orders WHERE user_id = $1', [1]), +]) +client.release() +``` + +## Pipelining pool.query() + +Set `maxPipeline` to more than 1 to let `pool.query()` pipeline too: + +```js +const pool = new Pool({ max: 10, maxPipeline: 10 }) const [users, orders] = await Promise.all([ pool.query('SELECT * FROM users WHERE id = $1', [1]), @@ -59,10 +76,11 @@ const [users, orders] = await Promise.all([ ]) ``` -Without pipelining `pool.query()` holds a connection for one query, so a query has to wait for a -free connection. With `pipeline: true` the pool sends it on a connection that is already working, -once every connection is busy. It opens connections up to `max` first, then picks the one with the -fewest queries in flight, up to `maxPipeline` per connection (default 10). +By default (`maxPipeline: 1`) `pool.query()` holds a connection for one query, so a query has to +wait for a free connection. With a higher `maxPipeline` the pool sends it on a connection that is +already working, once every connection is busy. It opens connections up to `max` first, then picks +the one with the fewest queries in flight, up to `maxPipeline` per connection. The clients it +creates are pipelining clients, so `pipeline: true` is implied. `max` is still the number of connections. It is not the number of queries in flight anymore, those can reach `max * maxPipeline`. This does not add load on the server, PostgreSQL runs a pipeline diff --git a/packages/pg-pool/README.md b/packages/pg-pool/README.md index 5b9948d95..80b5812e0 100644 --- a/packages/pg-pool/README.md +++ b/packages/pg-pool/README.md @@ -36,8 +36,7 @@ const pool2 = new Pool({ idleTimeoutMillis: 1000, // close idle clients after 1 second connectionTimeoutMillis: 1000, // return an error after 1 second if connection could not be established maxUses: 7500, // close (and replace) a connection after it has been used 7500 times (see below for discussion) - pipeline: false, // when true, pool.query() can send a query on a connection that is still waiting for results - maxPipeline: 10, // queries pool.query() may send on the same connection before waiting, only used with pipeline: true + maxPipeline: 1, // queries pool.query() may send on the same connection before waiting, more than 1 lets it use a connection that is still working }) // you can supply a custom client constructor diff --git a/packages/pg-pool/index.js b/packages/pg-pool/index.js index 48cbbf3ba..6f96d0d24 100644 --- a/packages/pg-pool/index.js +++ b/packages/pg-pool/index.js @@ -109,8 +109,13 @@ class Pool extends EventEmitter { this.options.max = this.options.max || this.options.poolSize || 10 this.options.min = this.options.min || 0 // max stays the number of connections, queries in flight can reach max * maxPipeline - this.options.pipeline = Boolean(this.options.pipeline) - this.options.maxPipeline = this.options.maxPipeline || 10 + this.options.maxPipeline = this.options.maxPipeline || 1 + // maxPipeline > 1 is what turns pool.query() pipelining on, and it needs pipelining clients. + // `pipeline` keeps its own meaning, it is passed to the Client as any other option + this._pipeline = this.options.maxPipeline > 1 + if (this._pipeline) { + this.options.pipeline = true + } this.options.maxUses = this.options.maxUses || Infinity this.options.allowExitOnIdle = this.options.allowExitOnIdle || false this.options.maxLifetimeSeconds = this.options.maxLifetimeSeconds || 0 @@ -200,7 +205,7 @@ class Pool extends EventEmitter { if (!this._idle.length) { return undefined } - if (!this.options.pipeline) { + if (!this._pipeline) { return this._idle.pop() } @@ -539,7 +544,7 @@ class Pool extends EventEmitter { const response = promisify(this.Promise, cb) cb = response.callback - if (this.options.pipeline) { + if (this._pipeline) { this._pipelineQuery(text, values, cb) return response.result } diff --git a/packages/pg-pool/test/idle-timeout-exit.js b/packages/pg-pool/test/idle-timeout-exit.js index 54591cc84..e869055c8 100644 --- a/packages/pg-pool/test/idle-timeout-exit.js +++ b/packages/pg-pool/test/idle-timeout-exit.js @@ -6,7 +6,7 @@ if (module === require.main) { const pool = new Pool({ maxLifetimeSeconds: 2, idleTimeoutMillis: 200, - pipeline: process.env.PIPELINE === '1', + ...(process.env.PIPELINE === '1' ? { maxPipeline: 10 } : {}), ...(allowExitOnIdle ? { allowExitOnIdle: true } : {}), }) pool.query('SELECT NOW()', (err, res) => console.log('completed first')) diff --git a/packages/pg-pool/test/pipeline.js b/packages/pg-pool/test/pipeline.js index 16b3274f1..d841e54ba 100644 --- a/packages/pg-pool/test/pipeline.js +++ b/packages/pg-pool/test/pipeline.js @@ -19,7 +19,7 @@ const warm = async (options) => { describe('pipeline', () => { it('sends a query on a connection that is already working', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) const slow = pool.query('SELECT pg_sleep(0.3)') const fast = pool.query('SELECT 1 AS num') await wait(50) @@ -33,7 +33,7 @@ describe('pipeline', () => { }) it('opens connections up to max before pipelining', async () => { - const pool = await warm({ max: 3, pipeline: true }) + const pool = await warm({ max: 3, maxPipeline: 10 }) const queries = [1, 2, 3, 4, 5, 6].map((num) => pool.query('SELECT pg_sleep(0.2), $1::int AS num', [num])) await wait(50) @@ -46,7 +46,7 @@ describe('pipeline', () => { }) it('does not send more than maxPipeline on one connection', async () => { - const pool = await warm({ max: 1, maxPipeline: 2, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 2 }) const queries = [1, 2, 3, 4, 5].map((num) => pool.query('SELECT pg_sleep(0.1), $1::int AS num', [num])) await wait(50) @@ -58,7 +58,7 @@ describe('pipeline', () => { }) it('answers every query of a large burst', async () => { - const pool = await warm({ max: 2, pipeline: true }) + const pool = await warm({ max: 2, maxPipeline: 10 }) const nums = Array.from({ length: 500 }, (_, i) => i) const results = await Promise.all(nums.map((num) => pool.query('SELECT $1::int AS num', [num]))) @@ -68,7 +68,7 @@ describe('pipeline', () => { }) it('accepts the callback and the values forms', async () => { - const pool = new Pool({ max: 1, pipeline: true }) + const pool = new Pool({ max: 1, maxPipeline: 10 }) const withValues = await new Promise((resolve, reject) => { pool.query('SELECT $1::int AS num', [7], (err, res) => (err ? reject(err) : resolve(res))) }) @@ -82,7 +82,7 @@ describe('pipeline', () => { }) it('pipelines named prepared statements', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) const nums = Array.from({ length: 20 }, (_, i) => i) const results = await Promise.all( nums.map((num) => pool.query({ name: 'pipeline-test', text: 'SELECT $1::int AS num', values: [num] })) @@ -93,7 +93,7 @@ describe('pipeline', () => { }) it('keeps the connection when a query is rejected before it is written', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) await pool.query({ text: 'SELECT $1::int', values: 'not an array' }).then( () => { throw new Error('expected the query to fail') @@ -110,7 +110,7 @@ describe('pipeline', () => { it('delivers the right rows to the queries behind a failed write', async () => { const circular = {} circular.self = circular - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) const slow = pool.query('SELECT pg_sleep(0.2)') const bad = pool.query('SELECT $1::text AS bad', [circular]) const third = pool.query('SELECT 333 AS num') @@ -130,7 +130,7 @@ describe('pipeline', () => { it('rotates the connection on maxLifetimeSeconds under constant load', async function () { this.timeout(5000) - const pool = await warm({ max: 1, maxLifetimeSeconds: 1, pipeline: true }) + const pool = await warm({ max: 1, maxLifetimeSeconds: 1, maxPipeline: 10 }) const pids = new Set() const until = Date.now() + 2500 while (Date.now() < until) { @@ -145,7 +145,7 @@ describe('pipeline', () => { it('stays usable when a query fails while it is being written', async () => { const circular = {} circular.self = circular - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) await pool.query('SELECT $1::text', [circular]).then( () => { throw new Error('expected the query to fail') @@ -158,7 +158,7 @@ describe('pipeline', () => { }) it('refuses a submittable instead of holding the connection', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) await pool.query(new Cursor('SELECT * FROM generate_series(0, 10)')).then( () => { throw new Error('expected the query to be refused') @@ -172,7 +172,7 @@ describe('pipeline', () => { }) it('waits for the queries in flight when maxUses drops the connection', async () => { - const pool = new Pool({ max: 1, maxUses: 1, pipeline: true }) + const pool = new Pool({ max: 1, maxUses: 1, maxPipeline: 10 }) const query = pool.query('SELECT pg_sleep(0.3), 1 AS num') await wait(60) @@ -185,7 +185,7 @@ describe('pipeline', () => { it('does not hold more connections than max while recycling', async () => { const application_name = 'pipeline-recycle-test' const spy = new Pool({ max: 1 }) - const pool = new Pool({ max: 2, maxUses: 3, pipeline: true, application_name }) + const pool = new Pool({ max: 2, maxUses: 3, maxPipeline: 10, application_name }) const count = async () => { const res = await spy.query('SELECT count(*)::int AS c FROM pg_stat_activity WHERE application_name = $1', [ application_name, @@ -206,7 +206,7 @@ describe('pipeline', () => { }) it('recycles connections on maxUses while pipelining', async () => { - const pool = await warm({ max: 1, maxUses: 3, pipeline: true }) + const pool = await warm({ max: 1, maxUses: 3, maxPipeline: 10 }) const removed = [] pool.on('remove', () => removed.push(1)) @@ -220,7 +220,7 @@ describe('pipeline', () => { }) it('answers in order on a single connection', async () => { - const pool = await warm({ max: 1, maxPipeline: 100, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 100 }) const nums = Array.from({ length: 100 }, (_, i) => i) const order = [] await Promise.all( @@ -236,7 +236,7 @@ describe('pipeline', () => { let verifies = 0 const pool = new Pool({ max: 2, - pipeline: true, + maxPipeline: 10, onConnect: (client) => client.query("SET application_name = 'pipeline-test'").then(() => connects++), verify: (client, cb) => { verifies++ @@ -254,8 +254,8 @@ describe('pipeline', () => { }) it('times out a query that waits too long for a connection', async () => { - const pool = await warm({ max: 1, maxPipeline: 1, connectionTimeoutMillis: 60, pipeline: true }) - const slow = pool.query('SELECT pg_sleep(0.3)') + const pool = await warm({ max: 1, maxPipeline: 2, connectionTimeoutMillis: 60 }) + const slow = Promise.all([pool.query('SELECT pg_sleep(0.3)'), pool.query('SELECT pg_sleep(0.3)')]) await pool.query('SELECT 1 AS num').then( () => { throw new Error('expected the query to time out') @@ -270,7 +270,7 @@ describe('pipeline', () => { }) it('rotates a pipelining connection on maxLifetimeSeconds', async () => { - const pool = await warm({ max: 1, maxLifetimeSeconds: 1, pipeline: true }) + const pool = await warm({ max: 1, maxLifetimeSeconds: 1, maxPipeline: 10 }) const first = await pool.query('SELECT pg_backend_pid() AS pid') await wait(1400) const second = await pool.query('SELECT pg_backend_pid() AS pid') @@ -280,7 +280,7 @@ describe('pipeline', () => { }) it('spreads queries over the connection with the fewest in flight', async () => { - const pool = await warm({ max: 2, pipeline: true }) + const pool = await warm({ max: 2, maxPipeline: 10 }) // occupy both connections once so they are both open and idle await Promise.all([pool.query('SELECT pg_sleep(0.05)'), pool.query('SELECT pg_sleep(0.05)')]) @@ -295,7 +295,7 @@ describe('pipeline', () => { }) it('emits acquire and release once per query', async () => { - const pool = await warm({ max: 2, pipeline: true }) + const pool = await warm({ max: 2, maxPipeline: 10 }) let acquires = 0 let releases = 0 pool.on('acquire', () => acquires++) @@ -308,7 +308,7 @@ describe('pipeline', () => { }) it('rejects the query when the connection cannot be made', async () => { - const pool = new Pool({ port: 1, host: 'localhost', pipeline: true, connectionTimeoutMillis: 2000 }) + const pool = new Pool({ port: 1, host: 'localhost', maxPipeline: 10, connectionTimeoutMillis: 2000 }) await pool.query('SELECT 1').then( () => { throw new Error('expected the query to fail') @@ -320,7 +320,7 @@ describe('pipeline', () => { }) it('rejects the query when the verify hook fails', async () => { - const pool = new Pool({ pipeline: true, verify: (client, cb) => cb(new Error('verify says no')) }) + const pool = new Pool({ maxPipeline: 10, verify: (client, cb) => cb(new Error('verify says no')) }) await pool.query('SELECT 1').then( () => { throw new Error('expected the query to fail') @@ -332,7 +332,7 @@ describe('pipeline', () => { }) it('keeps a transaction on a connection of its own', async () => { - const pool = await warm({ max: 2, pipeline: true }) + const pool = await warm({ max: 2, maxPipeline: 10 }) const client = await pool.connect() await client.query('BEGIN') const inside = await client.query('SELECT pg_backend_pid() AS pid') @@ -349,7 +349,7 @@ describe('pipeline', () => { }) it('gives pool.connect a connection nobody else is using', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) const query = pool.query('SELECT pg_sleep(0.2)') await wait(50) @@ -368,7 +368,7 @@ describe('pipeline', () => { }) it('keeps the connection after a query error', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) const bad = pool.query('SELECT * FROM table_that_does_not_exist') const good = pool.query('SELECT 1 AS num') @@ -386,7 +386,7 @@ describe('pipeline', () => { }) it('removes a connection that goes idle after a long query', async () => { - const pool = await warm({ max: 1, idleTimeoutMillis: 100, pipeline: true }) + const pool = await warm({ max: 1, idleTimeoutMillis: 100, maxPipeline: 10 }) await pool.query('SELECT pg_sleep(0.15)') expect(pool.totalCount).to.equal(1) @@ -396,7 +396,7 @@ describe('pipeline', () => { }) it('reports a broken connection to every query on it', async () => { - const pool = new Pool({ max: 1, pipeline: true }) + const pool = new Pool({ max: 1, maxPipeline: 10 }) let client const poolErrors = [] pool.once('connect', (c) => (client = c)) @@ -418,7 +418,7 @@ describe('pipeline', () => { }) it('removes a client only once when it dies while waiting for removal', async () => { - const pool = new Pool({ max: 1, maxUses: 1, pipeline: true }) + const pool = new Pool({ max: 1, maxUses: 1, maxPipeline: 10 }) let client const removed = [] pool.once('connect', (c) => (client = c)) @@ -454,7 +454,7 @@ describe('pipeline', () => { }) it('waits for the queries in flight on end', async () => { - const pool = await warm({ max: 1, pipeline: true }) + const pool = await warm({ max: 1, maxPipeline: 10 }) const queries = [1, 2, 3].map((num) => pool.query('SELECT pg_sleep(0.1), $1::int AS num', [num])) await wait(50) expect(pool.waitingCount).to.equal(0)