added LISTEN tuning config and docs, versioning, deps

Author: timgit

docs/api/constructor.md

@@ -40,6 +40,18 @@ The following options can be set as properties in an object for additional confi
 
   Number of milliseconds to wait before timing out when acquiring a new client from the pool. Set to `0` to disable the timeout and wait indefinitely.
 
+* **notifyHeartbeatIntervalMs** - int, defaults to 10000
+
+  Interval between heartbeat checks on the dedicated LISTEN/NOTIFY connection. Lower values detect silent connection drops faster at the cost of more heartbeat queries.
+
+* **notifyHeartbeatTimeoutMs** - int, defaults to 5000
+
+  Timeout for each LISTEN/NOTIFY heartbeat query. If a heartbeat does not complete within this window the listener is torn down and reconnected. Raise this on a loaded database where the default is too aggressive.
+
+* **notifyKeepAliveInitialDelayMs** - int, defaults to 10000
+
+  TCP keepalive initial delay for the dedicated LISTEN/NOTIFY connection.
+
 * **db** - object
 
     Passing an object named db allows you "bring your own database connection". This option may be beneficial if you'd like to use an existing database service with its own connection pool. Setting this option will bypass the above configuration.

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "pg-boss",
-  "version": "12.26.1",
+  "version": "12.26.2",
   "description": "Queueing jobs in Postgres from Node.js like a boss",
   "type": "module",
   "main": "./dist/index.js",
