fix: ensure abort listeners are removed from queue jobs (#2482)

Fixes a bug where aborted queue jobs were not removing abort signal
listeners properly.

Also removes the aborted job from the queue immediately to reduce
size and allow garbage collection sooner.

packages/utils/src/queue/index.ts

@@ -243,6 +243,16 @@ export class Queue<JobReturnType = unknown, JobOptions extends QueueAddOptions =
         return result
       })
       .catch(err => {
+        if (job.status === 'queued') {
+          // job was aborted before it started - remove the job from the queue
+          for (let i = 0; i < this.queue.length; i++) {
+            if (this.queue[i] === job) {
+              this.queue.splice(i, 1)
+              break
+            }
+          }
+        }
+
         this.safeDispatchEvent('error', { detail: err })
         this.safeDispatchEvent('failure', { detail: { job, error: err } })
 

packages/utils/src/queue/job.ts

@@ -56,6 +56,7 @@ export class Job <JobOptions extends AbortOptions = AbortOptions, JobReturnType
     // if all recipients have aborted the job, actually abort the job
     if (allAborted) {
       this.controller.abort(new AbortError())
+      this.cleanup()
     }
   }
 
@@ -99,6 +100,7 @@ export class Job <JobOptions extends AbortOptions = AbortOptions, JobReturnType
 
   cleanup (): void {
     this.recipients.forEach(recipient => {
+      recipient.cleanup()
       recipient.signal?.removeEventListener('abort', this.onAbort)
     })
   }

packages/utils/src/queue/recipient.ts

@@ -17,7 +17,7 @@ export class JobRecipient<JobReturnType> {
   }
 
   onAbort (): void {
-    this.deferred.reject(new AbortError())
+    this.deferred.reject(this.signal?.reason ?? new AbortError())
   }
 
   cleanup (): void {

packages/utils/test/fixtures/test-signal.ts

@@ -0,0 +1,30 @@
+import { TypedEventEmitter } from '@libp2p/interface'
+
+export interface TestSignalEvents {
+  abort: CustomEvent
+}
+
+export class TestSignal extends TypedEventEmitter<TestSignalEvents> {
+  public aborted: boolean
+  public reason: any
+
+  constructor () {
+    super()
+
+    this.aborted = false
+  }
+
+  throwIfAborted (): void {
+
+  }
+
+  onabort (): void {
+
+  }
+
+  abort (reason?: any): void {
+    this.aborted = true
+    this.reason = reason
+    this.safeDispatchEvent('abort')
+  }
+}

packages/utils/test/queue.spec.ts

@@ -2,7 +2,8 @@ import { expect } from 'aegir/chai'
 import delay from 'delay'
 import all from 'it-all'
 import pDefer from 'p-defer'
-import { Queue } from '../src/queue/index.js'
+import { Queue, type QueueAddOptions } from '../src/queue/index.js'
+import { TestSignal } from './fixtures/test-signal.js'
 
 const fixture = Symbol('fixture')
 
@@ -12,6 +13,10 @@ function randomInt (minimum: number, maximum: number): number {
   )
 }
 
+interface SlowJobQueueOptions extends QueueAddOptions {
+  slow: boolean
+}
+
 describe('queue', () => {
   it('adds', async () => {
     const queue = new Queue<symbol>({})
@@ -716,4 +721,97 @@ describe('queue', () => {
     expect(collected).to.deep.equal([0, 1])
     expect(queue.size).to.equal(0)
   })
+
+  it('cleans up listeners after all job recipients abort', async () => {
+    const queue = new Queue<void, SlowJobQueueOptions>({ concurrency: 1 })
+    void queue.add(async () => {
+      await delay(100)
+    }, {
+      slow: true
+    })
+
+    const signal = new TestSignal()
+
+    const jobResult = queue.add(async () => {}, {
+      slow: false,
+      signal
+    })
+
+    expect(queue.size).to.equal(2)
+    expect(queue.queued).to.equal(1)
+    expect(queue.running).to.equal(1)
+
+    const slowJob = queue.queue.find(job => !job.options.slow)
+
+    expect(slowJob?.recipients).to.have.lengthOf(1)
+    expect(slowJob?.recipients[0].signal).to.equal(signal)
+
+    // listeners added
+    expect(signal.listenerCount('abort')).to.equal(2)
+
+    // abort job stuck in queue
+    signal.abort()
+
+    // all listeners removed
+    expect(signal.listenerCount('abort')).to.equal(0)
+
+    await expect(jobResult).to.eventually.be.rejected
+      .with.property('code', 'ABORT_ERR')
+  })
+
+  it('rejects aborted jobs with the abort reason if supplied', async () => {
+    const queue = new Queue<void, SlowJobQueueOptions>({ concurrency: 1 })
+    void queue.add(async () => {
+      await delay(100)
+    }, {
+      slow: true
+    })
+
+    const signal = new TestSignal()
+
+    const jobResult = queue.add(async () => {}, {
+      slow: false,
+      signal
+    })
+
+    const err = new Error('Took too long')
+
+    // abort job stuck in queue
+    signal.abort(err)
+
+    // result rejects
+    await expect(jobResult).to.eventually.be.rejectedWith(err)
+  })
+
+  it('immediately removes aborted job', async () => {
+    const signal = new TestSignal()
+    const queue = new Queue<void, SlowJobQueueOptions>({ concurrency: 1 })
+    void queue.add(async () => {
+      await delay(100)
+    }, {
+      slow: true
+    })
+    const jobResult = queue.add(async () => {}, {
+      slow: false,
+      signal
+    })
+
+    expect(queue.size).to.equal(2)
+    expect(queue.queued).to.equal(1)
+    expect(queue.running).to.equal(1)
+
+    // abort job stuck in queue
+    signal.abort()
+
+    await expect(jobResult).to.eventually.be.rejected
+      .with.property('code', 'ABORT_ERR')
+
+    // counts updated
+    expect(queue.size).to.equal(1)
+    expect(queue.queued).to.equal(0)
+    expect(queue.running).to.equal(1)
+
+    // job not in queue any more
+    expect(queue.queue.find(job => !job.options.slow)).to.be.undefined()
+  })
 })