fix(sync): handle delayed execution errors

Author: RomainLanz

src/drivers/sync_adapter.ts

@@ -41,7 +41,12 @@ export class SyncAdapter implements Adapter {
 
   pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
     setTimeout(() => {
-      void this.#execute(jobData, queue)
+      void this.#execute(jobData, queue).catch((error) => {
+        QueueManager.getLogger().error(
+          { err: error, jobId: jobData.id, jobName: jobData.name, queue },
+          'Failed to execute delayed sync job'
+        )
+      })
     }, delay)
 
     return Promise.resolve()

src/queue_manager.ts

@@ -252,6 +252,13 @@ class QueueManagerSingleton {
     return this.#jobFactory
   }
 
+  /**
+   * Get the configured logger used by the queue runtime.
+   */
+  getLogger(): Logger {
+    return this.#logger
+  }
+
   /**
    * Get the resolver responsible for effective queue/job runtime config.
    */

tests/queue_manager.spec.ts

@@ -81,6 +81,18 @@ test.group('QueueManager', () => {
     assert.exists(resolver)
   })
 
+  test('should expose the configured logger', async ({ assert }) => {
+    const logger = new MemoryLogger()
+
+    await QueueManager.init({
+      default: 'sync',
+      adapters: { sync: sync() },
+      logger,
+    })
+
+    assert.strictEqual(QueueManager.getLogger(), logger)
+  })
+
   test('should throw E_QUEUE_NOT_INITIALIZED when use() called before init()', async ({
     assert,
   }) => {

tests/sync_adapter.spec.ts

@@ -1,9 +1,11 @@
+import { setTimeout as sleep } from 'node:timers/promises'
 import { test } from '@japa/runner'
 import { Job } from '../src/job.js'
 import { Locator } from '../src/locator.js'
 import { QueueManager } from '../src/queue_manager.js'
 import { sync } from '../src/drivers/sync_adapter.js'
 import * as errors from '../src/exceptions.js'
+import { MemoryLogger } from './_mocks/memory_logger.js'
 
 test.group('SyncAdapter', (group) => {
   group.each.teardown(async () => {
@@ -55,4 +57,64 @@ test.group('SyncAdapter', (group) => {
     assert.instanceOf(failedError, errors.E_JOB_MAX_ATTEMPTS_REACHED)
     assert.deepEqual(contextJobIds, Array(contextJobIds.length).fill(jobId))
   })
+
+  test('should log delayed sync job failures without unhandled rejections', async ({
+    assert,
+  }) => {
+    const logger = new MemoryLogger()
+    let unhandledError: unknown
+    const onUnhandledRejection = (error: unknown) => {
+      unhandledError = error
+    }
+
+    class DelayedFailingSyncJob extends Job<Record<string, never>> {
+      async execute() {
+        throw new Error('boom')
+      }
+
+      async failed() {
+        throw new Error('failed hook exploded')
+      }
+    }
+
+    process.once('unhandledRejection', onUnhandledRejection)
+
+    try {
+      await QueueManager.init({
+        default: 'sync',
+        adapters: { sync: sync() },
+        logger,
+      })
+
+      Locator.register('DelayedFailingSyncJob', DelayedFailingSyncJob)
+
+      const adapter = QueueManager.use()
+
+      await adapter.pushLaterOn(
+        'default',
+        {
+          id: 'delayed-sync-job',
+          name: 'DelayedFailingSyncJob',
+          payload: {},
+          attempts: 0,
+          priority: 0,
+        },
+        0
+      )
+
+      await sleep(20)
+    } finally {
+      process.removeListener('unhandledRejection', onUnhandledRejection)
+    }
+
+    assert.isUndefined(unhandledError)
+    assert.lengthOf(logger.logs, 1)
+    assert.equal(logger.logs[0].level, 'error')
+    assert.equal(logger.logs[0].message, 'Failed to execute delayed sync job')
+    assert.equal(logger.logs[0].obj?.jobId, 'delayed-sync-job')
+    assert.equal(logger.logs[0].obj?.jobName, 'DelayedFailingSyncJob')
+    assert.equal(logger.logs[0].obj?.queue, 'default')
+    assert.instanceOf(logger.logs[0].obj?.err, Error)
+    assert.equal((logger.logs[0].obj?.err as Error).message, 'failed hook exploded')
+  })
 })