fix: propagte job progress events and support synchronous events

Fixes two bugs with the Queue class:

1. When joining an existing job, progress events were not received by the second context
2. If a progress event was emitted before any async work was done in a queue job under the concurrency threshold, it was missed because the job result recipient had not been registered yet

Author: achingbrain

packages/utils/package.json

@@ -63,6 +63,7 @@
     "netmask": "^2.0.2",
     "p-defer": "^4.0.1",
     "p-event": "^7.0.0",
+    "progress-events": "^1.1.0",
     "race-signal": "^2.0.0",
     "uint8-varint": "^2.0.4",
     "uint8arraylist": "^2.4.8",

packages/utils/src/queue/index.ts

@@ -6,6 +6,7 @@ import { debounce } from '../debounce.js'
 import { QueueFullError } from '../errors.js'
 import { Job } from './job.js'
 import type { AbortOptions, Metrics } from '@libp2p/interface'
+import type { ProgressOptions } from 'progress-events'
 
 export type { Job, JobTimeline } from './job.js'
 export type { JobRecipient } from './recipient.js'
@@ -118,7 +119,7 @@ export interface QueueEvents<JobReturnType, JobOptions extends AbortOptions = Ab
  * 1. Items remain at the head of the queue while they are running so `queue.size` includes `queue.pending` items - this is so interested parties can join the results of a queue item while it is running
  * 2. The options for a job are stored separately to the job in order for them to be modified while they are still in the queue
  */
