diff --git a/docs/pages/apis/pool.mdx b/docs/pages/apis/pool.mdx index 3627dd1c5..fd086c116 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 } ``` @@ -229,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/docs/pages/features/pipelining.mdx b/docs/pages/features/pipelining.mdx index 7943aa490..4f2997eb7 100644 --- a/docs/pages/features/pipelining.mdx +++ b/docs/pages/features/pipelining.mdx @@ -51,24 +51,44 @@ 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]), ]) +``` + +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. -client.release() +`pool.connect()` is not affected: it still checks out a connection nobody else is using, so +transactions and `SET` 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() +} ``` - -
- 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. -
- +`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` 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 @@ -102,6 +122,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/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 ab514fa88..48cbbf3ba 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,23 +180,64 @@ 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 + } + 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] + } + + // 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 } - throw new Error('unexpected condition') + return this._remove(client, this._pulseQueue.bind(this)) } _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) { @@ -188,6 +256,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 +276,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 +285,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 +304,7 @@ class Pool extends EventEmitter { return result } - this.newClient(new PendingItem(response.callback)) + this.newClient(new PendingItem(response.callback, exclusive)) return result } @@ -389,38 +461,59 @@ 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') } - 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(() => { - if (this._isAboveMin()) { + 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) { + if (this.options.allowExitOnIdle && !inFlight(client)) { client.unref() } @@ -446,6 +539,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 +583,81 @@ 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) { + // 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) + } + + let released = false + const releaseOnce = (err) => { + if (!released) { + released = true + release(err) + } + } + + 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 + } + 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, answer) + written = true + } catch (err) { + answered = true + 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/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 new file mode 100644 index 000000000..16b3274f1 --- /dev/null +++ b/packages/pg-pool/test/pipeline.js @@ -0,0 +1,466 @@ +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') + +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('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('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 + 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('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() + 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)') + 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('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 + 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)')) + 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) + // 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() + }) + + 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'], + 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])) + 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]) + }) +})