From 1a267e65adffc5ede5888e2ca42cbdd1be0b6eca Mon Sep 17 00:00:00 2001 From: marcopiraccini Date: Sat, 7 Mar 2026 11:40:29 +0100 Subject: [PATCH] Replace @databases/pg with pg, make log optional with pino default - Use pg.Pool directly instead of @databases/pg connection pool - Default logger to pino() when not provided - Add docker-compose.yml for local PostgreSQL - Update tests and README --- README.md | 30 +++++++--- docker-compose.yml | 9 +++ index.js | 37 ++++++------ package.json | 4 +- test/leader.test.js | 136 ++++++++++++++++---------------------------- 5 files changed, 99 insertions(+), 117 deletions(-) create mode 100644 docker-compose.yml diff --git a/README.md b/README.md index 45165f1..4d1675d 100644 --- a/README.md +++ b/README.md @@ -15,16 +15,15 @@ npm install @platformatic/leader ```js 'use strict' -const createConnectionPool = require('@databases/pg') +const pg = require('pg') const createLeaderElector = require('@platformatic/leader') -const db = createConnectionPool({ - connectionString: 'postgres://localhost/mydb', - bigIntMode: 'bigint' +const pool = new pg.Pool({ + connectionString: 'postgres://localhost/mydb' }) const leader = createLeaderElector({ - db, + pool, lock: 4242, poll: 10000, channels: [ @@ -35,7 +34,6 @@ const leader = createLeaderElector({ } } ], - log: console, onLeadershipChange: (isLeader) => { console.log('Leadership changed:', isLeader) } @@ -51,7 +49,7 @@ console.log(leader.isLeader()) // Stop the leader elector await leader.stop() -await db.dispose() +await pool.end() ``` ## API @@ -64,11 +62,11 @@ Returns an object with `{ start, stop, notify, isLeader }`. | Option | Type | Required | Default | Description | |---|---|---|---|---| -| `db` | `ConnectionPool` | Yes | - | A `@databases/pg` connection pool | +| `pool` | `pg.Pool` | Yes | - | A [node-postgres](https://node-postgres.com/) pool instance | | `lock` | `number` | Yes | - | PostgreSQL advisory lock ID | | `poll` | `number` | No | `10000` | Polling interval in milliseconds | | `channels` | `Array` | Yes | - | Notification channel configurations | -| `log` | `object` | Yes | - | Logger with `info`, `debug`, `warn`, `error` methods | +| `log` | `object` | No | `pino()` | Logger with `info`, `debug`, `warn`, `error` methods | | `onLeadershipChange` | `function` | No | `null` | Callback invoked with `(isLeader: boolean)` when leadership status changes | Each channel in the `channels` array must have: @@ -90,6 +88,20 @@ Each channel in the `channels` array must have: 4. If the leader stops or loses the lock, another instance acquires it and takes over. 5. The `onLeadershipChange` callback fires whenever the leadership status transitions. +## Development + +Start PostgreSQL: + +``` +docker compose up -d +``` + +Run tests: + +``` +node --test test/*.test.js +``` + ## License Apache-2.0 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..a9973bf --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,9 @@ +services: + postgres: + ports: + - "127.0.0.1:5433:5432" + image: "postgres" + environment: + POSTGRES_PASSWORD: postgres + POSTGRES_DB: leader + PGDATA: /data/postgres diff --git a/index.js b/index.js index d39ecc5..1311f2e 100644 --- a/index.js +++ b/index.js @@ -2,30 +2,27 @@ const { scheduler } = require('timers/promises') const { on } = require('events') +const pino = require('pino') function createLeaderElector (options) { const { - db, + pool, lock, poll = 10000, channels, - log, + log = pino(), onLeadershipChange = null } = options - log.info('Acquiring advisory lock %d', lock) - - if (!db) { - throw new Error('db is required') + if (!pool) { + throw new Error('pool is required') } if (!lock) { throw new Error('lock is required') } - if (!log) { - throw new Error('log is required') - } + log.info('Acquiring advisory lock %d', lock) if (!channels || !Array.isArray(channels) || channels.length === 0) { throw new Error('channels array is required') @@ -56,22 +53,23 @@ function createLeaderElector (options) { } async function amITheLeader () { - const sql = db.sql - await db.task(async (t) => { + const client = await pool.connect() + try { while (!abortController.signal.aborted) { - const [{ leader }] = await t.query(sql` - SELECT pg_try_advisory_lock(${lock}) as leader; - `) + const { rows: [{ leader }] } = await client.query( + 'SELECT pg_try_advisory_lock($1) as leader', + [lock] + ) if (leader && !elected) { log.info('This instance is the leader') updateLeadershipStatus(true) ;(async () => { for (const ch of notificationChannels) { - await t.query(sql.__dangerous__rawValue(`LISTEN "${ch.channel}";`)) + await client.query(`LISTEN "${ch.channel}"`) log.info({ channel: ch.channel }, 'Listening to notification channel') } - for await (const notification of on(t._driver.client, 'notification', { signal: abortController.signal })) { + for await (const notification of on(client, 'notification', { signal: abortController.signal })) { log.debug({ notification }, 'Received notification') try { const msg = notification[0] @@ -110,7 +108,9 @@ function createLeaderElector (options) { break } } - }) + } finally { + client.release() + } log.debug('leader loop stopped') } @@ -140,8 +140,7 @@ function createLeaderElector (options) { payload = JSON.stringify(payload) - const sql = db.sql - await db.query(sql.__dangerous__rawValue(`NOTIFY "${channelName}", '${payload}';`)) + await pool.query(`NOTIFY "${channelName}", '${payload}'`) } async function stop () { diff --git a/package.json b/package.json index 1154fe6..01f17c2 100644 --- a/package.json +++ b/package.json @@ -8,10 +8,10 @@ "lint": "standard ." }, "dependencies": { - "@databases/pg": "^5.5.0" + "pg": "^8.20.0", + "pino": "^10.3.1" }, "devDependencies": { - "pg": "^8.13.3", "standard": "^17.1.2" }, "engines": { diff --git a/test/leader.test.js b/test/leader.test.js index a61a995..6597215 100644 --- a/test/leader.test.js +++ b/test/leader.test.js @@ -4,10 +4,9 @@ const { test } = require('node:test') const assert = require('node:assert') const { setTimeout: sleep } = require('node:timers/promises') const pg = require('pg') -const createConnectionPool = require('@databases/pg') const createLeaderElector = require('../index') -const connectionString = process.env.CONNECTION_STRING || 'postgres://postgres:postgres@127.0.0.1:5433/scaler' +const connectionString = process.env.CONNECTION_STRING || 'postgres://postgres:postgres@127.0.0.1:5433/leader' const silentLogger = { info: () => {}, @@ -16,18 +15,19 @@ const silentLogger = { error: () => {} } +function createPool () { + return new pg.Pool({ connectionString }) +} + test('should notify through PostgreSQL notification', async (t) => { const listenClient = new pg.Client(connectionString) await listenClient.connect() - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() t.after(async () => { await listenClient.end() - await pool.dispose() + await pool.end() }) let notificationReceived = false @@ -43,7 +43,7 @@ test('should notify through PostgreSQL notification', async (t) => { await listenClient.query('LISTEN "test_channel"') const leaderElection = createLeaderElector({ - db: pool, + pool, lock: 9999, channels: [ { @@ -67,14 +67,11 @@ test('leaderElector notifies through PostgreSQL notification with an object', as const listenClient = new pg.Client(connectionString) await listenClient.connect() - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() t.after(async () => { await listenClient.end() - await pool.dispose() + await pool.end() }) let notificationReceived = false @@ -90,7 +87,7 @@ test('leaderElector notifies through PostgreSQL notification with an object', as await listenClient.query('LISTEN "test_channel"') const leaderElection = createLeaderElector({ - db: pool, + pool, lock: 9999, channels: [ { @@ -114,18 +111,15 @@ test('leaderElector properly passes payload to callback', async (t) => { let callbackCount = 0 let callbackPayload = null - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() t.after(async () => { await leaderElector.stop() - await pool.dispose() + await pool.end() }) const leaderElector = createLeaderElector({ - db: pool, + pool, lock: 9999, poll: 200, channels: [ @@ -154,24 +148,16 @@ test('leaderElector properly passes payload to callback', async (t) => { }) test('if one instance is shut down, the other is elected', async (t) => { - const pool1 = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) - - const pool2 = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool1 = createPool() + const pool2 = createPool() const lockId = Math.floor(Math.random() * 1000) + 7000 - const sql = pool1.sql - await pool1.query(sql`SELECT pg_advisory_unlock_all();`) - await pool2.query(sql`SELECT pg_advisory_unlock_all();`) + await pool1.query('SELECT pg_advisory_unlock_all()') + await pool2.query('SELECT pg_advisory_unlock_all()') const leaderElector1 = createLeaderElector({ - db: pool1, + pool: pool1, lock: lockId, poll: 200, channels: [ @@ -188,7 +174,7 @@ test('if one instance is shut down, the other is elected', async (t) => { assert.ok(leaderElector1.isLeader()) const leaderElector2 = createLeaderElector({ - db: pool2, + pool: pool2, lock: lockId, poll: 200, channels: [ @@ -208,46 +194,38 @@ test('if one instance is shut down, the other is elected', async (t) => { await leaderElector1.stop() await sleep(500) - await pool1.dispose() + await pool1.end() await sleep(500) assert.ok(leaderElector2.isLeader()) await leaderElector2.stop() - await pool2.dispose() + await pool2.end() }) test('only the leader instance executes notification callbacks and leadership transfers properly', async (t) => { - const pool1 = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) - - const pool2 = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool1 = createPool() + const pool2 = createPool() const lockId = Math.floor(Math.random() * 1000) + 7000 const testChannel = `test_leader_notifications_${lockId}` t.after(async () => { try { - await pool1.dispose().catch(() => {}) - await pool2.dispose().catch(() => {}) + await pool1.end().catch(() => {}) + await pool2.end().catch(() => {}) } catch {} }) - const sql = pool1.sql - await pool1.query(sql`SELECT pg_advisory_unlock_all();`) - await pool2.query(sql`SELECT pg_advisory_unlock_all();`) + await pool1.query('SELECT pg_advisory_unlock_all()') + await pool2.query('SELECT pg_advisory_unlock_all()') let instance1Notifications = 0 let instance2Notifications = 0 const leaderElector1 = createLeaderElector({ - db: pool1, + pool: pool1, lock: lockId, poll: 100, channels: [ @@ -262,7 +240,7 @@ test('only the leader instance executes notification callbacks and leadership tr }) const leaderElector2 = createLeaderElector({ - db: pool2, + pool: pool2, lock: lockId, poll: 100, channels: [ @@ -294,7 +272,7 @@ test('only the leader instance executes notification callbacks and leadership tr await leaderElector1.stop() await sleep(500) - await pool1.dispose() + await pool1.end() await sleep(500) assert.ok(leaderElector2.isLeader()) @@ -312,32 +290,25 @@ test('only the leader instance executes notification callbacks and leadership tr test('should throw error when required parameters are missing', async () => { assert.throws(() => { - createLeaderElector({ log: { info: () => {} }, lock: 123 }) - }, { message: 'db is required' }) + createLeaderElector({ lock: 123 }) + }, { message: 'pool is required' }) assert.throws(() => { - createLeaderElector({ db: {}, log: { info: () => {} } }) + createLeaderElector({ pool: {} }) }, { message: 'lock is required' }) assert.throws(() => { - createLeaderElector({ db: {}, lock: 123, log: { info: () => {} } }) + createLeaderElector({ pool: {}, lock: 123 }) }, { message: 'channels array is required' }) - - assert.throws(() => { - createLeaderElector({}) - }) }) test('should trigger onLeadershipChange callback when leadership changes', async (t) => { - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() const leadershipChanges = [] const leaderElector = createLeaderElector({ - db: pool, + pool, lock: 8888, poll: 200, channels: [ @@ -360,18 +331,15 @@ test('should trigger onLeadershipChange callback when leadership changes', async assert.strictEqual(leadershipChanges[0], true) await leaderElector.stop() - await pool.dispose() + await pool.end() }) test('should handle errors in onNotification callback', async (t) => { - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() const logMessages = [] const leaderElector = createLeaderElector({ - db: pool, + pool, lock: 7777, poll: 200, channels: [ @@ -397,7 +365,7 @@ test('should handle errors in onNotification callback', async (t) => { await sleep(500) await leaderElector.stop() - await pool.dispose() + await pool.end() const warnLog = logMessages.find(log => log.level === 'warn' && log.data.err) assert.ok(warnLog) @@ -405,16 +373,13 @@ test('should handle errors in onNotification callback', async (t) => { }) test('should support multiple notification channels', async (t) => { - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() const channel1Notifications = [] const channel2Notifications = [] const leaderElector = createLeaderElector({ - db: pool, + pool, lock: 9999, poll: 200, channels: [ @@ -447,7 +412,7 @@ test('should support multiple notification channels', async (t) => { await sleep(300) await leaderElector.stop() - await pool.dispose() + await pool.end() assert.strictEqual(channel1Notifications.length, 2) assert.strictEqual(channel1Notifications[0].message, 'hello from channel 1') @@ -458,15 +423,12 @@ test('should support multiple notification channels', async (t) => { }) test('should route notifications to correct channel handler', async (t) => { - const pool = createConnectionPool({ - connectionString, - bigIntMode: 'bigint' - }) + const pool = createPool() const receivedNotifications = [] const leaderElector = createLeaderElector({ - db: pool, + pool, lock: 10000, poll: 200, channels: [ @@ -497,7 +459,7 @@ test('should route notifications to correct channel handler', async (t) => { await sleep(300) await leaderElector.stop() - await pool.dispose() + await pool.end() assert.strictEqual(receivedNotifications.length, 3) assert.strictEqual(receivedNotifications[0].channel, 'a') @@ -511,7 +473,7 @@ test('should route notifications to correct channel handler', async (t) => { test('should throw error when channels array has invalid entries', async () => { assert.throws(() => { createLeaderElector({ - db: {}, + pool: {}, lock: 123, log: { info: () => {} }, channels: [ @@ -523,7 +485,7 @@ test('should throw error when channels array has invalid entries', async () => { assert.throws(() => { createLeaderElector({ - db: {}, + pool: {}, lock: 123, log: { info: () => {} }, channels: [