-export class Queue<JobReturnType = unknown, JobOptions extends AbortOptions = AbortOptions> extends TypedEventEmitter<QueueEvents<JobReturnType, JobOptions>> {
+export class Queue<JobReturnType = unknown, JobOptions extends AbortOptions & ProgressOptions = AbortOptions> extends TypedEventEmitter<QueueEvents<JobReturnType, JobOptions>> {
   public concurrency: number
   public maxSize: number
   public queue: Array<Job<JobOptions, JobReturnType>>
@@ -257,9 +258,8 @@ export class Queue<JobReturnType = unknown, JobOptions extends AbortOptions = Ab
     const job = new Job<JobOptions, JobReturnType>(fn, options)
     this.enqueue(job)
     this.safeDispatchEvent('add')
-    this.tryToStartAnother()
 
-    return job.join(options)
+    const result = job.join(options)
       .then(result => {
         this.safeDispatchEvent('completed', { detail: result })
         this.safeDispatchEvent('success', { detail: { job, result } })
@@ -281,6 +281,10 @@ export class Queue<JobReturnType = unknown, JobOptions extends AbortOptions = Ab
 
         throw err
       })
+
+    this.tryToStartAnother()
+
+    return result
   }
 
   /**

packages/utils/src/queue/job.ts

@@ -4,6 +4,7 @@ import { raceSignal } from 'race-signal'
 import { JobRecipient } from './recipient.js'
 import type { JobStatus } from './index.js'
 import type { AbortOptions } from '@libp2p/interface'
+import type { ProgressOptions } from 'progress-events'
 
 /**
  * Returns a random string
@@ -18,7 +19,7 @@ export interface JobTimeline {
   finished?: number
 }
 
-export class Job <JobOptions extends AbortOptions = AbortOptions, JobReturnType = unknown> {
+export class Job <JobOptions extends AbortOptions & ProgressOptions = AbortOptions, JobReturnType = unknown> {
   public id: string
   public fn: (options: JobOptions) => Promise<JobReturnType>
   public options: JobOptions
@@ -59,11 +60,11 @@ export class Job <JobOptions extends AbortOptions = AbortOptions, JobReturnType
     }
   }
 
-  async join (options: AbortOptions = {}): Promise<JobReturnType> {
-    const recipient = new JobRecipient<JobReturnType>(options.signal)
+  async join (options?: Partial<Pick<JobOptions, 'signal' | 'onProgress'>>): Promise<JobReturnType> {
+    const recipient = new JobRecipient<JobReturnType>(options)
     this.recipients.push(recipient)
 
-    options.signal?.addEventListener('abort', this.onAbort)
+    options?.signal?.addEventListener('abort', this.onAbort)
 
     return recipient.deferred.promise
   }
@@ -77,7 +78,12 @@ export class Job <JobOptions extends AbortOptions = AbortOptions, JobReturnType
 
       const result = await raceSignal(this.fn({
         ...(this.options ?? {}),
-        signal: this.controller.signal
+        signal: this.controller.signal,
+        onProgress: (evt: any): void => {
+          this.recipients.forEach(recipient => {
+            recipient.onProgress?.(evt)
+          })
+        }
       }), this.controller.signal)
 
       this.recipients.forEach(recipient => {

packages/utils/src/queue/recipient.ts

@@ -1,13 +1,17 @@
 import { AbortError } from '@libp2p/interface'
 import pDefer from 'p-defer'
+import type { AbortOptions } from '@libp2p/interface'
 import type { DeferredPromise } from 'p-defer'
+import type { ProgressOptions, ProgressEventListener } from 'progress-events'
 
-export class JobRecipient<JobReturnType> {
+export class JobRecipient<JobReturnType, JobOptions extends AbortOptions & ProgressOptions = any> {
   public deferred: DeferredPromise<JobReturnType>
   public signal?: AbortSignal
+  public onProgress?: ProgressEventListener
 
-  constructor (signal?: AbortSignal) {
-    this.signal = signal
+  constructor (options?: Partial<Pick<JobOptions, 'signal' | 'onProgress'>>) {
+    this.signal = options?.signal
+    this.onProgress = options?.onProgress
     this.deferred = pDefer()
 
     this.onAbort = this.onAbort.bind(this)

packages/utils/test/queue.spec.ts

@@ -2,9 +2,11 @@ import { expect } from 'aegir/chai'
 import delay from 'delay'
 import all from 'it-all'
 import pDefer from 'p-defer'
+import { CustomProgressEvent } from 'progress-events'
 import { Queue } from '../src/queue/index.js'
 import { TestSignal } from './fixtures/test-signal.js'
 import type { AbortOptions } from '@libp2p/interface'
+import type { ProgressOptions } from 'progress-events'
 
 const fixture = Symbol('fixture')
 
@@ -864,4 +866,72 @@ describe('queue', () => {
     await expect(p).to.eventually.equal('hello')
     await expect(queue.add(job)).to.eventually.equal('hello')
   })
+
+  it('should forward progress events to multiple consumers', async () => {
+    interface ProgressJobOptions extends AbortOptions, ProgressOptions {
+
+    }
+    const queue = new Queue<string, ProgressJobOptions>({
+      concurrency: 1,
+      maxSize: 1
+    })
+    queue.pause()
+
+    const job = async (options: ProgressJobOptions): Promise<string> => {
+      options.onProgress?.(new CustomProgressEvent('before'))
+      await delay(100)
+      options.onProgress?.(new CustomProgressEvent('after'))
+      return 'hello'
+    }
+
+    const events: any[] = []
+
+    const p1 = queue.add(job, {
+      onProgress: (evt) => {
+        events.push(evt)
+      }
+    })
+
+    const p2 = queue.queue[0].join({
+      onProgress: (evt) => {
+        events.push(evt)
+      }
+    })
+
+    queue.resume()
+
+    await Promise.all([
+      p1,
+      p2
+    ])
+
+    expect(events).to.have.lengthOf(4)
+  })
+
+  it('should consume synchronous progress events', async () => {
+    interface ProgressJobOptions extends AbortOptions, ProgressOptions {
+
+    }
+    const queue = new Queue<string, ProgressJobOptions>({
+      concurrency: 1,
+      maxSize: 1
+    })
+
+    const events: any[] = []
+
+    const job = async (options: ProgressJobOptions): Promise<string> => {
+      options.onProgress?.(new CustomProgressEvent('before'))
+      await delay(100)
+      options.onProgress?.(new CustomProgressEvent('after'))
+      return 'hello'
+    }
+
+    await queue.add(job, {
+      onProgress: (evt) => {
+        events.push(evt)
+      }
+    })
+
+    expect(events).to.have.lengthOf(2)
+  })
 })