feat: add onConnectionError callback option

Add optional callback for handling terminal reconnection failures
that occur asynchronously after a connection drop. The callback is
only invoked for async errors — initial connect() failures are
handled via the rejected promise directly.

Also fixes Error cause being passed incorrectly (now uses { cause: err })
and ensures _reconnect() promise rejections are properly caught.

Author: aldis-ameriks

README.md

@@ -71,6 +71,7 @@ import PGPubSub from 'pg-notify'
 - `options` (`object`) Configuration options for pg-notify pubsub instance. Accepts same options as [pg](https://github.com/brianc/node-postgres) with few custom ones described below.
     - reconnectMaxRetries (`number`) Maximum number of reconnect attempts after losing connection. Pass `0` to disable reconnecting. Default: `10`.
     - maxPayloadSize (`number`) Maximum payload size, exceeding given size will throw an error. Default: `7999` ([In the default configuration it must be shorter than 8000 bytes.](https://www.postgresql.org/docs/current/sql-notify.html)).
+    - onConnectionError (`function`) Optional callback invoked with the error when reconnection fails after exhausting all retries. Only called for errors that occur asynchronously after a connection drop, preventing the process from crashing with an uncaught exception. Not called during the initial `connect()` — handle that via the rejected promise directly.
 
 ### emit(channel, payload)
 - `channel` (`string`)

cjs/src/pg-notify.js

@@ -21,6 +21,10 @@ class PGPubSub {
 
     this.reconnectMaxRetries = typeof opts.reconnectMaxRetries !== 'undefined' ? opts.reconnectMaxRetries : 10
     this.maxPayloadSize = opts.maxPayloadSize || 7999 // default on a standard pg installation
+    if (opts.onConnectionError && typeof opts.onConnectionError !== 'function') {
+      throw new TypeError('onConnectionError must be a function')
+    }
+    this.onConnectionError = opts.onConnectionError || null
 
     this.state = states.init
     this.reconnectRetries = 0
@@ -127,7 +131,7 @@ class PGPubSub {
     } catch (err) {
       if (this.reconnectRetries >= this.reconnectMaxRetries) {
         await this.close()
-        throw new Error('[PGPubSub]: max reconnect attempts reached, aborting', err)
+        throw new Error('[PGPubSub]: max reconnect attempts reached, aborting', { cause: err })
       }
       if (![states.closing, states.connected].includes((this.state))) {
         await sleep(10)
@@ -155,10 +159,21 @@ class PGPubSub {
 
       if (this.reconnectRetries > this.reconnectMaxRetries) {
         this.close()
-        throw new Error('[PGPubSub]: max reconnect attempts reached, aborting', err)
+        const error = new Error('[PGPubSub]: max reconnect attempts reached, aborting', { cause: err })
+        if (this.onConnectionError) {
+          try { this.onConnectionError(error) } catch (e) { this._debug('[onConnectionError] callback error', e) }
+          return
+        }
+        throw error
       }
 
-      this._reconnect()
+      this._reconnect().catch(reconnectError => {
+        if (this.onConnectionError) {
+          try { this.onConnectionError(reconnectError) } catch (e) { this._debug('[onConnectionError] callback error', e) }
+          return
+        }
+        process.nextTick(() => { throw reconnectError })
+      })
     })
 
     this._debug('[_setupClient] init listeners')

cjs/test/pg-notify.js

@@ -16,12 +16,16 @@ const opts = {
   ssl: process.env.DB_SSL === 'true' ? { rejectUnauthorized: false } : false
 }
 
-async function waitUntilTrue (cb) {
+async function waitUntilTrue (cb, timeout = 5000) {
+  const start = Date.now()
   while (true) {
     const result = cb()
     if (result) {
       break
     }
+    if (Date.now() - start > timeout) {
+      throw new Error('waitUntilTrue timed out')
+    }
     await sleep(1)
   }
 }
@@ -122,6 +126,13 @@ test('works with concurrent emits', async (t) => {
   await waitUntilTrue(() => emitCount === expected)
 })
 
+test('throws TypeError when onConnectionError is not a function', (t) => {
+  t.throws(() => new PGPubSub({ ...opts, onConnectionError: 'not-a-function' }), {
+    instanceOf: TypeError,
+    message: 'onConnectionError must be a function'
+  })
+})
+
 test('retries and throws when initial connection fails', async (t) => {
   const pubsub = new PGPubSub({
     ...opts,
@@ -142,6 +153,28 @@ test('retries and throws when initial connection fails', async (t) => {
   })
 })
 
+test('onConnectionError is not called during initial connect failure', async (t) => {
+  const errors = []
+  const pubsub = new PGPubSub({
+    ...opts,
+    reconnectMaxRetries: 2,
+    host: 'xxx',
+    onConnectionError: (err) => {
+      errors.push(err)
+    }
+  })
+
+  const err = await t.throwsAsync(() => pubsub.connect())
+  t.is(err.message, '[PGPubSub]: max reconnect attempts reached, aborting')
+
+  t.is(errors.length, 0)
+  t.is(pubsub.state, 'closing')
+
+  t.teardown(() => {
+    pubsub.close()
+  })
+})
+
 test('connection can be re-established', async (t) => {
   const channel = getChannel()
   const pubsub = new PGPubSub({
@@ -240,6 +273,78 @@ test('connection cannot be re-established', async (t) => {
   })
 })
 
+test('onConnectionError callback receives error when retries exhausted', async (t) => {
+  const channel = getChannel()
+  const errors = []
+  const pubsub = new PGPubSub({
+    ...opts,
+    reconnectMaxRetries: 1,
+    onConnectionError: (err) => {
+      errors.push(err)
+    }
+  })
+
+  await pubsub.connect()
+  t.is(pubsub.state, 'connected')
+
+  await new Promise(resolve => {
+    pubsub.on(channel, (payload) => {
+      t.deepEqual(payload, 'this-is-the-payload')
+      resolve()
+    })
+    pubsub.emit(channel, 'this-is-the-payload')
+  })
+
+  // emulate pg client emitting an error after retries exceeded
+  pubsub.reconnectRetries = 2
+  pubsub.client.emit('error', new Error('connection reset'))
+
+  await sleep(10)
+
+  await waitUntilTrue(() => pubsub.state === 'closing')
+  t.is(errors.length, 1)
+  t.is(errors[0].message, '[PGPubSub]: max reconnect attempts reached, aborting')
+
+  t.teardown(() => {
+    pubsub.close()
+  })
+})
+
+test('onConnectionError callback receives error when reconnect exhausts retries', async (t) => {
+  const channel = getChannel()
+  const errors = []
+  const pubsub = new PGPubSub({
+    ...opts,
+    reconnectMaxRetries: 1,
+    onConnectionError: (err) => {
+      errors.push(err)
+    }
+  })
+
+  await pubsub.connect()
+  t.is(pubsub.state, 'connected')
+
+  await new Promise(resolve => {
+    pubsub.on(channel, (payload) => {
+      t.deepEqual(payload, 'this-is-the-payload')
+      resolve()
+    })
+    pubsub.emit(channel, 'this-is-the-payload')
+  })
+
+  // point host to invalid address so _reconnect() will fail and exhaust retries
+  pubsub.opts.host = 'xxx'
+  pubsub.client.emit('error', new Error('connection reset'))
+
+  await waitUntilTrue(() => errors.length === 1)
+  t.is(errors[0].message, '[PGPubSub]: max reconnect attempts reached, aborting')
+  t.is(pubsub.state, 'closing')
+
+  t.teardown(() => {
+    pubsub.close()
+  })
+})
+
 test('disabled reconnection with zero', async (t) => {
   const channel = getChannel()
   const pubsub = new PGPubSub({

src/pg-notify.js

@@ -21,6 +21,10 @@ class PGPubSub {
 
     this.reconnectMaxRetries = typeof opts.reconnectMaxRetries !== 'undefined' ? opts.reconnectMaxRetries : 10
     this.maxPayloadSize = opts.maxPayloadSize || 7999 // default on a standard pg installation
+    if (opts.onConnectionError && typeof opts.onConnectionError !== 'function') {
+      throw new TypeError('onConnectionError must be a function')
+    }
+    this.onConnectionError = opts.onConnectionError || null
 
     this.state = states.init
     this.reconnectRetries = 0
@@ -127,7 +131,7 @@ class PGPubSub {
     } catch (err) {
       if (this.reconnectRetries >= this.reconnectMaxRetries) {
         await this.close()
-        throw new Error('[PGPubSub]: max reconnect attempts reached, aborting', err)
+        throw new Error('[PGPubSub]: max reconnect attempts reached, aborting', { cause: err })
       }
       if (![states.closing, states.connected].includes((this.state))) {
         await sleep(10)
@@ -155,10 +159,21 @@ class PGPubSub {
 
       if (this.reconnectRetries > this.reconnectMaxRetries) {
         this.close()
-        throw new Error('[PGPubSub]: max reconnect attempts reached, aborting', err)
+        const error = new Error('[PGPubSub]: max reconnect attempts reached, aborting', { cause: err })
+        if (this.onConnectionError) {
+          try { this.onConnectionError(error) } catch (e) { this._debug('[onConnectionError] callback error', e) }
+          return
+        }
+        throw error
       }
 
-      this._reconnect()
+      this._reconnect().catch(reconnectError => {
+        if (this.onConnectionError) {
+          try { this.onConnectionError(reconnectError) } catch (e) { this._debug('[onConnectionError] callback error', e) }
+          return
+        }
+        process.nextTick(() => { throw reconnectError })
+      })
     })
 
     this._debug('[_setupClient] init listeners')

test/pg-notify.js

@@ -16,12 +16,16 @@ const opts = {
   ssl: process.env.DB_SSL === 'true' ? { rejectUnauthorized: false } : false
 }
 
-async function waitUntilTrue (cb) {
+async function waitUntilTrue (cb, timeout = 5000) {
+  const start = Date.now()
   while (true) {
     const result = cb()
     if (result) {
       break
     }
+    if (Date.now() - start > timeout) {
+      throw new Error('waitUntilTrue timed out')
+    }
     await sleep(1)
   }
 }
@@ -122,6 +126,13 @@ test('works with concurrent emits', async (t) => {
   await waitUntilTrue(() => emitCount === expected)
 })
 
+test('throws TypeError when onConnectionError is not a function', (t) => {
+  t.throws(() => new PGPubSub({ ...opts, onConnectionError: 'not-a-function' }), {
+    instanceOf: TypeError,
+    message: 'onConnectionError must be a function'
+  })
+})
+
 test('retries and throws when initial connection fails', async (t) => {
   const pubsub = new PGPubSub({
     ...opts,
@@ -142,6 +153,28 @@ test('retries and throws when initial connection fails', async (t) => {
   })
 })
 
+test('onConnectionError is not called during initial connect failure', async (t) => {
+  const errors = []
+  const pubsub = new PGPubSub({
+    ...opts,
+    reconnectMaxRetries: 2,
+    host: 'xxx',
+    onConnectionError: (err) => {
+      errors.push(err)
+    }
+  })
+
+  const err = await t.throwsAsync(() => pubsub.connect())
+  t.is(err.message, '[PGPubSub]: max reconnect attempts reached, aborting')
+
+  t.is(errors.length, 0)
+  t.is(pubsub.state, 'closing')
+
+  t.teardown(() => {
+    pubsub.close()
+  })
+})
+
 test('connection can be re-established', async (t) => {
   const channel = getChannel()
   const pubsub = new PGPubSub({
@@ -240,6 +273,78 @@ test('connection cannot be re-established', async (t) => {
   })
 })
 
+test('onConnectionError callback receives error when retries exhausted', async (t) => {
+  const channel = getChannel()
+  const errors = []
+  const pubsub = new PGPubSub({
+    ...opts,
+    reconnectMaxRetries: 1,
+    onConnectionError: (err) => {
+      errors.push(err)
+    }
+  })
+
+  await pubsub.connect()
+  t.is(pubsub.state, 'connected')
+
+  await new Promise(resolve => {
+    pubsub.on(channel, (payload) => {
+      t.deepEqual(payload, 'this-is-the-payload')
+      resolve()
+    })
+    pubsub.emit(channel, 'this-is-the-payload')
+  })
+
+  // emulate pg client emitting an error after retries exceeded
+  pubsub.reconnectRetries = 2
+  pubsub.client.emit('error', new Error('connection reset'))
+
+  await sleep(10)
+
+  await waitUntilTrue(() => pubsub.state === 'closing')
+  t.is(errors.length, 1)
+  t.is(errors[0].message, '[PGPubSub]: max reconnect attempts reached, aborting')
+
+  t.teardown(() => {
+    pubsub.close()
+  })
+})
+
+test('onConnectionError callback receives error when reconnect exhausts retries', async (t) => {
+  const channel = getChannel()
+  const errors = []
+  const pubsub = new PGPubSub({
+    ...opts,
+    reconnectMaxRetries: 1,
+    onConnectionError: (err) => {
+      errors.push(err)
+    }
+  })
+
+  await pubsub.connect()
+  t.is(pubsub.state, 'connected')
+
+  await new Promise(resolve => {
+    pubsub.on(channel, (payload) => {
+      t.deepEqual(payload, 'this-is-the-payload')
+      resolve()
+    })
+    pubsub.emit(channel, 'this-is-the-payload')
+  })
+
+  // point host to invalid address so _reconnect() will fail and exhaust retries
+  pubsub.opts.host = 'xxx'
+  pubsub.client.emit('error', new Error('connection reset'))
+
+  await waitUntilTrue(() => errors.length === 1)
+  t.is(errors[0].message, '[PGPubSub]: max reconnect attempts reached, aborting')
+  t.is(pubsub.state, 'closing')
+
+  t.teardown(() => {
+    pubsub.close()
+  })
+})
+
 test('disabled reconnection with zero', async (t) => {
   const channel = getChannel()
   const pubsub = new PGPubSub({

types/index.d.ts

@@ -3,6 +3,7 @@ import { ClientConfig } from 'pg'
 export interface Options extends ClientConfig {
   reconnectMaxRetries?: number;
   maxPayloadSize?: number;
+  onConnectionError?: (error: Error) => void;
 }
 
 declare class PGPubSub {