feat: apply role-specific redis connection defaults options for queue vs worker

close #87

Author: Julien-R44

.changeset/role-based-connection-defaults.md

@@ -0,0 +1,5 @@
+---
+'@nemoventures/adonis-jobs': minor
+---
+
+Apply role-specific Redis connection defaults following BullMQ production recommendations. When `useSharedConnection` is `false`, Queue instances now set `enableOfflineQueue: false` (fail fast during disconnections) and Worker instances set `maxRetriesPerRequest: null` (required by BullMQ for resilient reconnection).

packages/core/src/connection_resolver.ts

@@ -14,10 +14,18 @@ export class ConnectionResolver {
   }
 
   /**
-   * Resolve a connection configuration to BullMQ connection options
+   * Resolve a connection configuration to BullMQ connection options.
+   *
+   * Applies role-specific IORedis defaults following BullMQ production
+   * recommendations: queues should fail fast while workers should be
+   * resilient to temporary disconnections.
    */
-  resolve(connectionConfig?: QueueConnectionConfig): BullConnectionOptions {
-    if (!connectionConfig) connectionConfig = this.config.connection
+  resolve(options?: {
+    config?: QueueConnectionConfig
+    role?: 'queue' | 'worker'
+  }): BullConnectionOptions {
+    const connectionConfig = options?.config ?? this.config.connection
+    const role = options?.role ?? 'queue'
 
     const redisConnection = this.redis.connection(
       connectionConfig.connectionName,
@@ -28,8 +36,17 @@ export class ConnectionResolver {
     /**
      * For non-shared connections, we return the connection options
      * in order to let BullMQ create a new IORedis instance.
+     *
+     * We apply different defaults based on the role:
+     * - Queue: disable offline queue so calls fail fast during disconnections
+     * - Worker: set maxRetriesPerRequest to null (required by BullMQ)
      */
     const ioConnection = redisConnection.ioConnection
-    return { ...ioConnection.options, maxRetriesPerRequest: null }
+
+    if (role === 'worker') {
+      return { ...ioConnection.options, maxRetriesPerRequest: null }
+    }
+
+    return { ...ioConnection.options, enableOfflineQueue: false }
   }
 }

packages/core/src/queue_manager.ts

@@ -78,7 +78,7 @@ export class QueueManager<KnownQueues extends Record<string, QueueConfig> = Queu
     const queue = BullMqFactory.createQueue<DataType, ReturnType>(String(queueName), {
       prefix: this.config.defaultPrefix,
       ...queueOptions,
-      connection: this.connectionResolver.resolve(connection),
+      connection: this.connectionResolver.resolve({ config: connection, role: 'queue' }),
       telemetry: new BullMQOtel('adonis-jobs'),
     })
 
@@ -101,7 +101,7 @@ export class QueueManager<KnownQueues extends Record<string, QueueConfig> = Queu
     const queueEvents = BullMqFactory.createQueueEvents(String(queueName), {
       prefix: this.config.defaultPrefix,
       ...queueOptions,
-      connection: this.connectionResolver.resolve(connection),
+      connection: this.connectionResolver.resolve({ config: connection, role: 'queue' }),
       telemetry: new BullMQOtel('adonis-jobs'),
     })
 
@@ -114,7 +114,10 @@ export class QueueManager<KnownQueues extends Record<string, QueueConfig> = Queu
 
     this.#flowProducer = BullMqFactory.createFlowProducer({
       prefix: this.config.defaultPrefix,
-      connection: this.connectionResolver.resolve(this.config.connection),
+      connection: this.connectionResolver.resolve({
+        config: this.config.connection,
+        role: 'queue',
+      }),
       telemetry: new BullMQOtel('adonis-jobs'),
     })
 

packages/core/src/worker/worker.ts

