Merge pull request #635 from timgit/offwork-wait

stop() async resource cleanup

Author: timgit

docs/api/workers.md

@@ -83,11 +83,20 @@ await boss.work(queue, async ([ job ]) => {
 Notifies a worker by id to bypass the job polling interval (see `pollingIntervalSeconds`) for this iteration in the loop.
 
 
-### `offWork(value)`
+### `offWork(name, options)`
 
 Removes a worker by name or id and stops polling.
 
 ** Arguments **
-- value: string or object
+- name: string
+- options: object
 
-  If a string, removes all workers found matching the name.  If an object, only the worker with a matching `id` will be removed.
+**Options**
+
+* **wait**, boolean, *(default=true)*
+
+  If the promise should wait until current jobs finish
+
+* **id**, string
+
+  Only stop polling by worker id

eslint.config.ts eslint.config.jseslint.config.ts renamed to eslint.config.js

@@ -1,6 +1,6 @@
 {
   "name": "pg-boss",
-  "version": "12.1.1",
+  "version": "12.2.0",
   "description": "Queueing jobs in Postgres from Node.js like a boss",
   "type": "module",
   "main": "./dist/index.mjs",
@@ -20,17 +20,17 @@
   },
   "devDependencies": {
     "@istanbuljs/nyc-config-typescript": "^1.0.2",
-    "@tsconfig/node22": "^22.0.2",
+    "@tsconfig/node22": "^22.0.3",
     "@types/mocha": "^10.0.10",
-    "@types/node": "^22.19.0",
+    "@types/node": "^22.19.1",
     "@types/pg": "^8.15.6",
     "eslint": "^9.39.1",
     "luxon": "^3.7.2",
     "mocha": "^11.7.5",
     "neostandard": "^0.12.2",
     "nyc": "^17.1.0",
     "source-map-support": "^0.5.21",
-    "tsdown": "^0.16.0",
+    "tsdown": "^0.16.4",
     "tsx": "^4.20.6",
     "typescript": "^5.9.3"
   },

package-lock.json

@@ -79,7 +79,7 @@ function checkWorkArgs (name: string, args: any[]): {
 } {
   let options, callback
 
-  assert(name, 'missing job name')
+  assert(name, 'queue name is required')
 
   if (args.length === 1) {
     callback = args[0]

package.json

@@ -14,6 +14,7 @@ class Db extends EventEmitter implements types.IDatabase, types.EventsMixin {
     super()
 
     config.application_name = config.application_name || 'pgboss'
+    config.connectionTimeoutMillis = config.connectionTimeoutMillis || 10000
     // config.maxUses = config.maxUses || 1000
 
     this.config = config

src/attorney.ts

@@ -137,6 +137,9 @@ export class PgBoss extends EventEmitter<types.PgBossEventMap> {
 
       if (this.#db._pgbdb && this.#db.opened && close) {
         await this.#db.close()
+
+        // Give event loop time to process socket closes
+        await delay(10)
       }
 
       this.#stopped = true
@@ -150,9 +153,7 @@ export class PgBoss extends EventEmitter<types.PgBossEventMap> {
       return await shutdown()
     }
 
-    const isWip = () => this.#manager.getWipData({ includeInternal: false }).length > 0
-
-    while ((Date.now() - this.#stoppingOn!) < timeout && isWip()) {
+    while ((Date.now() - this.#stoppingOn!) < timeout && this.#manager.hasPendingCleanups()) {
       await delay(500)
     }
 
@@ -197,14 +198,12 @@ export class PgBoss extends EventEmitter<types.PgBossEventMap> {
     return this.#manager.work(...args as Parameters<Manager['work']>)
   }
 
-  offWork (name: string): Promise<void>
-  offWork (options: types.OffWorkOptions): Promise<void>
-  offWork (value: string | types.OffWorkOptions): Promise<void> {
-    return this.#manager.offWork(value)
+  offWork (name: string, options?: types.OffWorkOptions): Promise<void> {
+    return this.#manager.offWork(name, options)
   }
 
   notifyWorker (workerId: string): void {
-    this.#manager.notifyWorker(workerId)
+    return this.#manager.notifyWorker(workerId)
   }
 
   subscribe (event: string, name: string): Promise<void> {

src/db.ts

@@ -28,6 +28,7 @@ class Manager extends EventEmitter implements types.EventsMixin {
   queueCacheInterval: NodeJS.Timeout | undefined
   timekeeper: Timekeeper | undefined
   queues: Record<string, types.QueueResult> | null
+  pendingOffWorkCleanups: Set<Promise<void>>
 
   constructor (db: types.IDatabase, config: types.ResolvedConstructorOptions) {
     super()
@@ -37,6 +38,7 @@ class Manager extends EventEmitter implements types.EventsMixin {
     this.wipTs = Date.now()
     this.workers = new Map()
     this.queues = null
+    this.pendingOffWorkCleanups = new Set()
   }
 
   async start () {
@@ -80,11 +82,11 @@ class Manager extends EventEmitter implements types.EventsMixin {
 
     clearInterval(this.queueCacheInterval)
 
-    for (const worker of this.workers.values()) {
-      if (!INTERNAL_QUEUES[worker.name]) {
-        await this.offWork(worker.name)
-      }
-    }
+    await Promise.allSettled(
+      [...this.workers.values()]
+        .filter(worker => !INTERNAL_QUEUES[worker.name])
+        .map(async worker => await this.offWork(worker.name, { wait: false }))
+    )
   }
 
   async failWip () {
@@ -132,11 +134,15 @@ class Manager extends EventEmitter implements types.EventsMixin {
 
     const data = this.getWorkers()
       .map(i => i.toWipData())
-      .filter(i => i.count > 0 && (!INTERNAL_QUEUES[i.name] || includeInternal))
+      .filter(i => i.state !== 'stopped' && (!INTERNAL_QUEUES[i.name] || includeInternal))
 
     return data
   }
 
+  hasPendingCleanups (): boolean {
+    return this.pendingOffWorkCleanups.size > 0
+  }
+
   private async watch<T> (name: string, options: types.ResolvedWorkOptions, callback: types.WorkHandler<T>): Promise<string> {
     if (this.stopped) {
       throw new Error('Workers are disabled. pg-boss is stopped')
@@ -190,18 +196,13 @@ class Manager extends EventEmitter implements types.EventsMixin {
     return id
   }
 
-  async offWork (value: string | types.OffWorkOptions): Promise<void> {
-    assert(value, 'Missing required argument')
+  async offWork (name: string, options: types.OffWorkOptions = { wait: true }): Promise<void> {
+    assert(name, 'queue name is required')
+    assert(typeof name === 'string', 'queue name must be a string')
 
-    const query = (typeof value === 'string')
-      ? { filter: (i: Worker<any>) => i.name === value }
-      : (typeof value === 'object' && value.id)
-          ? { filter: (i: Worker<any>) => i.id === value.id }
-          : null
+    const query = (i: Worker<any>) => options?.id ? i.id === options.id : i.name === name
 
-    assert(query, 'Invalid argument. Expected string or object: { id }')
-
-    const workers = this.getWorkers().filter(i => query.filter(i) && !i.stopping && !i.stopped)
+    const workers = this.getWorkers().filter(i => query(i) && !i.stopping && !i.stopped)
 
     if (workers.length === 0) {
       return
@@ -211,15 +212,22 @@ class Manager extends EventEmitter implements types.EventsMixin {
       worker.stop()
     }
 
-    setImmediate(async () => {
+    const cleanupPromise = (async () => {
       while (!workers.every(w => w.stopped)) {
         await delay(1000)
       }
 
       for (const worker of workers) {
         this.removeWorker(worker)
       }
-    })
+    })()
+
+    if (options.wait) {
+      await cleanupPromise
+    } else {
+      this.pendingOffWorkCleanups.add(cleanupPromise)
+      cleanupPromise.finally(() => this.pendingOffWorkCleanups.delete(cleanupPromise))
+    }
   }
 
   notifyWorker (workerId: string): void {

src/index.ts

@@ -69,7 +69,7 @@ class Timekeeper extends EventEmitter implements types.EventsMixin {
 
     this.stopped = true
 
-    await this.manager.offWork(QUEUES.SEND_IT)
+    await this.manager.offWork(QUEUES.SEND_IT, { wait: true })
 
     if (this.skewMonitorInterval) {
       clearInterval(this.skewMonitorInterval)

src/manager.ts

@@ -30,7 +30,7 @@ export interface DatabaseOptions {
   connectionString?: string;
   max?: number;
   db?: IDatabase;
-
+  connectionTimeoutMillis?: number;
   /** @internal */
   debug?: boolean;
 }
@@ -246,11 +246,11 @@ export interface StopOptions {
   close?: boolean;
   graceful?: boolean;
   timeout?: number;
-  wait?: boolean;
 }
 
 export interface OffWorkOptions {
-  id: string
+  id?: string,
+  wait?: boolean
 }
 
 export interface EventsMixin extends NodeJS.EventEmitter {