From eb2d8ba72a4682d9da2a097d804f34b387a52e41 Mon Sep 17 00:00:00 2001 From: maxbronnikov10 Date: Mon, 29 Jun 2026 03:21:32 +0300 Subject: [PATCH 1/2] feat: Add multihost support for native js driver --- .gitignore | 1 + docs/pages/apis/client.mdx | 5 +- docs/pages/features/connecting.mdx | 68 ++ packages/pg/lib/client.js | 32 +- packages/pg/lib/connection-parameters.js | 20 +- packages/pg/lib/multi-connection.js | 312 ++++++++ packages/pg/lib/multihost.js | 71 ++ .../pg/test/unit/client/multihost-tests.js | 104 +++ .../connection-parameters/multihost-tests.js | 88 +++ .../unit/multi-connection/multihost-tests.js | 710 ++++++++++++++++++ 10 files changed, 1397 insertions(+), 14 deletions(-) create mode 100644 packages/pg/lib/multi-connection.js create mode 100644 packages/pg/lib/multihost.js create mode 100644 packages/pg/test/unit/client/multihost-tests.js create mode 100644 packages/pg/test/unit/connection-parameters/multihost-tests.js create mode 100644 packages/pg/test/unit/multi-connection/multihost-tests.js 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..7902760ed 100644 --- a/docs/pages/features/connecting.mdx +++ b/docs/pages/features/connecting.mdx @@ -129,6 +129,74 @@ client = new Client({ }) ``` +## Multiple hosts + +node-postgres supports connecting to multiple PostgreSQL hosts. Pass arrays to `host` and `port` to enable automatic failover — the client tries each host in order and uses the first one it can reach. + +```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 +``` + +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. + ## 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..da2739644 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 MultiConnection = require('./multi-connection') const crypto = require('./crypto/utils') const activeQueryDeprecationNotice = nodeUtils.deprecate( @@ -86,16 +87,23 @@ 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) + const targetSessionAttrs = c.targetSessionAttrs || this.connectionParameters.targetSessionAttrs || null + 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', + targetSessionAttrs: targetSessionAttrs, + } + const needsMultiConnection = + Array.isArray(this.host) || + Array.isArray(this.port) || + Boolean(targetSessionAttrs && targetSessionAttrs !== 'any') + 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', - }) + c.connection || (needsMultiConnection ? new MultiConnection(connectionConfig) : new Connection(connectionConfig)) this._queryQueue = [] this._sentQueryQueue = [] this.pipeline = Boolean(c.pipeline) @@ -175,7 +183,9 @@ class Client extends EventEmitter { } } - if (this.host && this.host.indexOf('/') === 0) { + if (con instanceof MultiConnection || Array.isArray(this.host)) { + con.connect(this.port, this.host) + } else if (this.host && this.host.indexOf('/') === 0) { con.connect(this.host + '/.s.PGSQL.' + this.port) } else { con.connect(this.port, this.host) @@ -575,7 +585,7 @@ class Client extends EventEmitter { if (client.activeQuery === query) { const con = this.connection - if (this.host && this.host.indexOf('/') === 0) { + if (!Array.isArray(this.host) && this.host && this.host.indexOf('/') === 0) { con.connect(this.host + '/.s.PGSQL.' + this.port) } else { con.connect(this.port, this.host) diff --git a/packages/pg/lib/connection-parameters.js b/packages/pg/lib/connection-parameters.js index 37987fd68..b39ce99d7 100644 --- a/packages/pg/lib/connection-parameters.js +++ b/packages/pg/lib/connection-parameters.js @@ -67,9 +67,16 @@ 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 (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 +130,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/multi-connection.js b/packages/pg/lib/multi-connection.js new file mode 100644 index 000000000..c912e2397 --- /dev/null +++ b/packages/pg/lib/multi-connection.js @@ -0,0 +1,312 @@ +'use strict' + +const EventEmitter = require('events').EventEmitter +const Connection = require('./connection') +const multiHost = require('./multihost') + +const PHASE = { + STARTUP: 'startup', + PROBE: 'probe', + DONE: 'done', +} + +const PROBE_QUERY = { + tx_read_only: 'SHOW transaction_read_only', + is_in_recovery: 'SELECT pg_catalog.pg_is_in_recovery()', +} + +function isUnixSocketHost(host) { + return typeof host === 'string' && host.startsWith('/') +} + +function unixSocketPath(host, port) { + return host.replace(/\/+$/, '') + '/.s.PGSQL.' + port +} + +function connectEndpoint(connection, port, host) { + if (isUnixSocketHost(host)) { + connection.connect(unixSocketPath(host, port)) + return + } + connection.connect(port, host) +} + +class MultiConnection extends EventEmitter { + constructor(config) { + super() + this._config = config || {} + this._targetSessionAttrs = this._config.targetSessionAttrs || null + this._connection = null + this._attempt = null + this._connecting = false + this._isEnding = false + this._emitMessage = false + + this.on('newListener', (eventName) => { + if (eventName === 'message') { + this._emitMessage = true + } + }) + } + + get stream() { + return this._connection && this._connection.stream + } + + get parsedStatements() { + return this._connection && this._connection.parsedStatements + } + + get host() { + return this._hosts && this._hosts[this._hostIndex] + } + + get port() { + if (!this._ports) { + return undefined + } + return this._ports.length === 1 ? this._ports[0] : this._ports[this._hostIndex] + } + + get _ending() { + return this._isEnding + } + + set _ending(value) { + this._isEnding = value + if (this._connection) { + this._connection._ending = value + } + } + + connect(port, host) { + this._connecting = true + this._hosts = Array.isArray(host) ? host : [host] + this._ports = Array.isArray(port) ? port : [port] + this._hostIndex = 0 + this._preferStandbyPass = 1 + this._needsSessionAttrsCheck = Boolean(this._targetSessionAttrs && this._targetSessionAttrs !== 'any') + this._probeType = this._needsSessionAttrsCheck ? multiHost.probeType(this._targetSessionAttrs) : null + this._startAttempt() + } + + _startAttempt() { + const connection = new Connection(this._config) + connection._ending = this._isEnding + const attempt = { + connection: connection, + connected: false, + phase: PHASE.STARTUP, + probeRows: [], + probeError: false, + backendParams: {}, + } + + this._connection = connection + this._attempt = attempt + + connection.on('message', (msg) => this._onMessage(attempt, msg)) + connection.once('connect', () => { + if (attempt !== this._attempt) { + return + } + attempt.connected = true + this.emit('connect') + }) + connection.once('sslconnect', () => { + if (attempt === this._attempt) { + this.emit('sslconnect') + } + }) + connection.on('error', (error) => this._onAttemptError(attempt, error)) + connection.once('end', () => { + if (attempt === this._attempt) { + this.emit('end') + } + }) + + connectEndpoint(connection, this.port, this.host) + } + + _onAttemptError(attempt, error) { + if (attempt !== this._attempt) { + return + } + if (this._ending && (error.code === 'ECONNRESET' || error.code === 'EPIPE')) { + return + } + if (!attempt.connected) { + this._disposeAttempt(attempt) + if (this._advanceEndpoint()) { + this._startAttempt() + return + } + } + this._connecting = false + this.emit('error', error) + } + + _advanceEndpoint() { + if (this._hostIndex + 1 < this._hosts.length) { + this._hostIndex++ + return true + } + if (this._targetSessionAttrs === 'prefer-standby' && this._preferStandbyPass === 1) { + this._preferStandbyPass = 2 + this._hostIndex = 0 + return true + } + return false + } + + _onMessage(attempt, msg) { + if (attempt !== this._attempt) { + return + } + const eventName = msg.name === 'error' ? 'errorMessage' : msg.name + + if (!this._needsSessionAttrsCheck || attempt.phase === PHASE.DONE) { + this._forwardMessage(msg, eventName) + return + } + + if (eventName === 'parameterStatus') { + attempt.backendParams[msg.parameterName] = msg.parameterValue + this._forwardMessage(msg, eventName) + return + } + + if (attempt.phase === PHASE.STARTUP) { + if (eventName !== 'readyForQuery') { + this._forwardMessage(msg, eventName) + return + } + + const canDecide = multiHost.canDecideFromParams(this._targetSessionAttrs, attempt.backendParams) + if (canDecide) { + if (this._hostMatches(attempt)) { + this._acceptAttempt(attempt, msg) + } else { + this._rejectAttempt(attempt) + } + return + } + + attempt.phase = PHASE.PROBE + attempt.connection.query(PROBE_QUERY[this._probeType]) + return + } + + if (eventName === 'dataRow') { + attempt.probeRows.push(msg) + return + } + if (eventName === 'rowDescription' || eventName === 'commandComplete') { + return + } + if (eventName === 'errorMessage') { + attempt.probeError = true + return + } + if (eventName !== 'readyForQuery') { + this._forwardMessage(msg, eventName) + return + } + + if (!attempt.probeError && attempt.probeRows.length > 0) { + attempt.backendParams = multiHost.applyProbeResult(this._probeType, attempt.probeRows[0], attempt.backendParams) + } + + if (!attempt.probeError && this._hostMatches(attempt)) { + this._acceptAttempt(attempt, msg) + } else { + this._rejectAttempt(attempt) + } + } + + _hostMatches(attempt) { + return multiHost.hostMatches( + this._targetSessionAttrs, + attempt.backendParams, + this._hostIndex, + this._hosts.length, + this._preferStandbyPass + ) + } + + _acceptAttempt(attempt, readyMessage) { + attempt.phase = PHASE.DONE + attempt.probeRows = null + attempt.backendParams = null + this._forwardMessage(readyMessage, 'readyForQuery') + } + + _rejectAttempt(attempt) { + this._disposeAttempt(attempt) + if (this._advanceEndpoint()) { + this._startAttempt() + return + } + this._connecting = false + this.emit('error', new Error('None of the hosts satisfy target_session_attrs="' + this._targetSessionAttrs + '"')) + } + + _disposeAttempt(attempt) { + attempt.connection.removeAllListeners() + // The stream still reports through Connection until teardown finishes. + // Keep EventEmitter's special "error" event from becoming unhandled. + attempt.connection.on('error', function () {}) + attempt.connection._ending = true + if (attempt.connected) { + attempt.connection.end() + } else if (typeof attempt.connection.stream.destroy === 'function') { + attempt.connection.stream.destroy() + } + } + + _forwardMessage(msg, eventName) { + if (this._emitMessage) { + this.emit('message', msg) + } + this.emit(eventName, msg) + } + + sync() { + this._ending = true + return this._connection.sync() + } + + end() { + this._ending = true + return this._connection.end() + } +} + +const delegatedMethods = [ + 'requestSsl', + 'startup', + 'cancel', + 'password', + 'sendSASLInitialResponseMessage', + 'sendSCRAMClientFinalMessage', + 'query', + 'parse', + 'bind', + 'execute', + 'flush', + 'ref', + 'unref', + 'close', + 'describe', + 'sendCopyFromChunk', + 'endCopyFrom', + 'sendCopyFail', +] + +for (const method of delegatedMethods) { + MultiConnection.prototype[method] = function (...args) { + return this._connection[method](...args) + } +} + +module.exports = MultiConnection diff --git a/packages/pg/lib/multihost.js b/packages/pg/lib/multihost.js new file mode 100644 index 000000000..6c8927268 --- /dev/null +++ b/packages/pg/lib/multihost.js @@ -0,0 +1,71 @@ +'use strict' + +// What probe query type to run for the given targetSessionAttrs +function probeType(targetAttrs) { + switch (targetAttrs) { + case 'read-write': + case 'read-only': + return 'tx_read_only' + case 'primary': + case 'standby': + case 'prefer-standby': + return 'is_in_recovery' + default: + return null + } +} + +// Return params merged with values from a probe row so hostMatches() can decide +function applyProbeResult(probeType, row, params) { + const val = row.fields[0]?.toString('utf8') ?? null + if (val === null) { + return params + } + if (probeType === 'tx_read_only') { + return { ...params, default_transaction_read_only: val, in_hot_standby: val } + } + return { ...params, in_hot_standby: val === 't' ? 'on' : 'off' } +} + +// Can we decide host suitability from ParameterStatus messages alone (skip probe)? +function canDecideFromParams(targetAttrs, params) { + switch (targetAttrs) { + case 'read-write': + case 'read-only': + return params.in_hot_standby !== undefined && params.default_transaction_read_only !== undefined + case 'primary': + case 'standby': + case 'prefer-standby': + return params.in_hot_standby !== undefined + default: + return false + } +} + +// Does this host satisfy targetSessionAttrs? +function hostMatches(targetAttrs, params, hostIndex, hostCount, preferStandbyPass) { + switch (targetAttrs) { + case 'read-write': + return params.in_hot_standby !== 'on' && params.default_transaction_read_only !== 'on' + case 'read-only': + return params.in_hot_standby === 'on' || params.default_transaction_read_only === 'on' + case 'primary': + return params.in_hot_standby !== 'on' + case 'standby': + return params.in_hot_standby !== 'off' + case 'prefer-standby': + if (preferStandbyPass === 2) { + return true + } + return params.in_hot_standby !== 'off' || hostIndex + 1 >= hostCount + default: + return true + } +} + +module.exports = { + applyProbeResult, + canDecideFromParams, + hostMatches, + probeType, +} 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..acae313d7 --- /dev/null +++ b/packages/pg/test/unit/client/multihost-tests.js @@ -0,0 +1,104 @@ +'use strict' +const assert = require('assert') +const EventEmitter = require('events') +const helper = require('./test-helper') +const Connection = require('../../../lib/connection') +const MultiConnection = require('../../../lib/multi-connection') +const { Client } = helper + +const suite = new helper.Suite() + +function makeFakeConnection() { + const con = new EventEmitter() + con.connectCalls = [] + con.connect = function (port, host) { + con.connectCalls.push({ port, host }) + } + con.on = con.addListener.bind(con) + con.once = EventEmitter.prototype.once.bind(con) + con.removeAllListeners = EventEmitter.prototype.removeAllListeners.bind(con) + con._ending = false + con.requestSsl = function () {} + con.startup = function () {} + con.end = function () {} + return con +} + +suite.test('passes port array to connection.connect', function () { + const con = makeFakeConnection() + const client = new Client({ connection: con, host: ['localhost', '127.0.0.1'], port: [5432, 5433] }) + client._connect(function () {}) + assert.deepStrictEqual(client.port, [5432, 5433]) + assert.deepStrictEqual(con.connectCalls[0].port, [5432, 5433]) +}) + +suite.test('passes host array to connection.connect', function () { + const con = makeFakeConnection() + const client = new Client({ connection: con, host: ['h1', 'h2'], port: 5432 }) + client._connect(function () {}) + assert.deepStrictEqual(client.host, ['h1', 'h2']) + assert.deepStrictEqual(con.connectCalls[0].host, ['h1', 'h2']) +}) + +suite.test('passes host and port arrays together to connection.connect', function () { + const con = makeFakeConnection() + const client = new Client({ connection: con, host: ['h1', 'h2'], port: [5432, 5433] }) + client._connect(function () {}) + assert.deepStrictEqual(con.connectCalls[0], { port: [5432, 5433], host: ['h1', 'h2'] }) +}) + +// --- Unix socket path is not broken by the array guard --- + +suite.test('Unix socket path still works with single string host', function () { + const con = makeFakeConnection() + con.connect = function (path) { + con.connectCalls.push({ path }) + } + const client = new Client({ connection: con, host: '/tmp/', port: 5432 }) + client._connect(function () {}) + assert.ok(con.connectCalls[0].path.startsWith('/tmp/'), 'should use Unix socket path') +}) + +// --- single host / single port unchanged --- + +suite.test('single host and port are still passed as scalars', function () { + const con = makeFakeConnection() + const client = new Client({ connection: con, host: 'localhost', port: 5432 }) + client._connect(function () {}) + assert.strictEqual(con.connectCalls[0].port, 5432) + assert.strictEqual(con.connectCalls[0].host, 'localhost') +}) + +suite.test('uses MultiConnection for a host array', function () { + const client = new Client({ host: ['host1', 'host2'], port: 5432 }) + assert.ok(client.connection instanceof MultiConnection) +}) + +suite.test('uses MultiConnection for a port array', function () { + const client = new Client({ host: 'localhost', port: [5432] }) + assert.ok(client.connection instanceof MultiConnection) +}) + +suite.test('uses MultiConnection for targetSessionAttrs on one host', function () { + const client = new Client({ + host: 'localhost', + port: 5432, + targetSessionAttrs: 'read-write', + }) + assert.ok(client.connection instanceof MultiConnection) +}) + +suite.test('keeps Connection for a plain single host', function () { + const client = new Client({ host: 'localhost', port: 5432 }) + assert.ok(client.connection instanceof Connection) +}) + +suite.test('keeps an injected connection unchanged', function () { + const con = makeFakeConnection() + const client = new Client({ + connection: con, + host: ['host1', 'host2'], + port: 5432, + }) + assert.strictEqual(client.connection, con) +}) diff --git a/packages/pg/test/unit/connection-parameters/multihost-tests.js b/packages/pg/test/unit/connection-parameters/multihost-tests.js new file mode 100644 index 000000000..dfb59bb98 --- /dev/null +++ b/packages/pg/test/unit/connection-parameters/multihost-tests.js @@ -0,0 +1,88 @@ +'use strict' +const assert = require('assert') +const helper = require('../test-helper') +const ConnectionParameters = require('../../../lib/connection-parameters') + +for (const key in process.env) { + delete process.env[key] +} + +const suite = new helper.Suite() + +suite.test('single port as number is parsed to integer', function () { + const subject = new ConnectionParameters({ port: 5432 }) + assert.strictEqual(subject.port, 5432) +}) + +suite.test('single port as string is parsed to integer', function () { + const subject = new ConnectionParameters({ port: '5433' }) + assert.strictEqual(subject.port, 5433) +}) + +suite.test('port array of numbers is preserved as integer array', function () { + const subject = new ConnectionParameters({ host: ['h1', 'h2'], port: [5432, 5433] }) + assert.deepStrictEqual(subject.port, [5432, 5433]) +}) + +suite.test('port array of strings is mapped to integers', function () { + const subject = new ConnectionParameters({ host: ['h1', 'h2', 'h3'], port: ['5432', '5433', '5434'] }) + assert.deepStrictEqual(subject.port, [5432, 5433, 5434]) +}) + +suite.test('port array with single element is preserved as array', function () { + const subject = new ConnectionParameters({ port: [5432] }) + assert.deepStrictEqual(subject.port, [5432]) +}) + +suite.test('single host string is preserved', function () { + const subject = new ConnectionParameters({ host: 'localhost' }) + assert.strictEqual(subject.host, 'localhost') +}) + +suite.test('host array is passed through unchanged', function () { + const subject = new ConnectionParameters({ host: ['host1', 'host2', 'host3'] }) + assert.deepStrictEqual(subject.host, ['host1', 'host2', 'host3']) +}) + +suite.test('host array with single element is preserved as array', function () { + const subject = new ConnectionParameters({ host: ['localhost'] }) + assert.deepStrictEqual(subject.host, ['localhost']) +}) + +suite.test('host and port arrays are both passed through', function () { + const subject = new ConnectionParameters({ host: ['h1', 'h2'], port: [5432, 5433] }) + assert.deepStrictEqual(subject.host, ['h1', 'h2']) + assert.deepStrictEqual(subject.port, [5432, 5433]) +}) + +suite.test('scalar port with host array is valid and preserved as number', function () { + const subject = new ConnectionParameters({ host: ['h1', 'h2', 'h3'], port: 5432 }) + assert.deepStrictEqual(subject.host, ['h1', 'h2', 'h3']) + assert.strictEqual(subject.port, 5432) +}) + +suite.test('isDomainSocket is false when host is an array', function () { + const subject = new ConnectionParameters({ host: ['/tmp/', 'localhost'] }) + assert.strictEqual(subject.isDomainSocket, false) +}) + +suite.test('invalid targetSessionAttrs throws', function () { + assert.throws( + () => new ConnectionParameters({ targetSessionAttrs: 'read-mostly' }), + /invalid targetSessionAttrs value/ + ) +}) + +suite.test('valid targetSessionAttrs values do not throw', function () { + const valid = ['any', 'read-write', 'read-only', 'primary', 'standby', 'prefer-standby'] + for (const value of valid) { + assert.doesNotThrow(() => new ConnectionParameters({ targetSessionAttrs: value })) + } +}) + +suite.test('mismatched ports and hosts count throws', function () { + assert.throws( + () => new ConnectionParameters({ host: ['h1', 'h2', 'h3'], port: [5432, 5433] }), + /ports must have either 1 entry/ + ) +}) diff --git a/packages/pg/test/unit/multi-connection/multihost-tests.js b/packages/pg/test/unit/multi-connection/multihost-tests.js new file mode 100644 index 000000000..f423f1e84 --- /dev/null +++ b/packages/pg/test/unit/multi-connection/multihost-tests.js @@ -0,0 +1,710 @@ +'use strict' +const helper = require('../test-helper') +const MultiConnection = require('../../../lib/multi-connection') +const assert = require('assert') + +const suite = new helper.Suite() +const { MemoryStream } = helper + +function makeStream() { + const stream = new MemoryStream() + stream.destroy = function () {} + return stream +} + +function makeErrorMessageBuf() { + // 'E' + length + 'S' + 'ERROR\0' + '\0' + const content = Buffer.concat([Buffer.from('SERROR\0'), Buffer.from([0x00])]) + const len = 4 + content.length + const buf = Buffer.allocUnsafe(1 + len) + buf[0] = 0x45 // 'E' + buf.writeUInt32BE(len, 1) + content.copy(buf, 5) + return buf +} + +function makeParameterStatusBuf(name, value) { + const n = Buffer.from(name + '\0') + const v = Buffer.from(value + '\0') + const len = 4 + n.length + v.length + const buf = Buffer.allocUnsafe(1 + len) + buf[0] = 0x53 // 'S' + buf.writeUInt32BE(len, 1) + n.copy(buf, 5) + v.copy(buf, 5 + n.length) + return buf +} + +function makeReadyForQueryBuf() { + return Buffer.from([0x5a, 0x00, 0x00, 0x00, 0x05, 0x49]) // 'Z' len=5 status='I' +} + +function makeDataRowBuf(fields) { + const bufs = fields.map((f) => (Buffer.isBuffer(f) ? f : Buffer.from(f))) + let dataLen = 2 // Int16 field count + for (const f of bufs) dataLen += 4 + f.length // Int32 len + data + const totalLen = 4 + dataLen // Int32 length field includes itself + const buf = Buffer.allocUnsafe(1 + totalLen) + buf[0] = 0x44 // 'D' + buf.writeUInt32BE(totalLen, 1) + buf.writeUInt16BE(bufs.length, 5) + let offset = 7 + for (const f of bufs) { + buf.writeInt32BE(f.length, offset) + offset += 4 + f.copy(buf, offset) + offset += f.length + } + return buf +} + +function makeRowDescriptionBuf() { + // 'T', length=6, field count=0 + return Buffer.from([0x54, 0x00, 0x00, 0x00, 0x06, 0x00, 0x00]) +} + +function makeCommandCompleteBuf() { + const tag = Buffer.from('SELECT 1\0') + const len = 4 + tag.length + const buf = Buffer.allocUnsafe(1 + len) + buf[0] = 0x43 // 'C' + buf.writeUInt32BE(len, 1) + tag.copy(buf, 5) + return buf +} + +function simulateReadyForQuery(stream, params) { + for (const [key, value] of Object.entries(params)) { + stream.emit('data', makeParameterStatusBuf(key, value)) + } + stream.emit('data', makeReadyForQueryBuf()) +} + +suite.test('connects to single host', function (done) { + const stream = makeStream() + let connectPort, connectHost + stream.connect = function (port, host) { + connectPort = port + connectHost = host + } + const con = new MultiConnection({ stream: stream }) + con.once('connect', function () { + assert.equal(connectPort, 5432) + assert.equal(connectHost, 'localhost') + done() + }) + con.connect(5432, 'localhost') + stream.emit('connect') +}) + +suite.test('connects to first host when multiple are given', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const connectCalls = [] + streams.forEach((s) => { + s.connect = function (port, host) { + connectCalls.push({ port, host }) + } + }) + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.equal(connectCalls.length, 1) + assert.equal(connectCalls[0].host, 'host1') + assert.equal(connectCalls[0].port, 5432) + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + streams[0].emit('connect') +}) + +suite.test('stream factory receives same config on failover streams', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const factoryArgs = [] + const config = { + ssl: false, + stream: function (opts) { + factoryArgs.push(opts) + return streams[streamIndex++] + }, + } + const con = new MultiConnection(config) + con.once('connect', function () { + assert.equal(factoryArgs.length, 2) + assert.strictEqual(factoryArgs[0], config) + assert.strictEqual(factoryArgs[1], config) + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + const err = new Error('Connection refused') + err.code = 'ECONNREFUSED' + streams[0].emit('error', err) + streams[1].emit('connect') +}) + +suite.test('falls back to second host on connection error', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const connectCalls = [] + streams.forEach((s) => { + s.connect = function (port, host) { + connectCalls.push({ port, host }) + } + }) + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.equal(connectCalls.length, 2) + assert.equal(connectCalls[0].host, 'host1') + assert.equal(connectCalls[1].host, 'host2') + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + const err = new Error('Connection refused') + err.code = 'ECONNREFUSED' + streams[0].emit('error', err) + streams[1].emit('connect') +}) + +suite.test('uses matching port for each host by index', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const connectCalls = [] + streams.forEach((s) => { + s.connect = function (port, host) { + connectCalls.push({ port, host }) + } + }) + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.equal(connectCalls[0].port, 5432) + assert.equal(connectCalls[1].port, 5433) + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + const err = new Error('Connection refused') + err.code = 'ECONNREFUSED' + streams[0].emit('error', err) + streams[1].emit('connect') +}) + +suite.test('reuses single port for all hosts when port is not an array', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const connectPorts = [] + streams.forEach((s) => { + s.connect = function (port) { + connectPorts.push(port) + } + }) + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.equal(connectPorts[0], 5432) + assert.equal(connectPorts[1], 5432) + done() + }) + con.connect(5432, ['host1', 'host2']) + const err = new Error('Connection refused') + err.code = 'ECONNREFUSED' + streams[0].emit('error', err) + streams[1].emit('connect') +}) + +suite.test('emits error after all hosts fail', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + assert.emits(con, 'error', function () { + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + const err1 = new Error('Connection refused') + err1.code = 'ECONNREFUSED' + streams[0].emit('error', err1) + const err2 = new Error('Connection refused') + err2.code = 'ECONNREFUSED' + streams[1].emit('error', err2) +}) + +suite.test('does not fall back after successful connect', function (done) { + const stream = makeStream() + const con = new MultiConnection({ stream: stream }) + con.once('connect', function () { + assert.emits(con, 'error', function (err) { + assert.equal(err.code, 'ECONNRESET') + done() + }) + const err = new Error('Connection reset') + err.code = 'ECONNRESET' + stream.emit('error', err) + }) + con.connect([5432, 5433], ['host1', 'host2']) + stream.emit('connect') +}) + +suite.test('targetSessionAttrs=any does not intercept readyForQuery', function (done) { + const stream = makeStream() + const con = new MultiConnection({ targetSessionAttrs: 'any', stream: stream }) + con.once('readyForQuery', function () { + done() + }) + con.connect(5432, 'localhost') + stream.emit('connect') + con.emit('readyForQuery', {}) +}) + +suite.test('targetSessionAttrs=read-write skips hot standby and uses primary', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + assert.equal(streamIndex, 2) + done() + }) + con.connect([5432, 5433], ['standby', 'primary']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'off' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('targetSessionAttrs=read-write skips read-only and uses writable', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + done() + }) + con.connect([5432, 5433], ['readonly', 'writable']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'on' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('targetSessionAttrs=read-only skips primary and uses standby', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-only', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + done() + }) + con.connect([5432, 5433], ['primary', 'standby']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) +}) + +suite.test('targetSessionAttrs=primary skips standby and uses primary', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'primary', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + done() + }) + con.connect([5432, 5433], ['standby', 'primary']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('targetSessionAttrs=standby skips primary and uses hot standby', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'standby', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + done() + }) + con.connect([5432, 5433], ['primary', 'standby']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) +}) + +suite.test('targetSessionAttrs=prefer-standby uses standby when available', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'prefer-standby', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + assert.equal(streamIndex, 2) + done() + }) + con.connect([5432, 5433], ['primary', 'standby']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) +}) + +suite.test('targetSessionAttrs=prefer-standby falls back to primary when no standby available', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'prefer-standby', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + done() + }) + con.connect([5432, 5433], ['primary1', 'primary2']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('emits error when no host satisfies targetSessionAttrs', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: () => streams[streamIndex++], + }) + assert.emits(con, 'error', function (err) { + assert.ok(err.message.includes('read-write')) + done() + }) + con.connect([5432, 5433], ['standby1', 'standby2']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'off' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'off' }) +}) + +suite.test('resets backend params between hosts when checking targetSessionAttrs', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'primary', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + done() + }) + con.connect([5432, 5433], ['standby', 'primary']) + streams[0].emit('connect') + // standby sends in_hot_standby=on → skip + simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('fetches session state via SHOW query when not provided in ParameterStatus', function (done) { + const stream = makeStream() + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: stream, + }) + con.once('readyForQuery', function () { + done() + }) + con.connect(5432, 'localhost') + stream.emit('connect') + stream.emit('data', makeReadyForQueryBuf()) + stream.emit('data', makeDataRowBuf([Buffer.from('off')])) + stream.emit('data', makeReadyForQueryBuf()) +}) + +suite.test('tries next host when SHOW query returns standby state', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + assert.equal(streamIndex, 2) + done() + }) + con.connect([5432, 5433], ['standby', 'primary']) + streams[0].emit('connect') + streams[0].emit('data', makeReadyForQueryBuf()) + streams[0].emit('data', makeDataRowBuf([Buffer.from('on')])) // transaction_read_only=on + streams[0].emit('data', makeReadyForQueryBuf()) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('prefer-standby triggers pass 2 when all hosts fail TCP in pass 1', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream(), makeStream()] + const connectHosts = [] + streams.forEach((s) => { + s.connect = function (_port, host) { + connectHosts.push(host) + } + }) + const con = new MultiConnection({ + targetSessionAttrs: 'prefer-standby', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + // pass 2 reconnects from the beginning of the host list + assert.equal(connectHosts[2], 'host1') + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + const err1 = new Error('Connection refused') + err1.code = 'ECONNREFUSED' + streams[0].emit('error', err1) + const err2 = new Error('Connection refused') + err2.code = 'ECONNREFUSED' + streams[1].emit('error', err2) + streams[2].emit('connect') + simulateReadyForQuery(streams[2], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) +}) + +suite.test('probe error causes next host to be tried', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: () => streams[streamIndex++], + }) + con.once('readyForQuery', function () { + assert.equal(streamIndex, 2) + done() + }) + con.connect([5432, 5433], ['host1', 'host2']) + streams[0].emit('connect') + streams[0].emit('data', makeReadyForQueryBuf()) + streams[0].emit('data', makeErrorMessageBuf()) + streams[0].emit('data', makeReadyForQueryBuf()) + streams[1].emit('connect') + streams[1].emit('data', makeReadyForQueryBuf()) + streams[1].emit('data', makeDataRowBuf([Buffer.from('off')])) + streams[1].emit('data', makeReadyForQueryBuf()) +}) + +suite.test('read-only host accepted when tx_read_only probe returns on', function (done) { + const stream = makeStream() + const con = new MultiConnection({ + targetSessionAttrs: 'read-only', + stream: stream, + }) + con.once('readyForQuery', function () { + done() + }) + con.connect(5432, 'localhost') + stream.emit('connect') + stream.emit('data', makeReadyForQueryBuf()) + stream.emit('data', makeDataRowBuf([Buffer.from('on')])) + stream.emit('data', makeReadyForQueryBuf()) +}) + +suite.test('swallows rowDescription and commandComplete during SHOW fetch', function (done) { + const stream = makeStream() + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: stream, + }) + const unexpectedEvents = [] + for (const evt of ['rowDescription', 'commandComplete']) { + con.on(evt, function () { + unexpectedEvents.push(evt) + }) + } + con.once('readyForQuery', function () { + assert.equal(unexpectedEvents.length, 0) + done() + }) + con.connect(5432, 'localhost') + stream.emit('connect') + stream.emit('data', makeReadyForQueryBuf()) + stream.emit('data', makeRowDescriptionBuf()) + stream.emit('data', makeDataRowBuf([Buffer.from('off')])) // transaction_read_only=off + stream.emit('data', makeCommandCompleteBuf()) + stream.emit('data', makeReadyForQueryBuf()) +}) + +suite.test('forwards protocol events through the facade', function (done) { + const stream = makeStream() + const con = new MultiConnection({ stream: stream }) + con.once('readyForQuery', function (msg) { + assert.equal(msg.status, 'I') + done() + }) + con.connect(5432, 'localhost') + stream.emit('connect') + stream.emit('data', makeReadyForQueryBuf()) +}) + +suite.test('exposes the active stream and parsed statements', function () { + const stream = makeStream() + const con = new MultiConnection({ stream: stream }) + con.connect(5432, 'localhost') + assert.strictEqual(con.stream, stream) + assert.strictEqual(con.parsedStatements, con._connection.parsedStatements) +}) + +suite.test('delegates protocol writes to the active connection', function () { + const stream = makeStream() + const con = new MultiConnection({ stream: stream }) + con.connect(5432, 'localhost') + con.startup({ user: 'test', database: 'test' }) + assert.equal(stream.packets.length, 1) +}) + +suite.test('emits readyForQuery only for the accepted candidate', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ + targetSessionAttrs: 'read-write', + stream: () => streams[streamIndex++], + }) + let readyCount = 0 + con.on('readyForQuery', function () { + readyCount++ + assert.equal(readyCount, 1) + assert.equal(streamIndex, 2) + done() + }) + con.connect(5432, ['standby', 'primary']) + streams[0].emit('connect') + simulateReadyForQuery(streams[0], { + in_hot_standby: 'on', + default_transaction_read_only: 'off', + }) + assert.equal(readyCount, 0) + streams[1].emit('connect') + simulateReadyForQuery(streams[1], { + in_hot_standby: 'off', + default_transaction_read_only: 'off', + }) +}) + +suite.test('connects to a Unix socket selected from a host array', function (done) { + const stream = makeStream() + stream.connect = function (path, host) { + assert.equal(path, '/tmp/.s.PGSQL.5432') + assert.equal(host, undefined) + } + const con = new MultiConnection({ stream: stream }) + con.once('connect', done) + con.connect(5432, ['/tmp', 'localhost']) + stream.emit('connect') +}) + +suite.test('falls back from a Unix socket to TCP', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const calls = [] + streams[0].connect = function (path) { + calls.push({ path: path }) + } + streams[1].connect = function (port, host) { + calls.push({ port: port, host: host }) + } + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.deepStrictEqual(calls, [ + { path: '/var/run/postgresql/.s.PGSQL.5432' }, + { port: 5433, host: 'db.example.com' }, + ]) + done() + }) + con.connect([5432, 5433], ['/var/run/postgresql', 'db.example.com']) + const error = new Error('Connection refused') + error.code = 'ECONNREFUSED' + streams[0].emit('error', error) + streams[1].emit('connect') +}) + +suite.test('falls back from TCP to a Unix socket', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const calls = [] + streams[0].connect = function (port, host) { + calls.push({ port: port, host: host }) + } + streams[1].connect = function (path) { + calls.push({ path: path }) + } + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.deepStrictEqual(calls, [ + { port: 5432, host: 'db.example.com' }, + { path: '/var/run/postgresql/.s.PGSQL.5433' }, + ]) + done() + }) + con.connect([5432, 5433], ['db.example.com', '/var/run/postgresql/']) + const error = new Error('Connection refused') + error.code = 'ECONNREFUSED' + streams[0].emit('error', error) + streams[1].emit('connect') +}) + +suite.test('does not forward end from a rejected candidate', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + let endCount = 0 + con.on('end', function () { + endCount++ + }) + con.once('connect', function () { + assert.equal(streamIndex, 2) + streams[0].emit('close') + assert.equal(endCount, 0) + done() + }) + con.connect(5432, ['host1', 'host2']) + const error = new Error('Connection refused') + error.code = 'ECONNREFUSED' + streams[0].emit('error', error) + streams[1].emit('connect') +}) + +suite.test('absorbs late errors from a rejected candidate', function (done) { + let streamIndex = 0 + const streams = [makeStream(), makeStream()] + const con = new MultiConnection({ stream: () => streams[streamIndex++] }) + con.once('connect', function () { + assert.doesNotThrow(function () { + streams[0].emit('error', new Error('late candidate error')) + }) + done() + }) + con.connect(5432, ['host1', 'host2']) + const error = new Error('Connection refused') + error.code = 'ECONNREFUSED' + streams[0].emit('error', error) + streams[1].emit('connect') +}) + +suite.test('delegates end to the accepted candidate', function (done) { + const stream = makeStream() + let ended = false + stream.end = function () { + ended = true + } + const con = new MultiConnection({ stream: stream }) + con.connect(5432, 'localhost') + stream.emit('connect') + con.end() + setImmediate(function () { + assert.equal(ended, true) + done() + }) +}) From b9eb354e6eebf040b7eb981140b27d5283138e66 Mon Sep 17 00:00:00 2001 From: maxbronnikov10 Date: Wed, 16 Sep 2026 02:57:58 +0300 Subject: [PATCH 2/2] refactor(pg): simplify multihost connections --- docs/pages/features/connecting.mdx | 8 +- packages/pg/lib/client.js | 120 +-- packages/pg/lib/connection-parameters.js | 3 + packages/pg/lib/multi-connection.js | 312 -------- packages/pg/lib/multihost.js | 187 +++-- .../integration/client/multihost-tests.js | 137 ++++ .../pg/test/unit/client/multihost-tests.js | 378 ++++++++-- .../connection-parameters/multihost-tests.js | 88 --- .../unit/multi-connection/multihost-tests.js | 710 ------------------ 9 files changed, 650 insertions(+), 1293 deletions(-) delete mode 100644 packages/pg/lib/multi-connection.js create mode 100644 packages/pg/test/integration/client/multihost-tests.js delete mode 100644 packages/pg/test/unit/connection-parameters/multihost-tests.js delete mode 100644 packages/pg/test/unit/multi-connection/multihost-tests.js diff --git a/docs/pages/features/connecting.mdx b/docs/pages/features/connecting.mdx index 7902760ed..4a19cb29c 100644 --- a/docs/pages/features/connecting.mdx +++ b/docs/pages/features/connecting.mdx @@ -131,7 +131,7 @@ client = new Client({ ## Multiple hosts -node-postgres supports connecting to multiple PostgreSQL hosts. Pass arrays to `host` and `port` to enable automatic failover — the client tries each host in order and uses the first one it can reach. +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' @@ -147,6 +147,10 @@ const client = new Client({ 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 @@ -197,6 +201,8 @@ const client = new Client({ 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 da2739644..00dd39595 100644 --- a/packages/pg/lib/client.js +++ b/packages/pg/lib/client.js @@ -8,7 +8,7 @@ const ConnectionParameters = require('./connection-parameters') const Query = require('./query') const defaults = require('./defaults') const Connection = require('./connection') -const MultiConnection = require('./multi-connection') +const connectMultiHost = require('./multihost') const crypto = require('./crypto/utils') const activeQueryDeprecationNotice = nodeUtils.deprecate( @@ -87,7 +87,7 @@ 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) - const targetSessionAttrs = c.targetSessionAttrs || this.connectionParameters.targetSessionAttrs || null + const targetSessionAttrs = this.connectionParameters.targetSessionAttrs const connectionConfig = { stream: c.stream, ssl: this.connectionParameters.ssl, @@ -95,15 +95,14 @@ class Client extends EventEmitter { keepAlive: c.keepAlive || false, keepAliveInitialDelayMillis: c.keepAliveInitialDelayMillis || 0, encoding: this.connectionParameters.client_encoding || 'utf8', - targetSessionAttrs: targetSessionAttrs, } - const needsMultiConnection = + const needsMultiHost = Array.isArray(this.host) || Array.isArray(this.port) || Boolean(targetSessionAttrs && targetSessionAttrs !== 'any') - this.connection = - c.connection || (needsMultiConnection ? new MultiConnection(connectionConfig) : new Connection(connectionConfig)) + this.connection = c.connection || new Connection(connectionConfig) + this._multiHostConfig = !c.connection && needsMultiHost ? connectionConfig : null this._queryQueue = [] this._sentQueryQueue = [] this.pipeline = Boolean(c.pipeline) @@ -159,8 +158,6 @@ class Client extends EventEmitter { } _connect(callback) { - const self = this - const con = this.connection this._connectionCallback = callback if (this._connecting || this._connected) { @@ -174,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) { @@ -183,12 +180,33 @@ class Client extends EventEmitter { } } - if (con instanceof MultiConnection || Array.isArray(this.host)) { - con.connect(this.port, this.host) - } else 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 @@ -208,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') }) } @@ -257,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)) @@ -584,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 (!Array.isArray(this.host) && 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 b39ce99d7..ff0c55431 100644 --- a/packages/pg/lib/connection-parameters.js +++ b/packages/pg/lib/connection-parameters.js @@ -73,6 +73,9 @@ class ConnectionParameters { 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})`) } diff --git a/packages/pg/lib/multi-connection.js b/packages/pg/lib/multi-connection.js deleted file mode 100644 index c912e2397..000000000 --- a/packages/pg/lib/multi-connection.js +++ /dev/null @@ -1,312 +0,0 @@ -'use strict' - -const EventEmitter = require('events').EventEmitter -const Connection = require('./connection') -const multiHost = require('./multihost') - -const PHASE = { - STARTUP: 'startup', - PROBE: 'probe', - DONE: 'done', -} - -const PROBE_QUERY = { - tx_read_only: 'SHOW transaction_read_only', - is_in_recovery: 'SELECT pg_catalog.pg_is_in_recovery()', -} - -function isUnixSocketHost(host) { - return typeof host === 'string' && host.startsWith('/') -} - -function unixSocketPath(host, port) { - return host.replace(/\/+$/, '') + '/.s.PGSQL.' + port -} - -function connectEndpoint(connection, port, host) { - if (isUnixSocketHost(host)) { - connection.connect(unixSocketPath(host, port)) - return - } - connection.connect(port, host) -} - -class MultiConnection extends EventEmitter { - constructor(config) { - super() - this._config = config || {} - this._targetSessionAttrs = this._config.targetSessionAttrs || null - this._connection = null - this._attempt = null - this._connecting = false - this._isEnding = false - this._emitMessage = false - - this.on('newListener', (eventName) => { - if (eventName === 'message') { - this._emitMessage = true - } - }) - } - - get stream() { - return this._connection && this._connection.stream - } - - get parsedStatements() { - return this._connection && this._connection.parsedStatements - } - - get host() { - return this._hosts && this._hosts[this._hostIndex] - } - - get port() { - if (!this._ports) { - return undefined - } - return this._ports.length === 1 ? this._ports[0] : this._ports[this._hostIndex] - } - - get _ending() { - return this._isEnding - } - - set _ending(value) { - this._isEnding = value - if (this._connection) { - this._connection._ending = value - } - } - - connect(port, host) { - this._connecting = true - this._hosts = Array.isArray(host) ? host : [host] - this._ports = Array.isArray(port) ? port : [port] - this._hostIndex = 0 - this._preferStandbyPass = 1 - this._needsSessionAttrsCheck = Boolean(this._targetSessionAttrs && this._targetSessionAttrs !== 'any') - this._probeType = this._needsSessionAttrsCheck ? multiHost.probeType(this._targetSessionAttrs) : null - this._startAttempt() - } - - _startAttempt() { - const connection = new Connection(this._config) - connection._ending = this._isEnding - const attempt = { - connection: connection, - connected: false, - phase: PHASE.STARTUP, - probeRows: [], - probeError: false, - backendParams: {}, - } - - this._connection = connection - this._attempt = attempt - - connection.on('message', (msg) => this._onMessage(attempt, msg)) - connection.once('connect', () => { - if (attempt !== this._attempt) { - return - } - attempt.connected = true - this.emit('connect') - }) - connection.once('sslconnect', () => { - if (attempt === this._attempt) { - this.emit('sslconnect') - } - }) - connection.on('error', (error) => this._onAttemptError(attempt, error)) - connection.once('end', () => { - if (attempt === this._attempt) { - this.emit('end') - } - }) - - connectEndpoint(connection, this.port, this.host) - } - - _onAttemptError(attempt, error) { - if (attempt !== this._attempt) { - return - } - if (this._ending && (error.code === 'ECONNRESET' || error.code === 'EPIPE')) { - return - } - if (!attempt.connected) { - this._disposeAttempt(attempt) - if (this._advanceEndpoint()) { - this._startAttempt() - return - } - } - this._connecting = false - this.emit('error', error) - } - - _advanceEndpoint() { - if (this._hostIndex + 1 < this._hosts.length) { - this._hostIndex++ - return true - } - if (this._targetSessionAttrs === 'prefer-standby' && this._preferStandbyPass === 1) { - this._preferStandbyPass = 2 - this._hostIndex = 0 - return true - } - return false - } - - _onMessage(attempt, msg) { - if (attempt !== this._attempt) { - return - } - const eventName = msg.name === 'error' ? 'errorMessage' : msg.name - - if (!this._needsSessionAttrsCheck || attempt.phase === PHASE.DONE) { - this._forwardMessage(msg, eventName) - return - } - - if (eventName === 'parameterStatus') { - attempt.backendParams[msg.parameterName] = msg.parameterValue - this._forwardMessage(msg, eventName) - return - } - - if (attempt.phase === PHASE.STARTUP) { - if (eventName !== 'readyForQuery') { - this._forwardMessage(msg, eventName) - return - } - - const canDecide = multiHost.canDecideFromParams(this._targetSessionAttrs, attempt.backendParams) - if (canDecide) { - if (this._hostMatches(attempt)) { - this._acceptAttempt(attempt, msg) - } else { - this._rejectAttempt(attempt) - } - return - } - - attempt.phase = PHASE.PROBE - attempt.connection.query(PROBE_QUERY[this._probeType]) - return - } - - if (eventName === 'dataRow') { - attempt.probeRows.push(msg) - return - } - if (eventName === 'rowDescription' || eventName === 'commandComplete') { - return - } - if (eventName === 'errorMessage') { - attempt.probeError = true - return - } - if (eventName !== 'readyForQuery') { - this._forwardMessage(msg, eventName) - return - } - - if (!attempt.probeError && attempt.probeRows.length > 0) { - attempt.backendParams = multiHost.applyProbeResult(this._probeType, attempt.probeRows[0], attempt.backendParams) - } - - if (!attempt.probeError && this._hostMatches(attempt)) { - this._acceptAttempt(attempt, msg) - } else { - this._rejectAttempt(attempt) - } - } - - _hostMatches(attempt) { - return multiHost.hostMatches( - this._targetSessionAttrs, - attempt.backendParams, - this._hostIndex, - this._hosts.length, - this._preferStandbyPass - ) - } - - _acceptAttempt(attempt, readyMessage) { - attempt.phase = PHASE.DONE - attempt.probeRows = null - attempt.backendParams = null - this._forwardMessage(readyMessage, 'readyForQuery') - } - - _rejectAttempt(attempt) { - this._disposeAttempt(attempt) - if (this._advanceEndpoint()) { - this._startAttempt() - return - } - this._connecting = false - this.emit('error', new Error('None of the hosts satisfy target_session_attrs="' + this._targetSessionAttrs + '"')) - } - - _disposeAttempt(attempt) { - attempt.connection.removeAllListeners() - // The stream still reports through Connection until teardown finishes. - // Keep EventEmitter's special "error" event from becoming unhandled. - attempt.connection.on('error', function () {}) - attempt.connection._ending = true - if (attempt.connected) { - attempt.connection.end() - } else if (typeof attempt.connection.stream.destroy === 'function') { - attempt.connection.stream.destroy() - } - } - - _forwardMessage(msg, eventName) { - if (this._emitMessage) { - this.emit('message', msg) - } - this.emit(eventName, msg) - } - - sync() { - this._ending = true - return this._connection.sync() - } - - end() { - this._ending = true - return this._connection.end() - } -} - -const delegatedMethods = [ - 'requestSsl', - 'startup', - 'cancel', - 'password', - 'sendSASLInitialResponseMessage', - 'sendSCRAMClientFinalMessage', - 'query', - 'parse', - 'bind', - 'execute', - 'flush', - 'ref', - 'unref', - 'close', - 'describe', - 'sendCopyFromChunk', - 'endCopyFrom', - 'sendCopyFail', -] - -for (const method of delegatedMethods) { - MultiConnection.prototype[method] = function (...args) { - return this._connection[method](...args) - } -} - -module.exports = MultiConnection diff --git a/packages/pg/lib/multihost.js b/packages/pg/lib/multihost.js index 6c8927268..5890582c4 100644 --- a/packages/pg/lib/multihost.js +++ b/packages/pg/lib/multihost.js @@ -1,71 +1,148 @@ 'use strict' -// What probe query type to run for the given targetSessionAttrs -function probeType(targetAttrs) { - switch (targetAttrs) { - case 'read-write': - case 'read-only': - return 'tx_read_only' - case 'primary': - case 'standby': - case 'prefer-standby': - return 'is_in_recovery' - default: - return null - } -} +const { once } = require('events') +const Connection = require('./connection') -// Return params merged with values from a probe row so hostMatches() can decide -function applyProbeResult(probeType, row, params) { - const val = row.fields[0]?.toString('utf8') ?? null - if (val === null) { - return params +async function query(connection, text, signal) { + let value + const onRow = (row) => { + value = row.fields[0]?.toString('utf8') } - if (probeType === 'tx_read_only') { - return { ...params, default_transaction_read_only: val, in_hot_standby: val } - } - return { ...params, in_hot_standby: val === 't' ? 'on' : 'off' } -} - -// Can we decide host suitability from ParameterStatus messages alone (skip probe)? -function canDecideFromParams(targetAttrs, params) { - switch (targetAttrs) { - case 'read-write': - case 'read-only': - return params.in_hot_standby !== undefined && params.default_transaction_read_only !== undefined - case 'primary': - case 'standby': - case 'prefer-standby': - return params.in_hot_standby !== undefined - default: - return false + 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(() => {}) } } -// Does this host satisfy targetSessionAttrs? -function hostMatches(targetAttrs, params, hostIndex, hostCount, preferStandbyPass) { - switch (targetAttrs) { +async function matchesTarget(connection, target, params, signal) { + switch (target) { case 'read-write': - return params.in_hot_standby !== 'on' && params.default_transaction_read_only !== 'on' - case 'read-only': - return params.in_hot_standby === 'on' || params.default_transaction_read_only === 'on' + 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': - return params.in_hot_standby !== 'on' - case 'standby': - return params.in_hot_standby !== 'off' - case 'prefer-standby': - if (preferStandbyPass === 2) { - return true + case 'standby': { + if (params.in_hot_standby !== undefined) { + return params.in_hot_standby === (target === 'standby' ? 'on' : 'off') } - return params.in_hot_standby !== 'off' || hostIndex + 1 >= hostCount + const recovery = await query(connection, 'SELECT pg_catalog.pg_is_in_recovery()', signal) + return recovery === (target === 'standby' ? 't' : 'f') + } default: return true } } -module.exports = { - applyProbeResult, - canDecideFromParams, - hostMatches, - probeType, +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 index acae313d7..afa89aff2 100644 --- a/packages/pg/test/unit/client/multihost-tests.js +++ b/packages/pg/test/unit/client/multihost-tests.js @@ -1,104 +1,328 @@ 'use strict' + const assert = require('assert') -const EventEmitter = require('events') -const helper = require('./test-helper') -const Connection = require('../../../lib/connection') -const MultiConnection = require('../../../lib/multi-connection') -const { Client } = helper - -const suite = new helper.Suite() - -function makeFakeConnection() { - const con = new EventEmitter() - con.connectCalls = [] - con.connect = function (port, host) { - con.connectCalls.push({ port, host }) +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' }) } - con.on = con.addListener.bind(con) - con.once = EventEmitter.prototype.once.bind(con) - con.removeAllListeners = EventEmitter.prototype.removeAllListeners.bind(con) - con._ending = false - con.requestSsl = function () {} - con.startup = function () {} - con.end = function () {} - return con } -suite.test('passes port array to connection.connect', function () { - const con = makeFakeConnection() - const client = new Client({ connection: con, host: ['localhost', '127.0.0.1'], port: [5432, 5433] }) - client._connect(function () {}) - assert.deepStrictEqual(client.port, [5432, 5433]) - assert.deepStrictEqual(con.connectCalls[0].port, [5432, 5433]) -}) +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) + } +} -suite.test('passes host array to connection.connect', function () { - const con = makeFakeConnection() - const client = new Client({ connection: con, host: ['h1', 'h2'], port: 5432 }) - client._connect(function () {}) - assert.deepStrictEqual(client.host, ['h1', 'h2']) - assert.deepStrictEqual(con.connectCalls[0].host, ['h1', 'h2']) +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('passes host and port arrays together to connection.connect', function () { - const con = makeFakeConnection() - const client = new Client({ connection: con, host: ['h1', 'h2'], port: [5432, 5433] }) - client._connect(function () {}) - assert.deepStrictEqual(con.connectCalls[0], { port: [5432, 5433], host: ['h1', 'h2'] }) +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() + } }) -// --- Unix socket path is not broken by the array guard --- +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('Unix socket path still works with single string host', function () { - const con = makeFakeConnection() - con.connect = function (path) { - con.connectCalls.push({ path }) +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() } - const client = new Client({ connection: con, host: '/tmp/', port: 5432 }) - client._connect(function () {}) - assert.ok(con.connectCalls[0].path.startsWith('/tmp/'), 'should use Unix socket path') }) -// --- single host / single port unchanged --- +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('single host and port are still passed as scalars', function () { - const con = makeFakeConnection() - const client = new Client({ connection: con, host: 'localhost', port: 5432 }) - client._connect(function () {}) - assert.strictEqual(con.connectCalls[0].port, 5432) - assert.strictEqual(con.connectCalls[0].host, 'localhost') +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() }) -suite.test('uses MultiConnection for a host array', function () { - const client = new Client({ host: ['host1', 'host2'], port: 5432 }) - assert.ok(client.connection instanceof MultiConnection) +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('uses MultiConnection for a port array', function () { - const client = new Client({ host: 'localhost', port: [5432] }) - assert.ok(client.connection instanceof MultiConnection) +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() }) -suite.test('uses MultiConnection for targetSessionAttrs on one host', function () { - const client = new Client({ - host: 'localhost', - port: 5432, - targetSessionAttrs: 'read-write', +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) }) - assert.ok(client.connection instanceof MultiConnection) +} + +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('keeps Connection for a plain single host', function () { - const client = new Client({ host: 'localhost', port: 5432 }) - assert.ok(client.connection instanceof Connection) +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('keeps an injected connection unchanged', function () { - const con = makeFakeConnection() - const client = new Client({ - connection: con, - host: ['host1', 'host2'], - port: 5432, - }) - assert.strictEqual(client.connection, con) +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/) }) diff --git a/packages/pg/test/unit/connection-parameters/multihost-tests.js b/packages/pg/test/unit/connection-parameters/multihost-tests.js deleted file mode 100644 index dfb59bb98..000000000 --- a/packages/pg/test/unit/connection-parameters/multihost-tests.js +++ /dev/null @@ -1,88 +0,0 @@ -'use strict' -const assert = require('assert') -const helper = require('../test-helper') -const ConnectionParameters = require('../../../lib/connection-parameters') - -for (const key in process.env) { - delete process.env[key] -} - -const suite = new helper.Suite() - -suite.test('single port as number is parsed to integer', function () { - const subject = new ConnectionParameters({ port: 5432 }) - assert.strictEqual(subject.port, 5432) -}) - -suite.test('single port as string is parsed to integer', function () { - const subject = new ConnectionParameters({ port: '5433' }) - assert.strictEqual(subject.port, 5433) -}) - -suite.test('port array of numbers is preserved as integer array', function () { - const subject = new ConnectionParameters({ host: ['h1', 'h2'], port: [5432, 5433] }) - assert.deepStrictEqual(subject.port, [5432, 5433]) -}) - -suite.test('port array of strings is mapped to integers', function () { - const subject = new ConnectionParameters({ host: ['h1', 'h2', 'h3'], port: ['5432', '5433', '5434'] }) - assert.deepStrictEqual(subject.port, [5432, 5433, 5434]) -}) - -suite.test('port array with single element is preserved as array', function () { - const subject = new ConnectionParameters({ port: [5432] }) - assert.deepStrictEqual(subject.port, [5432]) -}) - -suite.test('single host string is preserved', function () { - const subject = new ConnectionParameters({ host: 'localhost' }) - assert.strictEqual(subject.host, 'localhost') -}) - -suite.test('host array is passed through unchanged', function () { - const subject = new ConnectionParameters({ host: ['host1', 'host2', 'host3'] }) - assert.deepStrictEqual(subject.host, ['host1', 'host2', 'host3']) -}) - -suite.test('host array with single element is preserved as array', function () { - const subject = new ConnectionParameters({ host: ['localhost'] }) - assert.deepStrictEqual(subject.host, ['localhost']) -}) - -suite.test('host and port arrays are both passed through', function () { - const subject = new ConnectionParameters({ host: ['h1', 'h2'], port: [5432, 5433] }) - assert.deepStrictEqual(subject.host, ['h1', 'h2']) - assert.deepStrictEqual(subject.port, [5432, 5433]) -}) - -suite.test('scalar port with host array is valid and preserved as number', function () { - const subject = new ConnectionParameters({ host: ['h1', 'h2', 'h3'], port: 5432 }) - assert.deepStrictEqual(subject.host, ['h1', 'h2', 'h3']) - assert.strictEqual(subject.port, 5432) -}) - -suite.test('isDomainSocket is false when host is an array', function () { - const subject = new ConnectionParameters({ host: ['/tmp/', 'localhost'] }) - assert.strictEqual(subject.isDomainSocket, false) -}) - -suite.test('invalid targetSessionAttrs throws', function () { - assert.throws( - () => new ConnectionParameters({ targetSessionAttrs: 'read-mostly' }), - /invalid targetSessionAttrs value/ - ) -}) - -suite.test('valid targetSessionAttrs values do not throw', function () { - const valid = ['any', 'read-write', 'read-only', 'primary', 'standby', 'prefer-standby'] - for (const value of valid) { - assert.doesNotThrow(() => new ConnectionParameters({ targetSessionAttrs: value })) - } -}) - -suite.test('mismatched ports and hosts count throws', function () { - assert.throws( - () => new ConnectionParameters({ host: ['h1', 'h2', 'h3'], port: [5432, 5433] }), - /ports must have either 1 entry/ - ) -}) diff --git a/packages/pg/test/unit/multi-connection/multihost-tests.js b/packages/pg/test/unit/multi-connection/multihost-tests.js deleted file mode 100644 index f423f1e84..000000000 --- a/packages/pg/test/unit/multi-connection/multihost-tests.js +++ /dev/null @@ -1,710 +0,0 @@ -'use strict' -const helper = require('../test-helper') -const MultiConnection = require('../../../lib/multi-connection') -const assert = require('assert') - -const suite = new helper.Suite() -const { MemoryStream } = helper - -function makeStream() { - const stream = new MemoryStream() - stream.destroy = function () {} - return stream -} - -function makeErrorMessageBuf() { - // 'E' + length + 'S' + 'ERROR\0' + '\0' - const content = Buffer.concat([Buffer.from('SERROR\0'), Buffer.from([0x00])]) - const len = 4 + content.length - const buf = Buffer.allocUnsafe(1 + len) - buf[0] = 0x45 // 'E' - buf.writeUInt32BE(len, 1) - content.copy(buf, 5) - return buf -} - -function makeParameterStatusBuf(name, value) { - const n = Buffer.from(name + '\0') - const v = Buffer.from(value + '\0') - const len = 4 + n.length + v.length - const buf = Buffer.allocUnsafe(1 + len) - buf[0] = 0x53 // 'S' - buf.writeUInt32BE(len, 1) - n.copy(buf, 5) - v.copy(buf, 5 + n.length) - return buf -} - -function makeReadyForQueryBuf() { - return Buffer.from([0x5a, 0x00, 0x00, 0x00, 0x05, 0x49]) // 'Z' len=5 status='I' -} - -function makeDataRowBuf(fields) { - const bufs = fields.map((f) => (Buffer.isBuffer(f) ? f : Buffer.from(f))) - let dataLen = 2 // Int16 field count - for (const f of bufs) dataLen += 4 + f.length // Int32 len + data - const totalLen = 4 + dataLen // Int32 length field includes itself - const buf = Buffer.allocUnsafe(1 + totalLen) - buf[0] = 0x44 // 'D' - buf.writeUInt32BE(totalLen, 1) - buf.writeUInt16BE(bufs.length, 5) - let offset = 7 - for (const f of bufs) { - buf.writeInt32BE(f.length, offset) - offset += 4 - f.copy(buf, offset) - offset += f.length - } - return buf -} - -function makeRowDescriptionBuf() { - // 'T', length=6, field count=0 - return Buffer.from([0x54, 0x00, 0x00, 0x00, 0x06, 0x00, 0x00]) -} - -function makeCommandCompleteBuf() { - const tag = Buffer.from('SELECT 1\0') - const len = 4 + tag.length - const buf = Buffer.allocUnsafe(1 + len) - buf[0] = 0x43 // 'C' - buf.writeUInt32BE(len, 1) - tag.copy(buf, 5) - return buf -} - -function simulateReadyForQuery(stream, params) { - for (const [key, value] of Object.entries(params)) { - stream.emit('data', makeParameterStatusBuf(key, value)) - } - stream.emit('data', makeReadyForQueryBuf()) -} - -suite.test('connects to single host', function (done) { - const stream = makeStream() - let connectPort, connectHost - stream.connect = function (port, host) { - connectPort = port - connectHost = host - } - const con = new MultiConnection({ stream: stream }) - con.once('connect', function () { - assert.equal(connectPort, 5432) - assert.equal(connectHost, 'localhost') - done() - }) - con.connect(5432, 'localhost') - stream.emit('connect') -}) - -suite.test('connects to first host when multiple are given', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const connectCalls = [] - streams.forEach((s) => { - s.connect = function (port, host) { - connectCalls.push({ port, host }) - } - }) - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.equal(connectCalls.length, 1) - assert.equal(connectCalls[0].host, 'host1') - assert.equal(connectCalls[0].port, 5432) - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - streams[0].emit('connect') -}) - -suite.test('stream factory receives same config on failover streams', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const factoryArgs = [] - const config = { - ssl: false, - stream: function (opts) { - factoryArgs.push(opts) - return streams[streamIndex++] - }, - } - const con = new MultiConnection(config) - con.once('connect', function () { - assert.equal(factoryArgs.length, 2) - assert.strictEqual(factoryArgs[0], config) - assert.strictEqual(factoryArgs[1], config) - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - const err = new Error('Connection refused') - err.code = 'ECONNREFUSED' - streams[0].emit('error', err) - streams[1].emit('connect') -}) - -suite.test('falls back to second host on connection error', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const connectCalls = [] - streams.forEach((s) => { - s.connect = function (port, host) { - connectCalls.push({ port, host }) - } - }) - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.equal(connectCalls.length, 2) - assert.equal(connectCalls[0].host, 'host1') - assert.equal(connectCalls[1].host, 'host2') - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - const err = new Error('Connection refused') - err.code = 'ECONNREFUSED' - streams[0].emit('error', err) - streams[1].emit('connect') -}) - -suite.test('uses matching port for each host by index', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const connectCalls = [] - streams.forEach((s) => { - s.connect = function (port, host) { - connectCalls.push({ port, host }) - } - }) - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.equal(connectCalls[0].port, 5432) - assert.equal(connectCalls[1].port, 5433) - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - const err = new Error('Connection refused') - err.code = 'ECONNREFUSED' - streams[0].emit('error', err) - streams[1].emit('connect') -}) - -suite.test('reuses single port for all hosts when port is not an array', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const connectPorts = [] - streams.forEach((s) => { - s.connect = function (port) { - connectPorts.push(port) - } - }) - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.equal(connectPorts[0], 5432) - assert.equal(connectPorts[1], 5432) - done() - }) - con.connect(5432, ['host1', 'host2']) - const err = new Error('Connection refused') - err.code = 'ECONNREFUSED' - streams[0].emit('error', err) - streams[1].emit('connect') -}) - -suite.test('emits error after all hosts fail', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - assert.emits(con, 'error', function () { - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - const err1 = new Error('Connection refused') - err1.code = 'ECONNREFUSED' - streams[0].emit('error', err1) - const err2 = new Error('Connection refused') - err2.code = 'ECONNREFUSED' - streams[1].emit('error', err2) -}) - -suite.test('does not fall back after successful connect', function (done) { - const stream = makeStream() - const con = new MultiConnection({ stream: stream }) - con.once('connect', function () { - assert.emits(con, 'error', function (err) { - assert.equal(err.code, 'ECONNRESET') - done() - }) - const err = new Error('Connection reset') - err.code = 'ECONNRESET' - stream.emit('error', err) - }) - con.connect([5432, 5433], ['host1', 'host2']) - stream.emit('connect') -}) - -suite.test('targetSessionAttrs=any does not intercept readyForQuery', function (done) { - const stream = makeStream() - const con = new MultiConnection({ targetSessionAttrs: 'any', stream: stream }) - con.once('readyForQuery', function () { - done() - }) - con.connect(5432, 'localhost') - stream.emit('connect') - con.emit('readyForQuery', {}) -}) - -suite.test('targetSessionAttrs=read-write skips hot standby and uses primary', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - assert.equal(streamIndex, 2) - done() - }) - con.connect([5432, 5433], ['standby', 'primary']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'off' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('targetSessionAttrs=read-write skips read-only and uses writable', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - done() - }) - con.connect([5432, 5433], ['readonly', 'writable']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'on' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('targetSessionAttrs=read-only skips primary and uses standby', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-only', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - done() - }) - con.connect([5432, 5433], ['primary', 'standby']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) -}) - -suite.test('targetSessionAttrs=primary skips standby and uses primary', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'primary', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - done() - }) - con.connect([5432, 5433], ['standby', 'primary']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('targetSessionAttrs=standby skips primary and uses hot standby', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'standby', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - done() - }) - con.connect([5432, 5433], ['primary', 'standby']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) -}) - -suite.test('targetSessionAttrs=prefer-standby uses standby when available', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'prefer-standby', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - assert.equal(streamIndex, 2) - done() - }) - con.connect([5432, 5433], ['primary', 'standby']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) -}) - -suite.test('targetSessionAttrs=prefer-standby falls back to primary when no standby available', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'prefer-standby', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - done() - }) - con.connect([5432, 5433], ['primary1', 'primary2']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('emits error when no host satisfies targetSessionAttrs', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: () => streams[streamIndex++], - }) - assert.emits(con, 'error', function (err) { - assert.ok(err.message.includes('read-write')) - done() - }) - con.connect([5432, 5433], ['standby1', 'standby2']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'off' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'on', default_transaction_read_only: 'off' }) -}) - -suite.test('resets backend params between hosts when checking targetSessionAttrs', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'primary', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - done() - }) - con.connect([5432, 5433], ['standby', 'primary']) - streams[0].emit('connect') - // standby sends in_hot_standby=on → skip - simulateReadyForQuery(streams[0], { in_hot_standby: 'on', default_transaction_read_only: 'on' }) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('fetches session state via SHOW query when not provided in ParameterStatus', function (done) { - const stream = makeStream() - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: stream, - }) - con.once('readyForQuery', function () { - done() - }) - con.connect(5432, 'localhost') - stream.emit('connect') - stream.emit('data', makeReadyForQueryBuf()) - stream.emit('data', makeDataRowBuf([Buffer.from('off')])) - stream.emit('data', makeReadyForQueryBuf()) -}) - -suite.test('tries next host when SHOW query returns standby state', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - assert.equal(streamIndex, 2) - done() - }) - con.connect([5432, 5433], ['standby', 'primary']) - streams[0].emit('connect') - streams[0].emit('data', makeReadyForQueryBuf()) - streams[0].emit('data', makeDataRowBuf([Buffer.from('on')])) // transaction_read_only=on - streams[0].emit('data', makeReadyForQueryBuf()) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('prefer-standby triggers pass 2 when all hosts fail TCP in pass 1', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream(), makeStream()] - const connectHosts = [] - streams.forEach((s) => { - s.connect = function (_port, host) { - connectHosts.push(host) - } - }) - const con = new MultiConnection({ - targetSessionAttrs: 'prefer-standby', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - // pass 2 reconnects from the beginning of the host list - assert.equal(connectHosts[2], 'host1') - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - const err1 = new Error('Connection refused') - err1.code = 'ECONNREFUSED' - streams[0].emit('error', err1) - const err2 = new Error('Connection refused') - err2.code = 'ECONNREFUSED' - streams[1].emit('error', err2) - streams[2].emit('connect') - simulateReadyForQuery(streams[2], { in_hot_standby: 'off', default_transaction_read_only: 'off' }) -}) - -suite.test('probe error causes next host to be tried', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: () => streams[streamIndex++], - }) - con.once('readyForQuery', function () { - assert.equal(streamIndex, 2) - done() - }) - con.connect([5432, 5433], ['host1', 'host2']) - streams[0].emit('connect') - streams[0].emit('data', makeReadyForQueryBuf()) - streams[0].emit('data', makeErrorMessageBuf()) - streams[0].emit('data', makeReadyForQueryBuf()) - streams[1].emit('connect') - streams[1].emit('data', makeReadyForQueryBuf()) - streams[1].emit('data', makeDataRowBuf([Buffer.from('off')])) - streams[1].emit('data', makeReadyForQueryBuf()) -}) - -suite.test('read-only host accepted when tx_read_only probe returns on', function (done) { - const stream = makeStream() - const con = new MultiConnection({ - targetSessionAttrs: 'read-only', - stream: stream, - }) - con.once('readyForQuery', function () { - done() - }) - con.connect(5432, 'localhost') - stream.emit('connect') - stream.emit('data', makeReadyForQueryBuf()) - stream.emit('data', makeDataRowBuf([Buffer.from('on')])) - stream.emit('data', makeReadyForQueryBuf()) -}) - -suite.test('swallows rowDescription and commandComplete during SHOW fetch', function (done) { - const stream = makeStream() - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: stream, - }) - const unexpectedEvents = [] - for (const evt of ['rowDescription', 'commandComplete']) { - con.on(evt, function () { - unexpectedEvents.push(evt) - }) - } - con.once('readyForQuery', function () { - assert.equal(unexpectedEvents.length, 0) - done() - }) - con.connect(5432, 'localhost') - stream.emit('connect') - stream.emit('data', makeReadyForQueryBuf()) - stream.emit('data', makeRowDescriptionBuf()) - stream.emit('data', makeDataRowBuf([Buffer.from('off')])) // transaction_read_only=off - stream.emit('data', makeCommandCompleteBuf()) - stream.emit('data', makeReadyForQueryBuf()) -}) - -suite.test('forwards protocol events through the facade', function (done) { - const stream = makeStream() - const con = new MultiConnection({ stream: stream }) - con.once('readyForQuery', function (msg) { - assert.equal(msg.status, 'I') - done() - }) - con.connect(5432, 'localhost') - stream.emit('connect') - stream.emit('data', makeReadyForQueryBuf()) -}) - -suite.test('exposes the active stream and parsed statements', function () { - const stream = makeStream() - const con = new MultiConnection({ stream: stream }) - con.connect(5432, 'localhost') - assert.strictEqual(con.stream, stream) - assert.strictEqual(con.parsedStatements, con._connection.parsedStatements) -}) - -suite.test('delegates protocol writes to the active connection', function () { - const stream = makeStream() - const con = new MultiConnection({ stream: stream }) - con.connect(5432, 'localhost') - con.startup({ user: 'test', database: 'test' }) - assert.equal(stream.packets.length, 1) -}) - -suite.test('emits readyForQuery only for the accepted candidate', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ - targetSessionAttrs: 'read-write', - stream: () => streams[streamIndex++], - }) - let readyCount = 0 - con.on('readyForQuery', function () { - readyCount++ - assert.equal(readyCount, 1) - assert.equal(streamIndex, 2) - done() - }) - con.connect(5432, ['standby', 'primary']) - streams[0].emit('connect') - simulateReadyForQuery(streams[0], { - in_hot_standby: 'on', - default_transaction_read_only: 'off', - }) - assert.equal(readyCount, 0) - streams[1].emit('connect') - simulateReadyForQuery(streams[1], { - in_hot_standby: 'off', - default_transaction_read_only: 'off', - }) -}) - -suite.test('connects to a Unix socket selected from a host array', function (done) { - const stream = makeStream() - stream.connect = function (path, host) { - assert.equal(path, '/tmp/.s.PGSQL.5432') - assert.equal(host, undefined) - } - const con = new MultiConnection({ stream: stream }) - con.once('connect', done) - con.connect(5432, ['/tmp', 'localhost']) - stream.emit('connect') -}) - -suite.test('falls back from a Unix socket to TCP', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const calls = [] - streams[0].connect = function (path) { - calls.push({ path: path }) - } - streams[1].connect = function (port, host) { - calls.push({ port: port, host: host }) - } - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.deepStrictEqual(calls, [ - { path: '/var/run/postgresql/.s.PGSQL.5432' }, - { port: 5433, host: 'db.example.com' }, - ]) - done() - }) - con.connect([5432, 5433], ['/var/run/postgresql', 'db.example.com']) - const error = new Error('Connection refused') - error.code = 'ECONNREFUSED' - streams[0].emit('error', error) - streams[1].emit('connect') -}) - -suite.test('falls back from TCP to a Unix socket', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const calls = [] - streams[0].connect = function (port, host) { - calls.push({ port: port, host: host }) - } - streams[1].connect = function (path) { - calls.push({ path: path }) - } - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.deepStrictEqual(calls, [ - { port: 5432, host: 'db.example.com' }, - { path: '/var/run/postgresql/.s.PGSQL.5433' }, - ]) - done() - }) - con.connect([5432, 5433], ['db.example.com', '/var/run/postgresql/']) - const error = new Error('Connection refused') - error.code = 'ECONNREFUSED' - streams[0].emit('error', error) - streams[1].emit('connect') -}) - -suite.test('does not forward end from a rejected candidate', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - let endCount = 0 - con.on('end', function () { - endCount++ - }) - con.once('connect', function () { - assert.equal(streamIndex, 2) - streams[0].emit('close') - assert.equal(endCount, 0) - done() - }) - con.connect(5432, ['host1', 'host2']) - const error = new Error('Connection refused') - error.code = 'ECONNREFUSED' - streams[0].emit('error', error) - streams[1].emit('connect') -}) - -suite.test('absorbs late errors from a rejected candidate', function (done) { - let streamIndex = 0 - const streams = [makeStream(), makeStream()] - const con = new MultiConnection({ stream: () => streams[streamIndex++] }) - con.once('connect', function () { - assert.doesNotThrow(function () { - streams[0].emit('error', new Error('late candidate error')) - }) - done() - }) - con.connect(5432, ['host1', 'host2']) - const error = new Error('Connection refused') - error.code = 'ECONNREFUSED' - streams[0].emit('error', error) - streams[1].emit('connect') -}) - -suite.test('delegates end to the accepted candidate', function (done) { - const stream = makeStream() - let ended = false - stream.end = function () { - ended = true - } - const con = new MultiConnection({ stream: stream }) - con.connect(5432, 'localhost') - stream.emit('connect') - con.end() - setImmediate(function () { - assert.equal(ended, true) - done() - }) -})