@@ -18,8 +18,8 @@
   },
   "devDependencies": {
     "@electric-sql/pglite": "^0.5.4",
-    "@prisma/adapter-pg": "^7.8.0",
-    "@prisma/client": "^7.8.0",
+    "@prisma/adapter-pg": "^7.9.0",
+    "@prisma/client": "^7.9.0",
     "@tsconfig/node-ts": "^23.6.4",
     "@tsconfig/node22": "^22.0.5",
     "@types/luxon": "^3.7.2",
@@ -31,10 +31,10 @@
     "drizzle-orm": "^1.0.0-rc.4",
     "eslint": "^9.39.5",
     "knex": "^3.3.0",
-    "kysely": "^0.29.3",
+    "kysely": "^0.29.4",
     "luxon": "^3.7.2",
     "neostandard": "^0.13.0",
-    "prisma": "^7.8.0",
+    "prisma": "^7.9.0",
     "tsx": "^4.23.1",
     "typescript": "^6.0.3",
     "vitest": "^4.0.18"

package.json

@@ -6,9 +6,9 @@ import type * as types from './types.ts'
 // Keep silent network failures below the default 30-second notify polling backstop: in the
 // worst case a failure happens immediately after a successful check, then takes one interval,
 // one query timeout, and the existing first reconnect backoff (1s) to restore LISTEN.
-const LISTEN_HEARTBEAT_INTERVAL_MS = 10000
-const LISTEN_HEARTBEAT_TIMEOUT_MS = 5000
-const LISTEN_KEEP_ALIVE_INITIAL_DELAY_MS = 10000
+const DEFAULT_LISTEN_HEARTBEAT_INTERVAL_MS = 10000
+const DEFAULT_LISTEN_HEARTBEAT_TIMEOUT_MS = 5000
+const DEFAULT_LISTEN_KEEP_ALIVE_INITIAL_DELAY_MS = 10000
 
 class Db extends EventEmitter implements types.IDatabase, types.EventsMixin {
   private pool!: pg.Pool
@@ -79,8 +79,9 @@ class Db extends EventEmitter implements types.IDatabase, types.EventsMixin {
     let reconnectTimer: ReturnType<typeof setTimeout> | null = null
     let heartbeatTimer: ReturnType<typeof setTimeout> | null = null
     let attempt = 0
-    const heartbeatInterval = this.config.__test__listenHeartbeatIntervalMs ?? LISTEN_HEARTBEAT_INTERVAL_MS
-    const heartbeatTimeout = this.config.__test__listenHeartbeatTimeoutMs ?? LISTEN_HEARTBEAT_TIMEOUT_MS
+    const heartbeatInterval = this.config.notifyHeartbeatIntervalMs ?? DEFAULT_LISTEN_HEARTBEAT_INTERVAL_MS
+    const heartbeatTimeout = this.config.notifyHeartbeatTimeoutMs ?? DEFAULT_LISTEN_HEARTBEAT_TIMEOUT_MS
+    const keepAliveInitialDelay = this.config.notifyKeepAliveInitialDelayMs ?? DEFAULT_LISTEN_KEEP_ALIVE_INITIAL_DELAY_MS
     // Only self-heal once the listener has been established at least once. If the INITIAL connect
     // fails, the rejection propagates to the caller (Notifier.start), which falls back to
     // polling-only and discards this subscription's close handle — so a reconnect scheduled from
@@ -161,7 +162,7 @@ class Db extends EventEmitter implements types.IDatabase, types.EventsMixin {
       const next = new pg.Client({
         ...this.config,
         keepAlive: true,
-        keepAliveInitialDelayMillis: LISTEN_KEEP_ALIVE_INITIAL_DELAY_MS
+        keepAliveInitialDelayMillis: keepAliveInitialDelay
       })
 
       next.on('error', error => {

src/db.ts

@@ -45,12 +45,25 @@ export interface DatabaseOptions {
   max?: number;
   db?: IDatabase;
   connectionTimeoutMillis?: number;
+  /**
+   * Interval in milliseconds between LISTEN/NOTIFY heartbeat checks on the dedicated
+   * listener connection. Lower values detect silent connection drops faster at the cost
+   * of more heartbeat queries. Defaults to 10000.
+   */
+  notifyHeartbeatIntervalMs?: number;
+  /**
+   * Timeout in milliseconds for each LISTEN/NOTIFY heartbeat query. If a heartbeat does
+   * not complete within this window the listener is torn down and reconnected. Raise this
+   * on a loaded database where the default is too aggressive. Defaults to 5000.
+   */
+  notifyHeartbeatTimeoutMs?: number;
+  /**
+   * TCP keepalive initial delay in milliseconds for the dedicated LISTEN/NOTIFY connection.
+   * Defaults to 10000.
+   */
+  notifyKeepAliveInitialDelayMs?: number;
   /** @internal */
   debug?: boolean;
-  /** @internal */
-  __test__listenHeartbeatIntervalMs?: number;
-  /** @internal */
-  __test__listenHeartbeatTimeoutMs?: number;
 }
 
 export interface SchedulingOptions {

src/types.ts

@@ -32,8 +32,8 @@ helper.describePglite('db listen/notify', function () {
       application_name: 'pgboss_half_open_test',
       host: '127.0.0.1',
       port: proxyPort,
-      __test__listenHeartbeatIntervalMs: 100,
-      __test__listenHeartbeatTimeoutMs: 100
+      notifyHeartbeatIntervalMs: 100,
+      notifyHeartbeatTimeoutMs: 100
     })
     const channel = 'pgboss_db_half_open_test'
     const payloads: string[] = []
@@ -77,8 +77,8 @@ helper.describePglite('db listen/notify', function () {
     const db = new Db({
       ...helper.getConfig(),
       application_name: 'pgboss_lost_subscription_test',
-      __test__listenHeartbeatIntervalMs: 100,
-      __test__listenHeartbeatTimeoutMs: 100
+      notifyHeartbeatIntervalMs: 100,
+      notifyHeartbeatTimeoutMs: 100
     })
     const channel = 'pgboss_db_lost_subscription_test'
     const payloads: string[] = []

test/dbListenTest.ts

@@ -221,6 +221,7 @@ describe('work lifecycle', function () {
     const localConcurrency = 3
     const abortedJobs: string[] = []
     const jobIds: (string | null)[] = []
+    let started = 0
 
     // Send 3 jobs
     for (let i = 0; i < 3; i++) {
@@ -229,6 +230,7 @@ describe('work lifecycle', function () {
     }
 
     await ctx.boss.work(ctx.schema, { localConcurrency, pollingIntervalSeconds: 0.5 }, async ([job]) => {
+      started++
       // All jobs check for abort signal
       for (let i = 0; i < 100; i++) {
         if (job.signal.aborted) {
@@ -239,8 +241,15 @@ describe('work lifecycle', function () {
       }
     })
 
-    // Wait for all workers to start
-    await delay(500)
+    // Wait until all 3 workers have actually picked up a job before stopping. A fixed
+    // delay races the last fetch under load: a worker that hasn't fetched when stop()
+    // halts fetching never runs its handler, so its signal is never checked (abortedJobs
+    // ends at 2 instead of 3). Gate on the real in-flight count instead.
+    for (let i = 0; i < 50; i++) {
+      if (started >= localConcurrency) break
+      await delay(100)
+    }
+    expect(started).toBe(localConcurrency)
 
     // Stop with short timeout - jobs take 10s, so timeout will expire
     await ctx.boss.stop({ timeout: 1000 })