Merge pull request #764 from bcomnes/corelate-wip-to-worker

Add workId to localConcurrency workers and WIP data

Author: timgit

docs/api/events.md

@@ -63,6 +63,7 @@ Emitted at most once every 2 seconds when workers are receiving jobs. The payloa
 [
   {
     id: 'fc738fb0-1de5-4947-b138-40d6a790749e',
+    workId: 'a1b2c3d4-5678-90ab-cdef-1234567890ab',
     name: 'my-queue',
     options: { pollingInterval: 2000 },
     state: 'active',
@@ -78,6 +79,19 @@ Emitted at most once every 2 seconds when workers are receiving jobs. The payloa
 ]
 ```
 
+`workId` is the value returned by `work()`. When using `localConcurrency`, multiple worker entries in the array will share the same `workId`, allowing you to correlate them back to a specific `work()` call.
+
+```js
+const workId = await boss.work('my-queue', { localConcurrency: 5 }, handler)
+
+boss.on('wip', workers => {
+  const myWorkers = workers.filter(w => w.workId === workId)
+  const working = myWorkers.filter(w => w.count > 0).length
+  const idle = myWorkers.length - working
+  console.log(`working: ${working}/${myWorkers.length}, idle: ${idle}`)
+})
+```
+
 ## `stopped`
 
 Emitted after `stop()` once all workers have completed their work and maintenance has been shut down.

src/manager.ts

@@ -350,7 +350,7 @@ class Manager extends EventEmitter implements types.EventsMixin {
 
     const firstWorkerId = randomUUID({ disableEntropyCache: true })
 
-    const createWorker = (workerId: string) => {
+    const createWorker = (workerId: string, workId: string) => {
       const fetch = () => {
         const ignoreGroups = localGroupConcurrency != null
           ? this.#getGroupsAtLocalCapacity(name)
@@ -395,13 +395,13 @@ class Manager extends EventEmitter implements types.EventsMixin {
         this.emit(events.error, { ...error, message: error.message, stack: error.stack, queue: name, worker: workerId })
       }
 
-      return new Worker<ReqData>({ id: workerId, name, options, interval, fetch, onFetch, onError })
+      return new Worker<ReqData>({ id: workerId, workId, name, options, interval, fetch, onFetch, onError })
     }
 
     // Spawn workers based on localConcurrency setting
     for (let i = 0; i < localConcurrency; i++) {
       const workerId = i === 0 ? firstWorkerId : randomUUID({ disableEntropyCache: true })
-      const worker = createWorker(workerId)
+      const worker = createWorker(workerId, firstWorkerId)
 
       this.addWorker(worker)
       worker.start()

src/types.ts

@@ -440,6 +440,7 @@ export type WorkerState = 'created' | 'active' | 'stopping' | 'stopped'
 
 export interface WipData {
   id: string;
+  workId: string;
   name: string;
   options: WorkOptions;
   state: WorkerState;

src/worker.ts

@@ -10,6 +10,7 @@ const WORKER_STATES = {
 
 interface WorkerOptions<T> {
   id: string
+  workId: string
   name: string
   options: types.WorkOptions
   interval: number
@@ -20,6 +21,7 @@ interface WorkerOptions<T> {
 
 class Worker<T = unknown> {
   readonly id: string
+  readonly workId: string
   readonly name: string
   readonly options: types.WorkOptions
   readonly fetch: () => Promise<types.Job<T>[]>
@@ -43,8 +45,9 @@ class Worker<T = unknown> {
   private beenNotified = false
   private runPromise: Promise<void> | null = null
 
-  constructor ({ id, name, options, interval, fetch, onFetch, onError }: WorkerOptions<T>) {
+  constructor ({ id, workId, name, options, interval, fetch, onFetch, onError }: WorkerOptions<T>) {
     this.id = id
+    this.workId = workId
     this.name = name
     this.options = options
     this.fetch = fetch
@@ -133,6 +136,7 @@ class Worker<T = unknown> {
   toWipData (): types.WipData {
     return {
       id: this.id,
+      workId: this.workId,
       name: this.name,
       options: this.options,
       state: this.state,

test/workTest.ts

@@ -328,6 +328,29 @@ describe('work', function () {
     expect(wip2.length).toBe(1)
   })
 
+  it('should correlate wip entries to work() call via workId', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    const firstWipEvent = new Promise<Array<any>>(resolve => ctx.boss!.once('wip', resolve))
+
+    let handlerCompletedResolve: () => void
+    const handlerCompleted = new Promise<void>(resolve => { handlerCompletedResolve = resolve })
+
+    await ctx.boss.send(ctx.schema)
+
+    const workId = await ctx.boss.work(ctx.schema, { localConcurrency: 3, pollingIntervalSeconds: 1 }, async () => {
+      handlerCompletedResolve()
+      await delay(3000)
+    })
+
+    const wip = await firstWipEvent
+
+    expect(wip.every((w: any) => w.workId === workId)).toBe(true)
+    expect(wip.length).toBe(3)
+
+    await handlerCompleted
+  })
+
   it('getWipData() should return current worker state', async function () {
     ctx.boss = await helper.start(ctx.bossConfig)