fix(batch): bullmq worker requires maxretriesperrequest null redis

packages/bot/src/workers/batchJobWorker.spec.ts

@@ -3,7 +3,9 @@ import { describe, expect, it, jest } from '@jest/globals'
 type ProcessorFn = (job: { data: { jobId: string } }) => Promise<unknown>
 
 function makeModule(opts: {
-    redis?: object | null
+    sharedRedis?: object | null
+    bullmqRedisCtor?: jest.Mock | null
+    bullmqRedisFails?: boolean
     dbJob?: object | null
     executorExecute?: jest.Mock
     executorMissing?: boolean
@@ -12,7 +14,9 @@ function makeModule(opts: {
     checkpointThrows?: boolean
 }) {
     const {
-        redis = {},
+        sharedRedis = { setex: jest.fn().mockResolvedValue('OK') },
+        bullmqRedisCtor = undefined,
+        bullmqRedisFails = false,
         dbJob = null,
         executorExecute,
         executorMissing = false,
@@ -50,6 +54,17 @@ function makeModule(opts: {
     const mockWorkerOn = jest.fn()
     const mockWorkerClose = jest.fn().mockResolvedValue(undefined)
 
+    // BullMQ redis instance mock (with disconnect method)
+    const mockBullmqRedisDisconnect = jest.fn().mockResolvedValue(undefined)
+    const mockBullmqRedis = bullmqRedisFails
+        ? null
+        : {
+              disconnect: mockBullmqRedisDisconnect,
+          }
+
+    const MockRedis =
+        bullmqRedisCtor ?? jest.fn().mockReturnValue(mockBullmqRedis)
+
     let capturedProcessor: ProcessorFn | null = null
     const MockWorker = workerCtorThrows
         ? jest.fn(() => {
@@ -73,8 +88,9 @@ function makeModule(opts: {
     }
 
     jest.isolateModules(() => {
+        jest.doMock('ioredis', () => MockRedis)
         jest.doMock('@lucky/shared/services', () => ({
-            redisClient: { getClient: () => redis },
+            redisClient: { getClient: () => sharedRedis },
         }))
         jest.doMock('@lucky/shared/services/batch', () => ({
             batchJobService: {
@@ -91,6 +107,16 @@ function makeModule(opts: {
             errorLog,
             debugLog,
         }))
+        jest.doMock('@lucky/shared/config', () => ({
+            ENVIRONMENT_CONFIG: {
+                REDIS: {
+                    HOST: 'localhost',
+                    PORT: 6379,
+                    PASSWORD: undefined,
+                    DB: 0,
+                },
+            },
+        }))
         jest.doMock('./executorRegistry', () => ({
             getExecutor: mockGetExecutor,
             registerExecutor: mockRegisterExecutor,
@@ -110,6 +136,9 @@ function makeModule(opts: {
         mod: mod!,
         capturedProcessor: () => capturedProcessor,
         MockWorker,
+        MockRedis,
+        mockBullmqRedis,
+        mockBullmqRedisDisconnect,
         mockWorkerClose,
         mockWorkerOn,
         mockGetById,
@@ -132,20 +161,33 @@ function makeJob(jobId = 'job-abc') {
 describe('batchJobWorker', () => {
     describe('startBatchJobWorker', () => {
         it('returns early and logs error when redis is unavailable', async () => {
-            const { mod, MockWorker, errorLog } = makeModule({ redis: null })
+            const { mod, MockWorker, errorLog } = makeModule({
+                sharedRedis: null,
+            })
             await mod.startBatchJobWorker()
             expect(MockWorker).not.toHaveBeenCalled()
             expect(errorLog).toHaveBeenCalled()
         })
 
+        it('creates BullMQ redis connection with maxRetriesPerRequest: null', async () => {
+            const { mod, MockRedis } = makeModule({})
+            await mod.startBatchJobWorker()
+            expect(MockRedis).toHaveBeenCalledWith(
+                expect.objectContaining({
+                    maxRetriesPerRequest: null,
+                }),
+            )
+        })
+
         it('registers executor and creates Worker when redis is available', async () => {
-            const { mod, MockWorker, mockRegisterExecutor } = makeModule({})
+            const { mod, MockWorker, mockBullmqRedis, mockRegisterExecutor } =
+                makeModule({})
             await mod.startBatchJobWorker()
             expect(mockRegisterExecutor).toHaveBeenCalled()
             expect(MockWorker).toHaveBeenCalledWith(
                 'batch-jobs',
                 expect.any(Function),
-                { connection: {}, concurrency: 1 },
+                { connection: mockBullmqRedis, concurrency: 1 },
             )
         })
 
@@ -175,7 +217,7 @@ describe('batchJobWorker', () => {
 
     describe('stopBatchJobWorker', () => {
         it('does nothing when no worker is active', async () => {
-            const { mod, mockWorkerClose } = makeModule({ redis: null })
+            const { mod, mockWorkerClose } = makeModule({ sharedRedis: null })
             await mod.stopBatchJobWorker()
             expect(mockWorkerClose).not.toHaveBeenCalled()
         })
@@ -194,6 +236,13 @@ describe('batchJobWorker', () => {
             await mod.stopBatchJobWorker()
             expect(errorLog).toHaveBeenCalled()
         })
+
+        it('disconnects the BullMQ redis connection', async () => {
+            const { mod, mockBullmqRedisDisconnect } = makeModule({})
+            await mod.startBatchJobWorker()
+            await mod.stopBatchJobWorker()
+            expect(mockBullmqRedisDisconnect).toHaveBeenCalled()
+        })
     })
 
     describe('processBatchJob (via worker processor)', () => {

packages/bot/src/workers/batchJobWorker.ts

@@ -3,6 +3,7 @@
  * Loads jobs from the queue, resolves executors, and runs them with progress tracking.
  */
 
+import Redis from 'ioredis'
 import { Worker, type Job } from 'bullmq'
 
 type BatchJobData = { jobId: string }
@@ -11,9 +12,11 @@ import { batchJobService } from '@lucky/shared/services/batch'
 import type { BatchProgress, BatchJobType } from '@lucky/shared/services/batch'
 import { errorLog, infoLog, debugLog } from '@lucky/shared/utils'
 import { getExecutor, registerExecutor } from './executorRegistry'
+import { ENVIRONMENT_CONFIG } from '@lucky/shared/config'
 
 const QUEUE_NAME = 'batch-jobs'
 let worker: Worker<BatchJobData> | null = null
+let bullmqRedis: Redis | null = null
 
 /**
  * Progress callback that checkpoints to the database and publishes to Redis.
@@ -61,7 +64,9 @@ async function onProgress(
  * Processes a single batch job.
  * Loads the job from the database, resolves the executor, and runs it.
  */
-async function processBatchJob(job: Job<BatchJobData>): Promise<Record<string, unknown>> {
+async function processBatchJob(
+    job: Job<BatchJobData>,
+): Promise<Record<string, unknown>> {
     const jobId = job.data.jobId
 
     try {
@@ -141,21 +146,44 @@ async function processBatchJob(job: Job<BatchJobData>): Promise<Record<string, u
     }
 }
 
+/**
+ * Creates a dedicated BullMQ redis connection with maxRetriesPerRequest: null.
+ * BullMQ strictly requires this setting to avoid connection state conflicts.
+ */
+function createBullMQRedisConnection(): Redis {
+    // No try/catch: with lazyConnect the constructor never touches the
+    // network — connection errors surface later through BullMQ's own
+    // error handling, so a catch here would be a false safety net.
+    return new Redis({
+        host: ENVIRONMENT_CONFIG.REDIS.HOST,
+        port: ENVIRONMENT_CONFIG.REDIS.PORT,
+        password: ENVIRONMENT_CONFIG.REDIS.PASSWORD,
+        db: ENVIRONMENT_CONFIG.REDIS.DB,
+        // BullMQ requirement: must be null to avoid request queueing conflicts
+        maxRetriesPerRequest: null,
+        lazyConnect: true,
+    })
+}
+
 /**
  * Starts the batch job worker (concurrency 1).
  * Must be called with an active Redis connection.
  * Gracefully handles Redis unavailability without crashing.
  */
 export async function startBatchJobWorker(): Promise<void> {
-    const redis = redisClient.getClient()
-    if (!redis) {
+    // Check that shared redis is available (signals connectivity)
+    const sharedRedis = redisClient.getClient()
+    if (!sharedRedis) {
         errorLog({
             message:
                 'Cannot start batch job worker: Redis client not available',
         })
         return
     }
 
+    // Create a dedicated redis connection for BullMQ with maxRetriesPerRequest: null
+    bullmqRedis = createBullMQRedisConnection()
+
     // Register batch executors before consuming jobs. A lazy import keeps the
     // executor — which statically pulls in the bot client (bot/start) and DB
     // services — out of this module's static graph: it avoids an import cycle
@@ -167,12 +195,17 @@ export async function startBatchJobWorker(): Promise<void> {
         registerExecutor(new ChannelMoveBatchExecutor())
     } catch (error) {
         errorLog({ message: 'Failed to register batch executors', error })
+        // Clean up the bullmq redis connection on executor registration failure
+        if (bullmqRedis) {
+            await bullmqRedis.disconnect()
+            bullmqRedis = null
+        }
         return
     }
 
     try {
         worker = new Worker(QUEUE_NAME, processBatchJob, {
-            connection: redis,
+            connection: bullmqRedis,
             concurrency: 1,
         })
 
@@ -199,25 +232,45 @@ export async function startBatchJobWorker(): Promise<void> {
             message: 'Failed to start batch job worker',
             error,
         })
+        // Clean up the bullmq redis connection on worker creation failure
+        if (bullmqRedis) {
+            try {
+                await bullmqRedis.disconnect()
+            } catch (disconnectError) {
+                errorLog({
+                    message:
+                        'Error disconnecting BullMQ redis on startup failure',
+                    error: disconnectError,
+                })
+            }
+            bullmqRedis = null
+        }
     }
 }
 
 /**
  * Stops the batch job worker gracefully.
  */
 export async function stopBatchJobWorker(): Promise<void> {
-    if (worker) {
-        try {
+    try {
+        if (worker) {
             await worker.close()
             worker = null
-            infoLog({
-                message: 'Batch job worker stopped',
-            })
-        } catch (error) {
-            errorLog({
-                message: 'Error stopping batch job worker',
-                error,
-            })
+        }
+        infoLog({
+            message: 'Batch job worker stopped',
+        })
+    } catch (error) {
+        errorLog({
+            message: 'Error stopping batch job worker',
+            error,
+        })
+    } finally {
+        // Always release the BullMQ redis connection — a throwing
+        // worker.close() must not leak it
+        if (bullmqRedis) {
+            bullmqRedis.disconnect()
+            bullmqRedis = null
         }
     }
 }