fix(worker): cleanup timeout abort listener and extract abort race helper

Author: RomainLanz

src/worker.ts

@@ -449,13 +449,43 @@ export class Worker {
     const signal = AbortSignal.timeout(timeout)
     instance.$hydrate(payload, context, signal)
 
+    const { abortPromise, cleanupAbortListener } = this.#createTimeoutAbortRace(
+      signal,
+      instance.constructor.name,
+      timeout
+    )
+
+    try {
+      await Promise.race([instance.execute(), abortPromise])
+    } finally {
+      cleanupAbortListener()
+    }
+  }
+
+  #createTimeoutAbortRace(signal: AbortSignal, jobName: string, timeout: number) {
+    let abortHandler: (() => void) | undefined
+
     const abortPromise = new Promise<never>((_, reject) => {
-      signal.addEventListener('abort', () => {
-        reject(new errors.E_JOB_TIMEOUT([instance.constructor.name, timeout]))
-      })
+      abortHandler = () => {
+        reject(new errors.E_JOB_TIMEOUT([jobName, timeout]))
+      }
+
+      if (signal.aborted) {
+        abortHandler()
+        return
+      }
+
+      signal.addEventListener('abort', abortHandler, { once: true })
     })
 
-    await Promise.race([instance.execute(), abortPromise])
+    return {
+      abortPromise,
+      cleanupAbortListener: () => {
+        if (abortHandler) {
+          signal.removeEventListener('abort', abortHandler)
+        }
+      },
+    }
   }
 
   async #acquireNextJob(queues: string[]): Promise<{ job: AcquiredJob; queue: string } | null> {

tests/worker.spec.ts

@@ -461,6 +461,74 @@ test.group('Worker', () => {
     assert.isBelow(elapsed, 150, 'Job should be killed before completing')
   })
 
+  test('should remove timeout abort listener when job completes before timeout', async ({
+    assert,
+    cleanup,
+  }) => {
+    class FastJob extends Job {
+      static options = { timeout: 5_000 }
+
+      async execute() {}
+    }
+
+    const sharedAdapter = memory()()
+
+    const localConfig = {
+      default: 'memory',
+      adapters: { memory: () => sharedAdapter },
+    }
+
+    const controller = new AbortController()
+    const originalTimeout = AbortSignal.timeout
+    const originalAddEventListener = controller.signal.addEventListener.bind(controller.signal)
+    const originalRemoveEventListener = controller.signal.removeEventListener.bind(controller.signal)
+
+    let addedAbortListeners = 0
+    let removedAbortListeners = 0
+
+    controller.signal.addEventListener = ((type, listener, options) => {
+      if (type === 'abort') {
+        addedAbortListeners++
+      }
+
+      return originalAddEventListener(type, listener, options)
+    }) as AbortSignal['addEventListener']
+
+    controller.signal.removeEventListener = ((type, listener, options) => {
+      if (type === 'abort') {
+        removedAbortListeners++
+      }
+
+      return originalRemoveEventListener(type, listener, options)
+    }) as AbortSignal['removeEventListener']
+
+    AbortSignal.timeout = (() => controller.signal) as typeof AbortSignal.timeout
+
+    Locator.register('FastJob', FastJob)
+
+    const worker = new Worker(localConfig)
+
+    cleanup(async () => {
+      AbortSignal.timeout = originalTimeout
+      Locator.clear()
+      await worker.stop()
+    })
+
+    await sharedAdapter.push({
+      id: 'cleanup-timeout-listener-job',
+      name: 'FastJob',
+      payload: {},
+      attempts: 0,
+      priority: 0,
+    })
+
+    await worker.processCycle(['default']) // started
+    await worker.processCycle(['default']) // completed
+
+    assert.equal(addedAbortListeners, 1)
+    assert.equal(removedAbortListeners, 1)
+  })
+
   test('should retry timed out job when failOnTimeout is false', async ({ assert, cleanup }) => {
     let attempts = 0