diff --git a/README.md b/README.md index 4d1675d..0450221 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,40 @@ npm install @platformatic/leader ## Usage +### Election only + +```js +'use strict' + +const pg = require('pg') +const createLeaderElector = require('@platformatic/leader') + +const pool = new pg.Pool({ + connectionString: 'postgres://localhost/mydb' +}) + +const leader = createLeaderElector({ + pool, + lock: 4242, + onLeadershipChange: (isLeader) => { + if (isLeader) { + console.log('I am the leader — starting work') + } else { + console.log('No longer the leader — stopping work') + } + } +}) + +leader.start() + +console.log(leader.isLeader()) // true or false + +await leader.stop() +await pool.end() +``` + +### With notification channels + ```js 'use strict' @@ -65,7 +99,7 @@ Returns an object with `{ start, stop, notify, isLeader }`. | `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 | +| `channels` | `Array` | No | - | Notification channel configurations (omit for election-only mode) | | `log` | `object` | No | `pino()` | Logger with `info`, `debug`, `warn`, `error` methods | | `onLeadershipChange` | `function` | No | `null` | Callback invoked with `(isLeader: boolean)` when leadership status changes | diff --git a/index.js b/index.js index 1311f2e..3bf70fe 100644 --- a/index.js +++ b/index.js @@ -24,20 +24,21 @@ function createLeaderElector (options) { log.info('Acquiring advisory lock %d', lock) - if (!channels || !Array.isArray(channels) || channels.length === 0) { - throw new Error('channels array is required') - } - - for (const ch of channels) { - if (!ch.channel) { - throw new Error('channel is required for each notification channel') + if (channels) { + if (!Array.isArray(channels)) { + throw new Error('channels must be an array') } - if (!ch.onNotification) { - throw new Error('onNotification is required for each notification channel') + for (const ch of channels) { + if (!ch.channel) { + throw new Error('channel is required for each notification channel') + } + if (!ch.onNotification) { + throw new Error('onNotification is required for each notification channel') + } } } - const notificationChannels = channels + const notificationChannels = channels || [] let elected = false const abortController = new AbortController() @@ -63,37 +64,39 @@ function createLeaderElector (options) { if (leader && !elected) { log.info('This instance is the leader') updateLeadershipStatus(true) - ;(async () => { - for (const ch of notificationChannels) { - await client.query(`LISTEN "${ch.channel}"`) - log.info({ channel: ch.channel }, 'Listening to notification channel') - } - - for await (const notification of on(client, 'notification', { signal: abortController.signal })) { - log.debug({ notification }, 'Received notification') - try { - const msg = notification[0] - const payload = JSON.parse(msg.payload) - const channelName = msg.channel - - const channelConfig = notificationChannels.find(ch => ch.channel === channelName) - if (channelConfig) { - await channelConfig.onNotification(payload) - } else { - log.warn({ channel: channelName }, 'No handler found for notification channel') - } - } catch (err) { - log.warn({ err }, 'error while processing notification') + if (notificationChannels.length > 0) { + ;(async () => { + for (const ch of notificationChannels) { + await client.query(`LISTEN "${ch.channel}"`) + log.info({ channel: ch.channel }, 'Listening to notification channel') } - } - })() - .catch((err) => { - if (err.name !== 'AbortError') { - log.error({ err }, 'Error in notification') - } else { - abortController.abort() + + for await (const notification of on(client, 'notification', { signal: abortController.signal })) { + log.debug({ notification }, 'Received notification') + try { + const msg = notification[0] + const payload = JSON.parse(msg.payload) + const channelName = msg.channel + + const channelConfig = notificationChannels.find(ch => ch.channel === channelName) + if (channelConfig) { + await channelConfig.onNotification(payload) + } else { + log.warn({ channel: channelName }, 'No handler found for notification channel') + } + } catch (err) { + log.warn({ err }, 'error while processing notification') + } } - }) + })() + .catch((err) => { + if (err.name !== 'AbortError') { + log.error({ err }, 'Error in notification') + } else { + abortController.abort() + } + }) + } } else if (leader && elected) { log.debug('This instance is still the leader') } else if (!leader && elected) { diff --git a/test/leader.test.js b/test/leader.test.js index 6597215..e8aa795 100644 --- a/test/leader.test.js +++ b/test/leader.test.js @@ -297,9 +297,10 @@ test('should throw error when required parameters are missing', async () => { createLeaderElector({ pool: {} }) }, { message: 'lock is required' }) - assert.throws(() => { + // channels is optional — no throw without it + assert.doesNotThrow(() => { createLeaderElector({ pool: {}, lock: 123 }) - }, { message: 'channels array is required' }) + }) }) test('should trigger onLeadershipChange callback when leadership changes', async (t) => { @@ -493,4 +494,78 @@ test('should throw error when channels array has invalid entries', async () => { ] }) }, { message: 'channel is required for each notification channel' }) + + assert.throws(() => { + createLeaderElector({ + pool: {}, + lock: 123, + channels: 'not-an-array' + }) + }, { message: 'channels must be an array' }) +}) + +test('election-only mode: works without channels', async (t) => { + const pool = createPool() + const leadershipChanges = [] + + t.after(async () => { + await leader.stop() + await pool.end() + }) + + const leader = createLeaderElector({ + pool, + lock: 70001, + poll: 200, + log: silentLogger, + onLeadershipChange: (isLeader) => { + leadershipChanges.push(isLeader) + } + }) + + leader.start() + await sleep(500) + + assert.ok(leader.isLeader()) + assert.deepStrictEqual(leadershipChanges, [true]) +}) + +test('election-only mode: two instances, only one becomes leader', async (t) => { + const pool1 = createPool() + const pool2 = createPool() + const changes1 = [] + const changes2 = [] + + t.after(async () => { + await leader1.stop() + await leader2.stop() + await pool1.end() + await pool2.end() + }) + + const leader1 = createLeaderElector({ + pool: pool1, + lock: 70002, + poll: 200, + log: silentLogger, + onLeadershipChange: (isLeader) => { changes1.push(isLeader) } + }) + + const leader2 = createLeaderElector({ + pool: pool2, + lock: 70002, + poll: 200, + log: silentLogger, + onLeadershipChange: (isLeader) => { changes2.push(isLeader) } + }) + + leader1.start() + await sleep(500) + leader2.start() + await sleep(500) + + assert.ok(leader1.isLeader()) + assert.ok(!leader2.isLeader()) + assert.deepStrictEqual(changes1, [true]) + assert.deepStrictEqual(changes2, []) }) diff --git a/test/notify-reliability.test.js b/test/notify-reliability.test.js new file mode 100644 index 0000000..454c910 --- /dev/null +++ b/test/notify-reliability.test.js @@ -0,0 +1,229 @@ +'use strict' + +const { test } = require('node:test') +const assert = require('node:assert') +const { setTimeout: sleep } = require('node:timers/promises') +const pg = require('pg') +const createLeaderElector = require('../index') + +const connectionString = process.env.CONNECTION_STRING || 'postgres://postgres:postgres@127.0.0.1:5433/leader' + +const silentLogger = { + info: () => {}, + debug: () => {}, + warn: () => {}, + error: () => {} +} + +function createPool () { + return new pg.Pool({ connectionString }) +} + +test('leader receives notifications sent via pg_notify from a different pool connection', async (t) => { + const pool = createPool() + const receivedNotifications = [] + + t.after(async () => { + await leaderElector.stop() + await pool.end() + }) + + const leaderElector = createLeaderElector({ + pool, + lock: 50001, + poll: 10000, + channels: [ + { + channel: 'test_pg_notify', + onNotification: (payload) => { + receivedNotifications.push(payload) + } + } + ], + log: silentLogger + }) + + leaderElector.start() + await sleep(500) + assert.ok(leaderElector.isLeader()) + + // Send notification using pg_notify from a pool connection (like the workflow queue plugin does) + await pool.query("SELECT pg_notify('test_pg_notify', '{\"msg\":\"hello\"}')") + + await sleep(500) + + assert.strictEqual(receivedNotifications.length, 1) + assert.deepStrictEqual(receivedNotifications[0], { msg: 'hello' }) +}) + +test('leader receives rapid-fire notifications from separate pool connections', async (t) => { + const pool = createPool() + const receivedNotifications = [] + const NOTIFICATION_COUNT = 20 + + t.after(async () => { + await leaderElector.stop() + await pool.end() + }) + + const leaderElector = createLeaderElector({ + pool, + lock: 50002, + poll: 10000, + channels: [ + { + channel: 'test_rapid_notify', + onNotification: (payload) => { + receivedNotifications.push(payload) + } + } + ], + log: silentLogger + }) + + leaderElector.start() + await sleep(500) + + // Send many notifications rapidly from pool connections + for (let i = 0; i < NOTIFICATION_COUNT; i++) { + await pool.query(`SELECT pg_notify('test_rapid_notify', '{"i":${i}}')`) + } + + await sleep(2000) + + assert.strictEqual(receivedNotifications.length, NOTIFICATION_COUNT, + `Expected ${NOTIFICATION_COUNT} notifications but received ${receivedNotifications.length}`) +}) + +test('notifications are received promptly (within 100ms), not delayed by poll interval', async (t) => { + const pool = createPool() + const timestamps = [] + + t.after(async () => { + await leaderElector.stop() + await pool.end() + }) + + const leaderElector = createLeaderElector({ + pool, + lock: 50003, + poll: 10000, // 10 second poll - notifications should NOT wait for this + channels: [ + { + channel: 'test_prompt_notify', + onNotification: () => { + timestamps.push(Date.now()) + } + } + ], + log: silentLogger + }) + + leaderElector.start() + await sleep(500) + + const sendTime = Date.now() + await pool.query("SELECT pg_notify('test_prompt_notify', '{}')") + + await sleep(1000) + + assert.strictEqual(timestamps.length, 1, 'Should have received exactly 1 notification') + const latency = timestamps[0] - sendTime + assert.ok(latency < 100, `Notification latency was ${latency}ms, expected < 100ms`) +}) + +test('notifications sent during leader advisory lock re-check are not lost', async (t) => { + const pool = createPool() + const receivedNotifications = [] + + t.after(async () => { + await leaderElector.stop() + await pool.end() + }) + + // Use a very short poll interval so the leader frequently re-checks the advisory lock + // This means the shared client is frequently busy with pg_try_advisory_lock queries + const leaderElector = createLeaderElector({ + pool, + lock: 50004, + poll: 50, // Very frequent re-checks — maximizes chance of notification during a query + channels: [ + { + channel: 'test_during_recheck', + onNotification: (payload) => { + receivedNotifications.push(payload) + } + } + ], + log: silentLogger + }) + + leaderElector.start() + await sleep(300) + + // Send 50 notifications with small delays, overlapping with advisory lock re-checks + const TOTAL = 50 + for (let i = 0; i < TOTAL; i++) { + await pool.query(`SELECT pg_notify('test_during_recheck', '{"i":${i}}')`) + await sleep(10) // Small delay to spread notifications across multiple poll cycles + } + + await sleep(1000) + + assert.strictEqual(receivedNotifications.length, TOTAL, + `Expected ${TOTAL} notifications but received ${receivedNotifications.length}. ` + + `Missing: ${TOTAL - receivedNotifications.length}`) +}) + +test('onNotification that fires-and-forgets an async function does not block subsequent notifications', async (t) => { + const pool = createPool() + const receivedNotifications = [] + + t.after(async () => { + await leaderElector.stop() + await pool.end() + }) + + // Simulate what the workflow poller does: onNotification calls an async function + // but does NOT return the promise (fire-and-forget) + async function execute () { + // Simulate some async work + await sleep(50) + } + + const leaderElector = createLeaderElector({ + pool, + lock: 50005, + poll: 10000, + channels: [ + { + channel: 'test_fire_forget', + onNotification: () => { + // Fire-and-forget — does NOT return the promise + // This is exactly what the workflow poller does: + // onNotification: () => { execute() } + execute() + receivedNotifications.push(Date.now()) + } + } + ], + log: silentLogger + }) + + leaderElector.start() + await sleep(500) + + // Send 5 notifications rapidly + for (let i = 0; i < 5; i++) { + await pool.query("SELECT pg_notify('test_fire_forget', '{}')") + } + + await sleep(2000) + + assert.strictEqual(receivedNotifications.length, 5, + `Expected 5 notifications but received ${receivedNotifications.length}`) + + // All notifications should arrive promptly, not delayed by the async work + const maxGap = Math.max(...receivedNotifications.slice(1).map((ts, i) => ts - receivedNotifications[i])) + assert.ok(maxGap < 200, `Max gap between notifications was ${maxGap}ms, expected < 200ms`) +}) diff --git a/test/poller-pattern.test.js b/test/poller-pattern.test.js new file mode 100644 index 0000000..7fee2e9 --- /dev/null +++ b/test/poller-pattern.test.js @@ -0,0 +1,407 @@ +'use strict' + +const { test } = require('node:test') +const assert = require('node:assert') +const { setTimeout: sleep } = require('node:timers/promises') +const pg = require('pg') +const createLeaderElector = require('../index') + +const connectionString = process.env.CONNECTION_STRING || 'postgres://postgres:postgres@127.0.0.1:5433/leader' + +const silentLogger = { + info: () => {}, + debug: () => {}, + warn: () => {}, + error: () => {} +} + +function createPool () { + return new pg.Pool({ connectionString }) +} + +// Simulates the workflow poller pattern exactly as used in platformatic-world +test('poller pattern: deferred timer wakes execute after notification-triggered cycle', async (t) => { + const pool = createPool() + const dispatches = [] + + t.after(async () => { + stopped = true + if (deferredTimer) clearTimeout(deferredTimer) + await leader.stop() + await pool.end() + }) + + let stopped = false + let executing = false + let pendingNotify = false + let deferredTimer = null + + async function execute () { + if (stopped || !leader.isLeader()) return + + if (executing) { + pendingNotify = true + return + } + executing = true + + try { + await executeOnce() + } finally { + executing = false + if (pendingNotify && !stopped && leader.isLeader()) { + pendingNotify = false + setImmediate(() => execute()) + } + } + } + + async function executeOnce () { + // Simulate: check for deferred messages, dispatch them + dispatches.push(Date.now()) + + // Simulate: dispatch creates a deferred message that should fire in 200ms + // This is the key pattern: scheduleNextWakeup is called WITHOUT await + scheduleNextWakeup(200) + } + + // Simulate scheduleNextWakeup — fire-and-forget async function + async function scheduleNextWakeup (ms) { + if (deferredTimer) { + clearTimeout(deferredTimer) + deferredTimer = null + } + + // Simulate the pool.query to find deferred messages + await pool.query('SELECT 1') + + if (!stopped) { + deferredTimer = setTimeout(() => { + deferredTimer = null + execute() + }, ms) + } + } + + const leader = createLeaderElector({ + pool, + lock: 60001, + poll: 10000, + channels: [{ + channel: 'poller_test_1', + onNotification: () => { execute() } + }], + log: silentLogger, + onLeadershipChange: (isLeader) => { + if (isLeader) execute() + } + }) + + leader.start() + await sleep(300) + assert.ok(leader.isLeader()) + + // Initial execute from onLeadershipChange plus deferred chain (200ms each) + assert.ok(dispatches.length >= 1, 'Should have at least 1 dispatch') + + const countAfterInit = dispatches.length + await sleep(500) + + // The deferred timer chain should have continued + assert.ok(dispatches.length > countAfterInit, + `Expected dispatches to increase from ${countAfterInit} but got ${dispatches.length}`) +}) + +test('poller pattern: notification sent during executeOnce triggers re-execution', async (t) => { + const pool = createPool() + const dispatches = [] + + t.after(async () => { + stopped = true + if (deferredTimer) clearTimeout(deferredTimer) + await leader.stop() + await pool.end() + }) + + let stopped = false + let executing = false + let pendingNotify = false + const deferredTimer = null + + async function execute () { + if (stopped || !leader.isLeader()) return + + if (executing) { + pendingNotify = true + return + } + executing = true + + try { + await executeOnce() + } finally { + executing = false + if (pendingNotify && !stopped && leader.isLeader()) { + pendingNotify = false + setImmediate(() => execute()) + } + } + } + + async function executeOnce () { + dispatches.push(Date.now()) + await sleep(50) + } + + const leader = createLeaderElector({ + pool, + lock: 60002, + poll: 10000, + channels: [{ + channel: 'poller_test_2', + onNotification: () => { execute() } + }], + log: silentLogger, + onLeadershipChange: (isLeader) => { + if (isLeader) execute() + } + }) + + leader.start() + await sleep(500) + + const countBefore = dispatches.length + + // Send a notification — should trigger execute + await pool.query("SELECT pg_notify('poller_test_2', '{}')") + + await sleep(500) + + assert.ok(dispatches.length > countBefore, + `Expected dispatches to increase from ${countBefore} but got ${dispatches.length}`) +}) + +test('poller pattern: simulates exact workflow failure scenario', async (t) => { + // This test simulates the exact scenario from the CI failure: + // 1. Initial batch dispatches messages + // 2. One dispatch returns timeoutSeconds=1, creating a "deferred" wakeup + // 3. No more notifications arrive (tests are polling, not queuing) + // 4. The deferred timer must fire to continue processing + + const pool = createPool() + const dispatches = [] + + t.after(async () => { + stopped = true + if (deferredTimer) clearTimeout(deferredTimer) + await leader.stop() + await pool.end() + }) + + let stopped = false + let executing = false + let pendingNotify = false + let deferredTimer = null + let cycle = 0 + + async function execute () { + if (stopped || !leader.isLeader()) return + + if (executing) { + pendingNotify = true + return + } + executing = true + + try { + await executeOnce() + } finally { + executing = false + if (pendingNotify && !stopped && leader.isLeader()) { + pendingNotify = false + setImmediate(() => execute()) + } + } + } + + async function executeOnce () { + cycle++ + const currentCycle = cycle + dispatches.push({ cycle: currentCycle, time: Date.now() }) + + // Simulate dispatch that takes some time + await sleep(20) + + // On every cycle, simulate creating a deferred message (like timeoutSeconds=1) + // scheduleNextWakeup is called WITHOUT await — this is the exact pattern from the poller + if (currentCycle <= 3) { + scheduleNextWakeup(200) // 200ms deferred + } + } + + async function scheduleNextWakeup (ms) { + if (deferredTimer) { + clearTimeout(deferredTimer) + deferredTimer = null + } + + // Simulate pool.query to check for deferred messages + await pool.query('SELECT 1') + + if (!stopped) { + deferredTimer = setTimeout(() => { + deferredTimer = null + execute() + }, ms) + } + } + + const leader = createLeaderElector({ + pool, + lock: 60003, + poll: 10000, + channels: [{ + channel: 'poller_test_3', + onNotification: () => { execute() } + }], + log: silentLogger, + onLeadershipChange: (isLeader) => { + if (isLeader) execute() + } + }) + + leader.start() + await sleep(500) + + // By now, the chain should be: + // cycle 1 (from onLeadershipChange) → sets deferred timer 200ms + // cycle 2 (from deferred timer) → sets deferred timer 200ms + // cycle 3 (from deferred timer) → sets deferred timer 200ms + // cycle 4 (from deferred timer) → does NOT set timer (cycle > 3) + + // Wait enough time for all cycles + await sleep(1500) + + assert.ok(dispatches.length >= 4, + `Expected at least 4 dispatch cycles but got ${dispatches.length}: ` + + JSON.stringify(dispatches.map(d => ({ cycle: d.cycle, elapsed: d.time - dispatches[0].time })))) + + // Verify timing: each cycle should be ~200ms apart (not 10 seconds) + for (let i = 1; i < Math.min(dispatches.length, 4); i++) { + const gap = dispatches[i].time - dispatches[i - 1].time + assert.ok(gap < 1000, + `Gap between cycle ${dispatches[i - 1].cycle} and ${dispatches[i].cycle} was ${gap}ms, expected < 1000ms`) + } +}) + +test('poller pattern: race between scheduleNextWakeup and pendingNotify re-execution', async (t) => { + // This tests the race condition where: + // 1. executeOnce calls scheduleNextWakeup (fire-and-forget) + // 2. A NOTIFY arrives during executeOnce, setting pendingNotify + // 3. executeOnce finishes, pendingNotify triggers re-execution + // 4. The re-execution's scheduleNextWakeup might clear the timer from step 1 + + const pool = createPool() + const dispatches = [] + + t.after(async () => { + stopped = true + if (deferredTimer) clearTimeout(deferredTimer) + await leader.stop() + await pool.end() + }) + + let stopped = false + let executing = false + let pendingNotify = false + let deferredTimer = null + + async function execute () { + if (stopped || !leader.isLeader()) return + + if (executing) { + pendingNotify = true + return + } + executing = true + + try { + await executeOnce() + } finally { + executing = false + if (pendingNotify && !stopped && leader.isLeader()) { + pendingNotify = false + setImmediate(() => execute()) + } + } + } + + let shouldCreateDeferred = false + + async function executeOnce () { + dispatches.push({ time: Date.now(), reason: shouldCreateDeferred ? 'deferred' : 'trigger' }) + + // Simulate dispatch work + await sleep(30) + + if (shouldCreateDeferred) { + shouldCreateDeferred = false + // Fire-and-forget scheduleNextWakeup + scheduleNextWakeup(300) + } + } + + async function scheduleNextWakeup (ms) { + if (deferredTimer) { + clearTimeout(deferredTimer) + deferredTimer = null + } + await pool.query('SELECT 1') + if (!stopped) { + deferredTimer = setTimeout(() => { + deferredTimer = null + shouldCreateDeferred = true + execute() + }, ms) + } + } + + const leader = createLeaderElector({ + pool, + lock: 60004, + poll: 10000, + channels: [{ + channel: 'poller_test_4', + onNotification: () => { execute() } + }], + log: silentLogger, + onLeadershipChange: (isLeader) => { + if (isLeader) { + shouldCreateDeferred = true + execute() + } + } + }) + + leader.start() + await sleep(300) + + // At this point, initial execute ran, set deferred timer for 300ms + // Send a NOTIFY while the timer is pending — this creates the race condition + await pool.query("SELECT pg_notify('poller_test_4', '{}')") + + // The NOTIFY triggers execute() (either directly or via pendingNotify) + // Then the deferred timer also fires + // Both should result in execute() calls + + await sleep(2000) + + // We should have at least 3 dispatches: + // 1. Initial (from onLeadershipChange) - creates deferred + // 2. From NOTIFY + // 3. From deferred timer - creates deferred + // 4. From second deferred timer + assert.ok(dispatches.length >= 3, + `Expected at least 3 dispatches but got ${dispatches.length}: ` + + JSON.stringify(dispatches.map(d => ({ ...d, elapsed: d.time - dispatches[0].time })))) +})