From efd9108ed6fcf3968e696aef9b933bd2be7609a6 Mon Sep 17 00:00:00 2001 From: denchick Date: Mon, 10 Aug 2026 17:30:13 +0200 Subject: [PATCH 1/6] feat: add proxy topology mode for SPQR-compatible routing --- README.md | 18 +++++- jest/unit.config.js | 6 ++ lib/constants.ts | 1 + lib/core.ts | 5 +- lib/dispatcher.ts | 18 ++++++ lib/types.ts | 3 + tests/dispatcher.test.js | 122 +++++++++++++++++++++++++++++++++++++++ 7 files changed, 169 insertions(+), 4 deletions(-) create mode 100644 jest/unit.config.js create mode 100644 tests/dispatcher.test.js diff --git a/README.md b/README.md index b094f0a..2137d9c 100644 --- a/README.md +++ b/README.md @@ -58,13 +58,25 @@ export const {db, CoreBaseModel, helpers} = initDB({ - `connectionString` is a set of [postgres connection strings](https://stackoverflow.com/questions/3582552/postgresql-connection-url) separated by a comma, with at least one host required, for example: `'postgresql://user:password@dbHost1:5432/dbName,postgresql://user:password@dbHost2:5432/dbName'` - `logger`: Provide the `info` and `error` callbacks for logging messages -- `dispatcherOptions`: Settings for the primary/replica balancing (none of the options are required) +- `dispatcherOptions`: Settings for connection balancing (none of the options are required) - `healthcheckInterval`: Health check interval in milliseconds, default value: 5000ms - `healthcheckTimeout`: Health check interval in milliseconds, default value: 700ms - - `suppressStatusLogs`: Boolean value that disables database health checks (useful for developers) + - `suppressStatusLogs`: Boolean value that disables database health-check status logs (useful for developers) - `beforeTerminate`: Function called before terminating a connection, must return a Promise + - `topologyMode`: Connection topology, either `primary-replica` (the default) or `proxy` - `knexOptions`: Non-required additional options that will be passed to Knex before initialization +When all connection strings point to equivalent proxy or router instances, such as SPQR routers, use `proxy` mode. Healthy endpoints are then eligible for both primary and replica queries, and the endpoint with the lowest latest health-check latency is selected: + +```typescript +initDB({ + connectionString: process.env.POSTGRES_DSN_LIST, + dispatcherOptions: { + topologyMode: 'proxy', + }, +}); +``` + Here is the recommended project structure (we only list the directories that have to do with working with the database): ``` @@ -82,7 +94,7 @@ The `initDB` constructor exports three elements: `db`, `CoreBaseModel`, and `hel ### db -`db`: Instance of the [PGDispatcher](https://github.com/gravity-ui/postgreskit/blob/main/lib/dispatcher.ts) module responsible for primary/replica connection balancing. Under the hood, this module creates N instances of knex (N is the number of hosts passed in `connectionString`). From these instances, it polls the database hosts every `healthcheckInterval` milliseconds, requesting if they are primary hosts or replica hosts (using the `SELECT pg_is_in_recovery()` query). +`db`: Instance of the [PGDispatcher](https://github.com/gravity-ui/postgreskit/blob/main/lib/dispatcher.ts) module responsible for connection balancing. Under the hood, this module creates N instances of knex (N is the number of hosts passed in `connectionString`). By default, it polls the database hosts every `healthcheckInterval` milliseconds, requesting if they are primary hosts or replica hosts (using the `SELECT pg_is_in_recovery()` query). In `proxy` topology mode, it uses `SELECT 1` instead and routes both `db.primary` and `db.replica` to the fastest healthy endpoint. Public `db` methods: diff --git a/jest/unit.config.js b/jest/unit.config.js new file mode 100644 index 0000000..23bc746 --- /dev/null +++ b/jest/unit.config.js @@ -0,0 +1,6 @@ +module.exports = { + clearMocks: true, + rootDir: '..', + testEnvironment: 'node', + testMatch: ['/tests/**/*.test.js'], +}; diff --git a/lib/constants.ts b/lib/constants.ts index f11166b..81824cd 100644 --- a/lib/constants.ts +++ b/lib/constants.ts @@ -20,6 +20,7 @@ export const defaultDispatcherOptions: PDOptions = { healthcheckTimeout: 700, suppressStatusLogs: false, beforeTerminate: () => Promise.resolve(), + topologyMode: 'primary-replica', }; export const defaultExLogger: ExLogger = { diff --git a/lib/core.ts b/lib/core.ts index e076748..271c544 100644 --- a/lib/core.ts +++ b/lib/core.ts @@ -4,13 +4,16 @@ import {type Constructor, Model} from 'objection'; import {defaultDispatcherOptions, defaultExLogger, defaultKnexOptions} from './constants'; import {PGDispatcher} from './dispatcher'; -import type {BaseModel, ExLogger} from './types'; +import type {BaseModel, ExLogger, TopologyMode} from './types'; + +export type {TopologyMode} from './types'; export interface CoreDBDispatcherOptions { healthcheckInterval?: number; healthcheckTimeout?: number; suppressStatusLogs?: boolean; beforeTerminate?: () => Promise; + topologyMode?: TopologyMode; } export type GetModelParams = {cancelOnTimeout?: boolean; useLimitInFirst?: boolean}; diff --git a/lib/dispatcher.ts b/lib/dispatcher.ts index 036398f..2735d78 100644 --- a/lib/dispatcher.ts +++ b/lib/dispatcher.ts @@ -138,6 +138,10 @@ export class PGDispatcher { get primary() { this.checkConnectionsAvailability(); + if (this.options.topologyMode === 'proxy') { + return this.fastestHealthyConnection.knex; + } + const primaryConnections = this.connections.filter((c) => c.primary); if (primaryConnections.length > 1) { this.logger.error({ @@ -160,6 +164,10 @@ export class PGDispatcher { get replica() { this.checkConnectionsAvailability(); + if (this.options.topologyMode === 'proxy') { + return this.fastestHealthyConnection.knex; + } + const replicaConnections = this.healthyConnections.filter((c) => !c.primary); if (replicaConnections.length) { @@ -213,6 +221,12 @@ export class PGDispatcher { } private async performCheckupQuery(knex: Knex): Promise { + if (this.options.topologyMode === 'proxy') { + await knex.raw('SELECT 1;').timeout(this.options.healthcheckTimeout); + + return {pingOk: true, primary: false}; + } + const result = await knex .raw('SELECT pg_is_in_recovery();') .timeout(this.options.healthcheckTimeout); @@ -278,6 +292,10 @@ export class PGDispatcher { return this.connections.filter((c) => c.healthy); } + private get fastestHealthyConnection() { + return this.healthyConnections.sort((a, b) => a.latency - b.latency)[0]; + } + private checkConnectionsAvailability() { if (!this.healthyConnections.length) { const error = new PDError('No connections available'); diff --git a/lib/types.ts b/lib/types.ts index 7352d1d..5aee243 100644 --- a/lib/types.ts +++ b/lib/types.ts @@ -1,11 +1,14 @@ import type {Model} from 'objection'; import type {PGDispatcher} from './dispatcher'; +export type TopologyMode = 'primary-replica' | 'proxy'; + export interface PDOptions { healthcheckInterval: number; healthcheckTimeout: number; suppressStatusLogs: boolean; beforeTerminate: () => Promise; + topologyMode: TopologyMode; } export type Dict = {[key: string]: unknown}; diff --git a/tests/dispatcher.test.js b/tests/dispatcher.test.js new file mode 100644 index 0000000..7ced52f --- /dev/null +++ b/tests/dispatcher.test.js @@ -0,0 +1,122 @@ +/* eslint-env jest */ + +jest.mock('knex', () => jest.fn()); + +const knexBuilder = require('knex'); + +const {defaultDispatcherOptions} = require('../build/constants'); +const {PGDispatcher} = require('../build/dispatcher'); + +const activeDispatchers = []; + +function createKnex(checkup) { + const timeout = jest.fn(() => checkup()); + + return { + destroy: jest.fn(() => Promise.resolve()), + raw: jest.fn(() => ({timeout})), + timeout, + }; +} + +function successfulCheckup(row, latency = 0) { + return () => + new Promise((resolve) => { + setTimeout(() => resolve({rows: [row]}), latency); + }); +} + +function failedCheckup() { + return Promise.reject(new Error('Proxy unavailable')); +} + +function createDispatcher(clients, options = {}) { + clients.forEach((client) => knexBuilder.mockImplementationOnce(() => client)); + + const logger = { + error: jest.fn(), + info: jest.fn(), + }; + const dispatcher = new PGDispatcher({ + connections: clients.map( + (_, index) => `postgresql://user:password@database-${index}.example/db`, + ), + logger, + options: { + ...defaultDispatcherOptions, + healthcheckInterval: 60_000, + healthcheckTimeout: 100, + ...options, + }, + }); + + activeDispatchers.push(dispatcher); + + return {dispatcher, logger}; +} + +function loggedErrorMessages(logger) { + return logger.error.mock.calls.map(([, error]) => error.message); +} + +afterEach(async () => { + await Promise.all(activeDispatchers.splice(0).map((dispatcher) => dispatcher.terminate())); +}); + +describe('PGDispatcher topology modes', () => { + test('the default mode keeps the primary/replica health check and routing behavior', async () => { + const primary = createKnex(successfulCheckup({pg_is_in_recovery: false})); + const replica = createKnex(successfulCheckup({pg_is_in_recovery: true})); + const {dispatcher} = createDispatcher([primary, replica]); + + await dispatcher.ready(); + + expect(defaultDispatcherOptions.topologyMode).toBe('primary-replica'); + expect(primary.raw).toHaveBeenCalledWith('SELECT pg_is_in_recovery();'); + expect(replica.raw).toHaveBeenCalledWith('SELECT pg_is_in_recovery();'); + expect(dispatcher.primary).toBe(primary); + expect(dispatcher.replica).toBe(replica); + }); + + test('proxy mode routes both roles to the fastest healthy endpoint without topology warnings', async () => { + const slowerProxy = createKnex(successfulCheckup({value: 1}, 30)); + const fasterProxy = createKnex(successfulCheckup({value: 1}, 5)); + const unhealthyProxy = createKnex(failedCheckup); + const {dispatcher, logger} = createDispatcher([slowerProxy, fasterProxy, unhealthyProxy], { + topologyMode: 'proxy', + }); + + await dispatcher.ready(); + + for (const proxy of [slowerProxy, fasterProxy, unhealthyProxy]) { + expect(proxy.raw).toHaveBeenCalledWith('SELECT 1;'); + expect(proxy.raw).not.toHaveBeenCalledWith('SELECT pg_is_in_recovery();'); + } + expect(dispatcher.primary).toBe(fasterProxy); + expect(dispatcher.replica).toBe(fasterProxy); + expect(loggedErrorMessages(logger)).not.toEqual( + expect.arrayContaining([ + 'Multiple primary connections detected, something is wrong', + 'No alive replica available, using master for read', + ]), + ); + }); + + test('the existing unavailable-database error is preserved when all proxies are unhealthy', async () => { + const {dispatcher} = createDispatcher( + [createKnex(failedCheckup), createKnex(failedCheckup)], + {topologyMode: 'proxy'}, + ); + + await dispatcher.ready(); + + for (const connection of ['primary', 'replica']) { + expect(() => dispatcher[connection]).toThrow( + expect.objectContaining({ + code: 'ERR_DB_NOT_AVAILABLE', + message: 'No connections available', + }), + ); + } + }); +}); From 4757237809497c11e3580d8b6017d9165a09395a Mon Sep 17 00:00:00 2001 From: denchick Date: Tue, 11 Aug 2026 15:15:35 +0200 Subject: [PATCH 2/6] test: wire dispatcher unit tests into npm test --- package.json | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/package.json b/package.json index 6a6f846..557a154 100644 --- a/package.json +++ b/package.json @@ -23,8 +23,9 @@ "prepublishOnly": "npm run build", "prepare": "husky install", "test:prepare": "npm run build && cd ./examples/demo && npm run build && cd ../../", + "test:unit": "jest -c ./jest/unit.config.js", "test:run": "JEST_TESTCONTAINERS_CONFIG_PATH='./jest/testcontainers-config.js' NODE_TLS_REJECT_UNAUTHORIZED=0 APP_LOGGING_LEVEL='silent' APP_ENV='test' jest -c './jest/jest.config.js' --detectOpenHandles", - "test": "npm run test:prepare && npm run test:run" + "test": "npm run test:prepare && npm run test:unit && npm run test:run" }, "devDependencies": { "@commitlint/cli": "^19.5.0", From 3394fbff73d44f0029c4770e3ae234fbe42828da Mon Sep 17 00:00:00 2001 From: denchick Date: Tue, 11 Aug 2026 15:18:56 +0200 Subject: [PATCH 3/6] refactor: extract proxy topology mode check --- lib/dispatcher.ts | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/lib/dispatcher.ts b/lib/dispatcher.ts index 2735d78..9162e07 100644 --- a/lib/dispatcher.ts +++ b/lib/dispatcher.ts @@ -138,7 +138,7 @@ export class PGDispatcher { get primary() { this.checkConnectionsAvailability(); - if (this.options.topologyMode === 'proxy') { + if (this.isProxyMode) { return this.fastestHealthyConnection.knex; } @@ -164,7 +164,7 @@ export class PGDispatcher { get replica() { this.checkConnectionsAvailability(); - if (this.options.topologyMode === 'proxy') { + if (this.isProxyMode) { return this.fastestHealthyConnection.knex; } @@ -221,7 +221,7 @@ export class PGDispatcher { } private async performCheckupQuery(knex: Knex): Promise { - if (this.options.topologyMode === 'proxy') { + if (this.isProxyMode) { await knex.raw('SELECT 1;').timeout(this.options.healthcheckTimeout); return {pingOk: true, primary: false}; @@ -292,6 +292,10 @@ export class PGDispatcher { return this.connections.filter((c) => c.healthy); } + private get isProxyMode() { + return this.options.topologyMode === 'proxy'; + } + private get fastestHealthyConnection() { return this.healthyConnections.sort((a, b) => a.latency - b.latency)[0]; } From af1c6ac749116802f8fb17b14b94be54c35e083e Mon Sep 17 00:00:00 2001 From: denchick Date: Tue, 11 Aug 2026 15:21:22 +0200 Subject: [PATCH 4/6] fix: omit primary role from proxy status logs --- lib/dispatcher.ts | 3 ++- tests/dispatcher.test.js | 9 +++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/lib/dispatcher.ts b/lib/dispatcher.ts index 9162e07..34851e0 100644 --- a/lib/dispatcher.ts +++ b/lib/dispatcher.ts @@ -202,9 +202,10 @@ export class PGDispatcher { this.logger.info({ message: 'Database current status', data: { + ...(this.isProxyMode ? {topologyMode: this.options.topologyMode} : {}), connections: this.connections.map((c) => ({ host: c.host, - primary: c.primary, + ...(this.isProxyMode ? {} : {primary: c.primary}), healthy: c.healthy, latency: c.latency, })), diff --git a/tests/dispatcher.test.js b/tests/dispatcher.test.js index 7ced52f..9cc720f 100644 --- a/tests/dispatcher.test.js +++ b/tests/dispatcher.test.js @@ -94,6 +94,15 @@ describe('PGDispatcher topology modes', () => { } expect(dispatcher.primary).toBe(fasterProxy); expect(dispatcher.replica).toBe(fasterProxy); + + const statusLog = logger.info.mock.calls.find( + ([message]) => message === 'Database current status', + ); + expect(statusLog[1].topologyMode).toBe('proxy'); + for (const connection of statusLog[1].connections) { + expect(connection).not.toHaveProperty('primary'); + } + expect(loggedErrorMessages(logger)).not.toEqual( expect.arrayContaining([ 'Multiple primary connections detected, something is wrong', From 3d4ff3d045673ae931518f6826328f18ff60d762 Mon Sep 17 00:00:00 2001 From: denchick Date: Tue, 11 Aug 2026 15:24:38 +0200 Subject: [PATCH 5/6] test: fix topology warning assertions --- tests/dispatcher.test.js | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/tests/dispatcher.test.js b/tests/dispatcher.test.js index 9cc720f..76553d1 100644 --- a/tests/dispatcher.test.js +++ b/tests/dispatcher.test.js @@ -103,12 +103,11 @@ describe('PGDispatcher topology modes', () => { expect(connection).not.toHaveProperty('primary'); } - expect(loggedErrorMessages(logger)).not.toEqual( - expect.arrayContaining([ - 'Multiple primary connections detected, something is wrong', - 'No alive replica available, using master for read', - ]), + const errorMessages = loggedErrorMessages(logger); + expect(errorMessages).not.toContain( + 'Multiple primary connections detected, something is wrong', ); + expect(errorMessages).not.toContain('No alive replica available, using master for read'); }); test('the existing unavailable-database error is preserved when all proxies are unhealthy', async () => { From 586c1f2e4c766faf0b70d8acc18c67928bc1dcbd Mon Sep 17 00:00:00 2001 From: denchick Date: Tue, 11 Aug 2026 15:59:30 +0200 Subject: [PATCH 6/6] test: verify default topology warnings --- tests/dispatcher.test.js | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/dispatcher.test.js b/tests/dispatcher.test.js index 76553d1..1518600 100644 --- a/tests/dispatcher.test.js +++ b/tests/dispatcher.test.js @@ -78,6 +78,23 @@ describe('PGDispatcher topology modes', () => { expect(dispatcher.replica).toBe(replica); }); + test('the default mode emits primary/replica topology warnings', async () => { + const firstPrimary = createKnex(successfulCheckup({pg_is_in_recovery: false})); + const secondPrimary = createKnex(successfulCheckup({pg_is_in_recovery: false})); + const {dispatcher, logger} = createDispatcher([firstPrimary, secondPrimary]); + + await dispatcher.ready(); + + expect(dispatcher.primary).toBe(firstPrimary); + expect(dispatcher.replica).toBe(firstPrimary); + + const errorMessages = loggedErrorMessages(logger); + expect(errorMessages).toContain( + 'Multiple primary connections detected, something is wrong', + ); + expect(errorMessages).toContain('No alive replica available, using master for read'); + }); + test('proxy mode routes both roles to the fastest healthy endpoint without topology warnings', async () => { const slowerProxy = createKnex(successfulCheckup({value: 1}, 30)); const fasterProxy = createKnex(successfulCheckup({value: 1}, 5));