diff --git a/.gitignore b/.gitignore index c6d5700ae..d17525a7b 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ dist .vscode/ manually-test-on-heroku.js tsconfig.tsbuildinfo +.history diff --git a/docs/pages/apis/client.mdx b/docs/pages/apis/client.mdx index 973b1a832..13cd56ae7 100644 --- a/docs/pages/apis/client.mdx +++ b/docs/pages/apis/client.mdx @@ -12,8 +12,8 @@ Every field of the `config` object is entirely optional. A `Client` instance wil type Config = { user?: string, // default process.env.PGUSER || process.env.USER password?: string or function, //default process.env.PGPASSWORD - host?: string, // default process.env.PGHOST - port?: number, // default process.env.PGPORT + host?: string | string[], // default process.env.PGHOST; array enables multi-host failover + port?: number | number[], // default process.env.PGPORT; one value or one per host database?: string, // default process.env.PGDATABASE || user connectionString?: string, // e.g. postgres://user:password@host:5432/database ssl?: any, // passed directly to node.TLSSocket, supports all tls.connect options @@ -31,6 +31,7 @@ type Config = { fallback_application_name?: string, // provide an application name to use if application_name is not set options?: string, // command-line options to be sent to the server pipeline?: boolean // when true, enables query pipelining. See /features/pipelining for details. Default false. + targetSessionAttrs?: 'any' | 'read-write' | 'read-only' | 'primary' | 'standby' | 'prefer-standby', // default 'any' } ``` diff --git a/docs/pages/features/connecting.mdx b/docs/pages/features/connecting.mdx index 97b5c779f..4a19cb29c 100644 --- a/docs/pages/features/connecting.mdx +++ b/docs/pages/features/connecting.mdx @@ -129,6 +129,80 @@ client = new Client({ }) ``` +## Multiple hosts + +The JavaScript driver supports connecting to multiple PostgreSQL hosts. Pass an array to `host` to try each host in order. `port` can be a single port or an array of ports. + +```js +import { Client } from 'pg' + +const client = new Client({ + host: ['primary.db.com', 'replica1.db.com', 'replica2.db.com'], + port: 5432, // single port reused for all hosts + database: 'mydb', + user: 'dbuser', + password: 'secretpassword', +}) + +await client.connect() // tries hosts left to right until one succeeds +``` + +Connection errors before TCP connects advance to the next host. Authentication and TLS errors stop the attempt. Once connected, the client stays on the selected host; it does not reconnect automatically if that connection is lost. `client.host` and `client.port` identify the selected endpoint. + +`connectionTimeoutMillis` limits the whole connection attempt, including all hosts and session checks. If you provide a custom `stream`, use a factory function so each attempt gets a fresh stream. + +You can also specify a different port for each host: + +```js +const client = new Client({ + host: ['host-a.db.com', 'host-b.db.com'], + port: [5432, 5433], + database: 'mydb', +}) +``` + +Host lists may mix TCP hosts and Unix socket directories. Each entry is interpreted independently: + +```js +const client = new Client({ + host: ['/var/run/postgresql', 'db.example.com'], + port: 5432, + database: 'mydb', +}) +``` + +For an absolute host path, the port is used as the Unix socket filename extension (`.s.PGSQL.5432`). Other host values use TCP. + +Port rules (same as libpq): +- **one port** — reused for every host +- **one port per host** — each port is paired with the corresponding host by index +- any other combination throws at construction time + +### target_session_attrs + +Use `targetSessionAttrs` to control which host is accepted based on its role. This mirrors the [libpq `target_session_attrs`](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-TARGET-SESSION-ATTRS) option. + +```js +const client = new Client({ + host: ['primary.db.com', 'replica.db.com'], + port: 5432, + targetSessionAttrs: 'read-write', // only connect to a writable primary +}) +``` + +| Value | Accepted server | +|---|---| +| `any` (default) | any server | +| `read-write` | server where `transaction_read_only = off` | +| `read-only` | server where `transaction_read_only = on` | +| `primary` | server that is not in hot standby | +| `standby` | server that is in hot standby | +| `prefer-standby` | standby if available, otherwise any | + +When all hosts are exhausted without finding a matching server, the client emits an error. + +Hosts that fail the session check are skipped. `prefer-standby` first tries every host for a standby, then retries from the beginning accepting any server if none matched. + ## Connection URI You can initialize both a pool and a client with a connection string URI as well. This is common in environments like Heroku where the database connection string is supplied to your application dyno through an environment variable. Connection string parsing brought to you by [pg-connection-string](https://github.com/brianc/node-postgres/tree/master/packages/pg-connection-string). diff --git a/packages/pg/lib/client.js b/packages/pg/lib/client.js index 2b13c1de7..00dd39595 100644 --- a/packages/pg/lib/client.js +++ b/packages/pg/lib/client.js @@ -8,6 +8,7 @@ const ConnectionParameters = require('./connection-parameters') const Query = require('./query') const defaults = require('./defaults') const Connection = require('./connection') +const connectMultiHost = require('./multihost') const crypto = require('./crypto/utils') const activeQueryDeprecationNotice = nodeUtils.deprecate( @@ -86,16 +87,22 @@ class Client extends EventEmitter { this.enableChannelBinding = Boolean(c.enableChannelBinding) // set true to use SCRAM-SHA-256-PLUS when offered this.scramMaxIterations = coerceNumberOrDefault(c.scramMaxIterations, sasl.DEFAULT_MAX_SCRAM_ITERATIONS) - this.connection = - c.connection || - new Connection({ - stream: c.stream, - ssl: this.connectionParameters.ssl, - sslNegotiation: this.connectionParameters.sslnegotiation, - keepAlive: c.keepAlive || false, - keepAliveInitialDelayMillis: c.keepAliveInitialDelayMillis || 0, - encoding: this.connectionParameters.client_encoding || 'utf8', - }) + const targetSessionAttrs = this.connectionParameters.targetSessionAttrs + const connectionConfig = { + stream: c.stream, + ssl: this.connectionParameters.ssl, + sslNegotiation: this.connectionParameters.sslnegotiation, + keepAlive: c.keepAlive || false, + keepAliveInitialDelayMillis: c.keepAliveInitialDelayMillis || 0, + encoding: this.connectionParameters.client_encoding || 'utf8', + } + const needsMultiHost = + Array.isArray(this.host) || + Array.isArray(this.port) || + Boolean(targetSessionAttrs && targetSessionAttrs !== 'any') + + this.connection = c.connection || new Connection(connectionConfig) + this._multiHostConfig = !c.connection && needsMultiHost ? connectionConfig : null this._queryQueue = [] this._sentQueryQueue = [] this.pipeline = Boolean(c.pipeline) @@ -151,8 +158,6 @@ class Client extends EventEmitter { } _connect(callback) { - const self = this - const con = this.connection this._connectionCallback = callback if (this._connecting || this._connected) { @@ -166,8 +171,8 @@ class Client extends EventEmitter { if (this._connectionTimeoutMillis > 0) { this.connectionTimeoutHandle = setTimeout(() => { - con._ending = true - con.stream.destroy(new Error('timeout expired')) + this.connection._ending = true + this.connection.stream.destroy(new Error('timeout expired')) }, this._connectionTimeoutMillis) if (this.connectionTimeoutHandle.unref) { @@ -175,10 +180,33 @@ class Client extends EventEmitter { } } - if (this.host && this.host.indexOf('/') === 0) { - con.connect(this.host + '/.s.PGSQL.' + this.port) + if (this._multiHostConfig) { + connectMultiHost(this, this._multiHostConfig).then( + (message) => { + if (!this._connectionError && !this._ended) { + this._handleReadyForQuery(message) + } + }, + (err) => { + this._handleErrorWhileConnecting(err) + this._handleConnectionEnd() + } + ) + return + } + + this._connectHost(this.port, this.host) + this._attachListeners(this.connection) + } + + _connectHost(port, host) { + const self = this + const con = this.connection + + if (typeof host === 'string' && host.startsWith('/')) { + con.connect(host.replace(/\/+$/, '') + '/.s.PGSQL.' + port) } else { - con.connect(this.port, this.host) + con.connect(port, host) } // once connection is established send startup message @@ -198,34 +226,43 @@ class Client extends EventEmitter { con.startup(self.getStartupConf()) }) - this._attachListeners(con) + // password request handling + con.on('authenticationCleartextPassword', this._handleAuthCleartextPassword.bind(this)) + // password request handling + con.on('authenticationMD5Password', this._handleAuthMD5Password.bind(this)) + // password request handling (SASL) + con.on('authenticationSASL', this._handleAuthSASL.bind(this)) + con.on('authenticationSASLContinue', this._handleAuthSASLContinue.bind(this)) + con.on('authenticationSASLFinal', this._handleAuthSASLFinal.bind(this)) + con.on('backendKeyData', this._handleBackendKeyData.bind(this)) + con.on('notice', this._handleNotice.bind(this)) + } - con.once('end', () => { - const error = this._ending ? new Error('Connection terminated') : new Error('Connection terminated unexpectedly') + _handleConnectionEnd() { + const error = this._ending ? new Error('Connection terminated') : new Error('Connection terminated unexpectedly') - clearTimeout(this.connectionTimeoutHandle) - this._errorAllQueries(error) - this._ended = true - - if (!this._ending) { - // if the connection is ended without us calling .end() - // on this client then we have an unexpected disconnection - // treat this as an error unless we've already emitted an error - // during connection. - if (this._connecting && !this._connectionError) { - if (this._connectionCallback) { - this._connectionCallback(error) - } else { - this._handleErrorEvent(error) - } - } else if (!this._connectionError) { + clearTimeout(this.connectionTimeoutHandle) + this._errorAllQueries(error) + this._ended = true + + if (!this._ending) { + // if the connection is ended without us calling .end() + // on this client then we have an unexpected disconnection + // treat this as an error unless we've already emitted an error + // during connection. + if (this._connecting && !this._connectionError) { + if (this._connectionCallback) { + this._connectionCallback(error) + } else { this._handleErrorEvent(error) } + } else if (!this._connectionError) { + this._handleErrorEvent(error) } + } - process.nextTick(() => { - this.emit('end') - }) + process.nextTick(() => { + this.emit('end') }) } @@ -247,19 +284,10 @@ class Client extends EventEmitter { } _attachListeners(con) { - // password request handling - con.on('authenticationCleartextPassword', this._handleAuthCleartextPassword.bind(this)) - // password request handling - con.on('authenticationMD5Password', this._handleAuthMD5Password.bind(this)) - // password request handling (SASL) - con.on('authenticationSASL', this._handleAuthSASL.bind(this)) - con.on('authenticationSASLContinue', this._handleAuthSASLContinue.bind(this)) - con.on('authenticationSASLFinal', this._handleAuthSASLFinal.bind(this)) - con.on('backendKeyData', this._handleBackendKeyData.bind(this)) + con.once('end', this._handleConnectionEnd.bind(this)) con.on('error', this._handleErrorEvent.bind(this)) con.on('errorMessage', this._handleErrorMessage.bind(this)) con.on('readyForQuery', this._handleReadyForQuery.bind(this)) - con.on('notice', this._handleNotice.bind(this)) con.on('rowDescription', this._handleRowDescription.bind(this)) con.on('dataRow', this._handleDataRow.bind(this)) con.on('portalSuspended', this._handlePortalSuspended.bind(this)) @@ -574,11 +602,13 @@ class Client extends EventEmitter { cancel(client, query) { if (client.activeQuery === query) { const con = this.connection + const host = client._multiHostConfig ? client.host : this.host + const port = client._multiHostConfig ? client.port : this.port - if (this.host && this.host.indexOf('/') === 0) { - con.connect(this.host + '/.s.PGSQL.' + this.port) + if (typeof host === 'string' && host.startsWith('/')) { + con.connect(host.replace(/\/+$/, '') + '/.s.PGSQL.' + port) } else { - con.connect(this.port, this.host) + con.connect(port, host) } // once connection is established send cancel message diff --git a/packages/pg/lib/connection-parameters.js b/packages/pg/lib/connection-parameters.js index 37987fd68..ff0c55431 100644 --- a/packages/pg/lib/connection-parameters.js +++ b/packages/pg/lib/connection-parameters.js @@ -67,9 +67,19 @@ class ConnectionParameters { this.database = this.user } - this.port = parseInt(val('port', config), 10) + const rawPort = val('port', config) + this.port = Array.isArray(rawPort) ? rawPort.map((p) => parseInt(p, 10)) : parseInt(rawPort, 10) this.host = val('host', config) + const hosts = Array.isArray(this.host) ? this.host : [this.host] + const ports = Array.isArray(this.port) ? this.port : [this.port] + if (hosts.length === 0) { + throw new Error('host must contain at least one entry') + } + if (ports.length !== 1 && ports.length !== hosts.length) { + throw new Error(`ports must have either 1 entry or the same number of entries as hosts (${hosts.length})`) + } + // "hiding" the password so it doesn't show up in stack traces // or if the client is console.logged Object.defineProperty(this, 'password', { @@ -123,6 +133,17 @@ class ConnectionParameters { this.idle_in_transaction_session_timeout = val('idle_in_transaction_session_timeout', config, false) this.query_timeout = val('query_timeout', config, false) + this.targetSessionAttrs = val('targetSessionAttrs', config) + + const validTargetSessionAttrs = ['any', 'read-write', 'read-only', 'primary', 'standby', 'prefer-standby'] + if (this.targetSessionAttrs && !validTargetSessionAttrs.includes(this.targetSessionAttrs)) { + throw new Error( + `invalid targetSessionAttrs value: "${this.targetSessionAttrs}". Must be one of: ${validTargetSessionAttrs.join( + ', ' + )}` + ) + } + if (config.connectionTimeoutMillis === undefined) { this.connect_timeout = process.env.PGCONNECT_TIMEOUT || 0 } else { diff --git a/packages/pg/lib/multihost.js b/packages/pg/lib/multihost.js new file mode 100644 index 000000000..5890582c4 --- /dev/null +++ b/packages/pg/lib/multihost.js @@ -0,0 +1,148 @@ +'use strict' + +const { once } = require('events') +const Connection = require('./connection') + +async function query(connection, text, signal) { + let value + const onRow = (row) => { + value = row.fields[0]?.toString('utf8') + } + connection.once('dataRow', onRow) + const ready = once(connection, 'readyForQuery', { signal }) + try { + connection.query(text) + await ready + return value + } finally { + connection.removeListener('dataRow', onRow) + // A custom stream can throw before we await the server's response. + ready.catch(() => {}) + } +} + +async function matchesTarget(connection, target, params, signal) { + switch (target) { + case 'read-write': + case 'read-only': { + let readOnly = params.default_transaction_read_only + if (params.in_hot_standby === 'on') { + readOnly = 'on' + } else if (readOnly === undefined || params.in_hot_standby === undefined) { + readOnly = await query(connection, 'SHOW transaction_read_only', signal) + } + return readOnly === (target === 'read-only' ? 'on' : 'off') + } + case 'primary': + case 'standby': { + if (params.in_hot_standby !== undefined) { + return params.in_hot_standby === (target === 'standby' ? 'on' : 'off') + } + const recovery = await query(connection, 'SELECT pg_catalog.pg_is_in_recovery()', signal) + return recovery === (target === 'standby' ? 't' : 'f') + } + default: + return true + } +} + +module.exports = async function connectMultiHost(client, config) { + const hosts = [].concat(client.host) + const ports = [].concat(client.port) + const target = client.connectionParameters.targetSessionAttrs || 'any' + const targets = target === 'prefer-standby' ? ['standby', 'any'] : [target] + const password = client.password + let connection = client.connection + let lastError + + for (const attrs of targets) { + for (let i = 0; i < hosts.length; i++) { + connection = connection || new Connection(config) + client.connection = connection + client.host = client.connectionParameters.host = hosts[i] + client.port = client.connectionParameters.port = ports.length === 1 ? ports[0] : ports[i] + client.connectionParameters.isDomainSocket = typeof client.host === 'string' && client.host.startsWith('/') + client.password = client.connectionParameters.password = password + const controller = new AbortController() + const { signal } = controller + const params = {} + let connected = false + let probing = false + let queryError + let accepted = false + const onConnect = () => { + connected = true + } + const onError = (err) => { + if (!signal.aborted) { + lastError = err + controller.abort() + } + } + const onErrorMessage = (err) => { + queryError = err + onError(err) + } + const onEnd = () => + onError(new Error(client._ending ? 'Connection terminated' : 'Connection terminated unexpectedly')) + const onParameter = (msg) => { + params[msg.parameterName] = msg.parameterValue + } + connection.once('connect', onConnect) + connection.on('error', onError) + connection.on('errorMessage', onErrorMessage) + connection.on('end', onEnd) + connection.on('parameterStatus', onParameter) + + try { + const ready = once(connection, 'readyForQuery', { signal }) + try { + client._connectHost(client.port, client.host) + } catch (err) { + onError(err) + } + const [message] = await ready + probing = true + const matches = await matchesTarget(connection, attrs, params, signal) + if (signal.aborted) { + throw lastError + } + if (client._ending) throw new Error('Connection terminated') + if (matches) { + accepted = true + client._attachListeners(connection) + return message + } + lastError = null + } catch (err) { + if (!signal.aborted) { + lastError = err + } + // Only a failed session probe may retry after the transport connects. + if (client._ending || connection._ending || (connected && (!probing || lastError !== queryError))) { + throw lastError + } + } finally { + connection.removeListener('end', onEnd) + if (accepted) { + connection.removeListener('connect', onConnect) + connection.removeListener('errorMessage', onErrorMessage) + connection.removeListener('parameterStatus', onParameter) + connection.removeListener('error', onError) + } else { + controller.abort() + // Preserve Client.end() callbacks while detaching the discarded backend. + for (const event of connection.eventNames()) { + if (event !== 'end') connection.removeAllListeners(event) + } + connection.on('error', () => {}) + connection._ending = true + if (connected) connection.end() + else if (connection.stream.destroy) connection.stream.destroy() + connection = null + } + } + } + } + throw lastError || new Error('None of the hosts satisfy target_session_attrs="' + target + '"') +} diff --git a/packages/pg/test/integration/client/multihost-tests.js b/packages/pg/test/integration/client/multihost-tests.js new file mode 100644 index 000000000..eb57cb658 --- /dev/null +++ b/packages/pg/test/integration/client/multihost-tests.js @@ -0,0 +1,137 @@ +'use strict' + +const assert = require('assert') +const net = require('net') +const { once } = require('events') +const helper = require('./test-helper') +const { Client, Pool } = helper.pg +const suite = new helper.Suite() + +if (helper.args.native) return + +async function unusedPort() { + const server = net.createServer() + server.listen(0, '127.0.0.1') + await once(server, 'listening') + const { port } = server.address() + await new Promise((resolve) => server.close(resolve)) + return port +} + +suite.test('TCP fallback supports prepared statements and both TLS negotiations', async () => { + let serverVersion + for (const sslnegotiation of [undefined, 'postgres', 'direct']) { + if (sslnegotiation === 'direct' && serverVersion < 170000) continue + const client = new Client({ + ...helper.config, + host: ['127.0.0.1', helper.config.host], + port: [await unusedPort(), helper.config.port], + ssl: sslnegotiation ? { rejectUnauthorized: false } : false, + sslnegotiation, + enableChannelBinding: true, + targetSessionAttrs: sslnegotiation ? 'read-write' : 'any', + connectionTimeoutMillis: 2000, + }) + try { + await client.connect() + assert.strictEqual(client.host, helper.config.host) + assert.strictEqual(client.port, Number(helper.config.port)) + assert.deepStrictEqual( + (await client.query({ name: 'multihost', text: 'SELECT $1::int AS value', values: [1] })).rows, + [{ value: 1 }] + ) + assert.deepStrictEqual((await client.query({ name: 'multihost', values: [2] })).rows, [{ value: 2 }]) + serverVersion = Number((await client.query('SHOW server_version_num')).rows[0].server_version_num) + if (sslnegotiation) assert.strictEqual(client.connection.stream.encrypted, true) + if (sslnegotiation === 'direct') assert.strictEqual(client.connection.stream.alpnProtocol, 'postgresql') + } finally { + await client.end() + } + } +}) + +suite.test('session checks accept and reject the real backend', async () => { + for (const [targetSessionAttrs, readOnly, accepted] of [ + ['primary', false, true], + ['standby', false, false], + ['read-only', true, true], + ['read-only', false, false], + ['read-write', true, false], + ]) { + const client = new Client({ + ...helper.config, + options: '-c default_transaction_read_only=' + (readOnly ? 'on' : 'off'), + targetSessionAttrs, + }) + try { + if (accepted) { + await client.connect() + assert.strictEqual( + (await client.query('SHOW transaction_read_only')).rows[0].transaction_read_only, + readOnly ? 'on' : 'off' + ) + } else { + await assert.rejects( + client.connect(), + new RegExp('None of the hosts satisfy target_session_attrs="' + targetSessionAttrs + '"') + ) + } + } finally { + await client.end() + } + } +}) + +suite.test('prefer-standby makes a second pass from the first host', async () => { + let attempts = 0 + const client = new Client({ + ...helper.config, + host: [helper.config.host, helper.config.host], + targetSessionAttrs: 'prefer-standby', + stream: () => { + attempts++ + return new net.Socket() + }, + }) + try { + await client.connect() + assert.strictEqual(attempts, 3) + assert.deepStrictEqual((await client.query('SELECT 1 AS value')).rows, [{ value: 1 }]) + } finally { + await client.end() + } +}) + +suite.test('a startup error stops host selection', async () => { + let attempts = 0 + const client = new Client({ + ...helper.config, + host: [helper.config.host, helper.config.host], + database: 'node_postgres_multihost_database_that_does_not_exist', + stream: () => { + attempts++ + return new net.Socket() + }, + }) + try { + await assert.rejects(client.connect(), { code: '3D000' }) + assert.strictEqual(attempts, 1) + } finally { + await client.end() + } +}) + +suite.test('pool queries work after skipping an unavailable Unix socket', async () => { + const pool = new Pool({ + ...helper.config, + host: ['/node-postgres-multihost-missing', helper.config.host], + targetSessionAttrs: 'read-write', + max: 1, + }) + try { + assert.deepStrictEqual((await pool.query('SELECT 1 AS value')).rows, [{ value: 1 }]) + assert.deepStrictEqual((await pool.query('SELECT 2 AS value')).rows, [{ value: 2 }]) + } finally { + await pool.end() + } +}) diff --git a/packages/pg/test/unit/client/multihost-tests.js b/packages/pg/test/unit/client/multihost-tests.js new file mode 100644 index 000000000..afa89aff2 --- /dev/null +++ b/packages/pg/test/unit/client/multihost-tests.js @@ -0,0 +1,328 @@ +'use strict' + +const assert = require('assert') +const { once } = require('events') +const { Client, MemoryStream, Suite } = require('./test-helper') + +const suite = new Suite() + +function makeClient(attempts, config = {}) { + const calls = [] + const streams = [] + const factoryConfigs = [] + const client = new Client({ + host: ['first', 'second'], + port: [5432, 5433], + ...config, + stream: (options) => { + const attempt = attempts[streams.length] + assert.ok(attempt, 'unexpected connection attempt') + factoryConfigs.push(options) + const stream = new MemoryStream() + streams.push(stream) + stream.destroy = (error) => { + if (stream.destroyed) return + stream.destroyed = true + process.nextTick(() => { + if (error) stream.emit('error', error) + stream.emit('close') + }) + } + stream.end = () => stream.destroy() + stream.connect = (port, host) => { + calls.push({ port, host }) + const connection = client.connection + process.nextTick(() => { + if (attempt.error) { + stream.emit('error', attempt.error) + return + } + stream.emit('connect') + attempt(connection, stream) + }) + } + return stream + }, + }) + return { client, calls, streams, factoryConfigs } +} + +function ready(params = {}) { + return (connection) => { + for (const [parameterName, parameterValue] of Object.entries(params)) { + connection.emit('parameterStatus', { parameterName, parameterValue }) + } + connection.emit('readyForQuery', { status: 'I' }) + } +} + +function probe(value, error) { + return (connection) => { + connection.query = (text) => { + connection.probeQuery = text + process.nextTick(() => { + connection.emit('rowDescription', { fields: [] }) + if (error) { + connection.emit('errorMessage', error) + } else { + connection.emit('dataRow', { fields: [value == null ? null : Buffer.from(value)] }) + connection.emit('commandComplete', { text: 'SELECT 1' }) + } + connection.emit('readyForQuery', { status: 'I' }) + }) + } + ready()(connection) + } +} + +const refused = Object.assign(new Error('Connection refused'), { code: 'ECONNREFUSED' }) +const primary = { in_hot_standby: 'off', default_transaction_read_only: 'off' } +const standby = { in_hot_standby: 'on', default_transaction_read_only: 'off' } + +suite.test('retries with matching ports and the same stream configuration', async () => { + for (const canDestroy of [true, false]) { + const { client, calls, streams, factoryConfigs } = makeClient([{ error: refused }, ready()]) + if (!canDestroy) streams[0].destroy = undefined + await client.connect() + assert.deepStrictEqual(calls, [ + { port: 5432, host: 'first' }, + { port: 5433, host: 'second' }, + ]) + assert.strictEqual(factoryConfigs[0], factoryConfigs[1]) + if (canDestroy) assert.ok(streams[0].destroyed) + assert.strictEqual(client.connection.stream, streams[1]) + assert.deepStrictEqual(client.connection.submittedNamedStatements, {}) + assert.strictEqual(client.host, 'second') + assert.strictEqual(client.connectionParameters.host, 'second') + assert.strictEqual(client.port, 5433) + assert.doesNotThrow(() => streams[0].emit('error', new Error('late error'))) + await client.end() + } +}) + +suite.test('reuses a scalar port and handles each Unix socket path independently', async () => { + for (const hosts of [ + ['/tmp/', 'db'], + ['db', '/tmp/'], + ]) { + const { client, calls } = makeClient([{ error: refused }, ready()], { host: hosts, port: 5432 }) + await client.connect() + assert.deepStrictEqual( + calls, + hosts.map((host) => + host.startsWith('/') ? { port: '/tmp/.s.PGSQL.5432', host: undefined } : { port: 5432, host } + ) + ) + await client.end() + } +}) + +suite.test('reports the last error once when all hosts fail', async () => { + const lastError = new Error('last host failed') + const { client, calls } = makeClient([{ error: refused }, { error: lastError }]) + let callbacks = 0 + let ends = 0 + client.on('end', () => ends++) + await new Promise((resolve) => + client.connect((err) => { + callbacks++ + assert.strictEqual(err, lastError) + resolve() + }) + ) + await new Promise(setImmediate) + assert.strictEqual(callbacks, 1) + assert.strictEqual(ends, 1) + assert.strictEqual(calls.length, 2) + await client.end() +}) + +suite.test('does not retry startup authentication or transport errors after TCP connect', async () => { + for (const event of ['error', 'errorMessage']) { + const error = new Error('startup failed') + const { client, calls } = makeClient([(connection) => connection.emit(event, error)]) + await assert.rejects(client.connect(), (err) => err === error) + assert.strictEqual(calls.length, 1) + await client.end() + } +}) + +for (const [target, rejected, accepted] of [ + ['read-write', standby, primary], + ['read-write', { ...primary, default_transaction_read_only: 'on' }, primary], + ['read-only', primary, standby], + ['primary', standby, primary], + ['standby', primary, standby], + ['prefer-standby', primary, standby], +]) { + suite.test('selects the matching host for ' + target, async () => { + const { client, calls } = makeClient([ready(rejected), ready(accepted)], { targetSessionAttrs: target }) + let connects = 0 + client.on('connect', () => connects++) + await client.connect() + assert.strictEqual(calls.length, 2) + assert.strictEqual(connects, 1) + await client.end() + }) +} + +suite.test('prefer-standby also retries after transport failures in the first pass', async () => { + const { client, calls } = makeClient([{ error: refused }, { error: refused }, ready()], { + targetSessionAttrs: 'prefer-standby', + }) + await client.connect() + assert.deepStrictEqual( + calls.map((call) => call.host), + ['first', 'second', 'first'] + ) + await client.end() +}) + +for (const [target, value, query] of [ + ['read-write', 'off', 'SHOW transaction_read_only'], + ['read-only', 'on', 'SHOW transaction_read_only'], + ['primary', 'f', 'SELECT pg_catalog.pg_is_in_recovery()'], + ['standby', 't', 'SELECT pg_catalog.pg_is_in_recovery()'], +]) { + suite.test('probes missing parameters for ' + target, async () => { + const { client } = makeClient([probe(value)], { host: ['first'], port: [5432], targetSessionAttrs: target }) + await client.connect() + assert.strictEqual(client.connection.probeQuery, query) + await client.end() + }) +} + +suite.test('retries rejected probes and does not retain previous backend parameters', async () => { + for (const first of [probe('on'), probe(null), probe('invalid'), probe(null, new Error('probe denied'))]) { + const { client, calls } = makeClient([first, probe('off')], { targetSessionAttrs: 'read-write' }) + await client.connect() + assert.strictEqual(calls.length, 2) + await client.end() + } + const { client } = makeClient([ready(standby), probe('f')], { targetSessionAttrs: 'primary' }) + await client.connect() + assert.strictEqual(client.connection.probeQuery, 'SELECT pg_catalog.pg_is_in_recovery()') + await client.end() +}) + +suite.test('times out the current attempt without starting another one', async () => { + const { client, calls, streams } = makeClient([{ error: refused }, () => {}], { + connectionTimeoutMillis: 20, + }) + await assert.rejects(client.connect(), /timeout expired/) + assert.strictEqual(calls.length, 2) + assert.ok(streams.every((stream) => stream.destroyed)) + await client.end() +}) + +for (const stage of ['startup', 'probe', 'ready']) { + suite.test('end during ' + stage + ' stops selection and settles both callbacks', async () => { + let ended + const { client, calls } = makeClient( + [ + (connection, stream) => { + stream.end = () => setImmediate(() => stream.destroy()) + const end = () => { + ended = client.end() + } + if (stage === 'startup') return end() + if (stage === 'probe') connection.query = end + else connection.once('readyForQuery', end) + ready()(connection) + }, + ], + { targetSessionAttrs: stage === 'probe' ? 'primary' : 'any' } + ) + await assert.rejects(client.connect(), { message: 'Connection terminated' }) + await ended + assert.strictEqual(calls.length, 1) + }) +} + +suite.test('selected connection supports Sync, cancel and runtime errors', async () => { + const { client, calls, streams } = makeClient([{ error: refused }, ready()]) + await client.connect() + client.connection.sync() + assert.strictEqual(client.connection._ending, false) + const stream = new MemoryStream() + let endpoint + stream.connect = (port, host) => { + endpoint = { port, host } + } + const canceller = new Client({ host: ['first', 'second'], port: [5432, 5433], stream }) + const query = {} + client._activeQuery = query + client.processID = 123 + client.secretKey = 456 + canceller.cancel(client, query) + stream.emit('connect') + assert.deepStrictEqual(endpoint, { port: 5433, host: 'second' }) + assert.strictEqual(stream.packets[0].readInt32BE(8), 123) + assert.strictEqual(stream.packets[0].readInt32BE(12), 456) + client._activeQuery = null + const error = Object.assign(new Error('Connection reset'), { code: 'ECONNRESET' }) + const failure = once(client, 'error') + streams[1].emit('error', error) + assert.strictEqual((await failure)[0], error) + assert.strictEqual(calls.length, 2) + await client.end() +}) + +suite.test('a synchronous probe write failure closes the attempt', async () => { + const failure = new Error('write failed') + const { client, calls, streams } = makeClient( + [ + (connection) => { + connection.query = () => { + throw failure + } + ready()(connection) + }, + ], + { targetSessionAttrs: 'primary' } + ) + await assert.rejects(client.connect(), (error) => error === failure) + assert.strictEqual(calls.length, 1) + assert.ok(streams[0].destroyed) + await client.end() +}) + +suite.test('discarded connections cannot change the selected backend or emit notices', async () => { + let discarded + const { client, streams } = makeClient( + [ + (connection) => { + discarded = connection + ready(standby)(connection) + }, + (connection) => { + connection.emit('backendKeyData', { processID: 123, secretKey: 456 }) + ready(primary)(connection) + }, + ], + { targetSessionAttrs: 'primary' } + ) + const notices = [] + client.on('notice', (notice) => notices.push(notice)) + await client.connect() + assert.ok( + streams[0].packets.some((packet) => packet[0] === 0x58), + 'rejected backend receives Terminate' + ) + discarded.emit('backendKeyData', { processID: 999, secretKey: 999 }) + discarded.emit('notice', { message: 'late notice' }) + assert.strictEqual(client.processID, 123) + assert.strictEqual(client.secretKey, 456) + assert.deepStrictEqual(notices, []) + await client.end() +}) + +suite.test('normalizes and validates multihost options', () => { + assert.deepStrictEqual(new Client({ host: ['first', 'second'], port: ['5432', '5433'] }).port, [5432, 5433]) + assert.deepStrictEqual(new Client({ host: 'localhost', port: ['5432'] }).port, [5432]) + assert.throws(() => new Client({ host: [] }), /host must contain at least one entry/) + for (const port of [[], [5432, 5433, 5434]]) { + assert.throws(() => new Client({ host: ['first', 'second'], port }), /ports must have either 1 entry/) + } + assert.throws(() => new Client({ targetSessionAttrs: 'read-mostly' }), /invalid targetSessionAttrs value/) +})