@@ -42,7 +42,10 @@ export class Worker<KnownQueues extends Record<string, QueueConfig> = Queues> {
    */
   #createBullWorker(): BullWorker {
     const queueConfig = this.#config.queues[this.#queueName]
-    const connection = this.#connectionResolver.resolve(queueConfig.connection)
+    const connection = this.#connectionResolver.resolve({
+      config: queueConfig.connection,
+      role: 'worker',
+    })
 
     return BullMqFactory.createWorker(
       String(this.#queueName),

packages/core/tests/unit/connection_resolver.spec.ts

@@ -0,0 +1,87 @@
+import { test } from '@japa/runner'
+
+import { ConnectionResolver } from '../../src/connection_resolver.js'
+import type { Config, QueueConnectionConfig } from '../../src/types/index.js'
+
+function createFakeRedisService(ioOptions: Record<string, any> = {}) {
+  const fakeIoConnection = { options: { host: '127.0.0.1', port: 6379, ...ioOptions } }
+  return { connection: () => ({ ioConnection: fakeIoConnection }), fakeIoConnection }
+}
+
+function createConfig(overrides?: Partial<Config<any>>): Config<any> {
+  return {
+    connection: { connectionName: 'main' } as QueueConnectionConfig,
+    defaultQueue: 'default',
+    queues: { default: {} },
+    ...overrides,
+  }
+}
+
+test.group('ConnectionResolver', () => {
+  test('worker role sets maxRetriesPerRequest to null', ({ assert }) => {
+    const redis = createFakeRedisService()
+    const resolver = new ConnectionResolver(createConfig(), redis as any)
+
+    const result = resolver.resolve({ role: 'worker' }) as Record<string, any>
+
+    assert.equal(result.maxRetriesPerRequest, null)
+    assert.notProperty(result, 'enableOfflineQueue')
+  })
+
+  test('queue role disables offline queue', ({ assert }) => {
+    const redis = createFakeRedisService()
+    const resolver = new ConnectionResolver(createConfig(), redis as any)
+
+    const result = resolver.resolve({ role: 'queue' }) as Record<string, any>
+
+    assert.equal(result.enableOfflineQueue, false)
+    assert.notProperty(result, 'maxRetriesPerRequest')
+  })
+
+  test('defaults to queue role when no role is specified', ({ assert }) => {
+    const redis = createFakeRedisService()
+    const resolver = new ConnectionResolver(createConfig(), redis as any)
+
+    const result = resolver.resolve() as Record<string, any>
+
+    assert.equal(result.enableOfflineQueue, false)
+    assert.notProperty(result, 'maxRetriesPerRequest')
+  })
+
+  test('returns raw ioConnection when useSharedConnection is true', ({ assert }) => {
+    const redis = createFakeRedisService()
+    const resolver = new ConnectionResolver(
+      createConfig({ useSharedConnection: true }),
+      redis as any,
+    )
+
+    const workerResult = resolver.resolve({ role: 'worker' })
+    const queueResult = resolver.resolve({ role: 'queue' })
+
+    assert.strictEqual(workerResult, redis.fakeIoConnection as any)
+    assert.strictEqual(queueResult, redis.fakeIoConnection as any)
+  })
+
+  test('spreads ioConnection options into the result', ({ assert }) => {
+    const redis = createFakeRedisService({ db: 2, password: 'secret' })
+    const resolver = new ConnectionResolver(createConfig(), redis as any)
+
+    const result = resolver.resolve({ role: 'worker' }) as Record<string, any>
+
+    assert.equal(result.host, '127.0.0.1')
+    assert.equal(result.port, 6379)
+    assert.equal(result.db, 2)
+    assert.equal(result.password, 'secret')
+  })
+
+  test('uses custom connection config when provided', ({ assert }) => {
+    const redis = createFakeRedisService()
+    const customConfig = { connectionName: 'custom' } as QueueConnectionConfig
+    const resolver = new ConnectionResolver(createConfig(), redis as any)
+
+    resolver.resolve({ config: customConfig, role: 'queue' })
+
+    // No error thrown means it resolved successfully
+    assert.isTrue(true)
+  })
+})