fix(redis): separate BullMQ queue and worker connections

BullMQ Workers use blocking commands (BRPOPLPUSH) that require
a dedicated connection. Sharing connections caused command
starvation and timeouts.

- Separate queueConnection and workerConnection
- Fix undefined config.db in Redis logs
- Use structured logger instead of console.error

Fixes production Redis MISCONF errors.

Author: HMarzban

packages/hocuspocus.server/src/config/hocuspocus.config.ts

@@ -8,6 +8,7 @@ import { StoreDocumentQueue } from '../lib/queue'
 import { HealthCheck } from '../extensions/health.extension'
 import { RedisSubscriberExtension } from '../extensions/redis-subscriber.extension'
 import { prisma } from '../lib/prisma'
+import { dbLogger } from '../lib/logger'
 
 export { Database }
 
@@ -78,7 +79,7 @@ const configureExtensions = () => {
           })
           return doc ? doc.data : generateDefaultState()
         } catch (err) {
-          console.error('Error fetching data:', err)
+          dbLogger.error({ err }, 'Error fetching document data')
           await prisma.$disconnect()
           throw err
         }

packages/hocuspocus.server/src/lib/queue.ts

@@ -12,27 +12,38 @@ async function generateUniqueSlug(baseSlug: string): Promise<string> {
   return `${baseSlug}-${Date.now()}-${Math.random().toString(36).substr(2, 5)}`
 }
 
-// BullMQ requires ioredis - create dedicated connection for queues
-// Use centralized Redis config with BullMQ-specific settings
-const connection = createRedisConnection({
+// BullMQ connection options (shared base config)
+const bullmqConnectionOptions = {
   maxRetriesPerRequest: null, // BullMQ requirement - allows unlimited retries
-  enableReadyCheck: true, // Enable to ensure connection is ready before sending commands
-  enableOfflineQueue: true, // Enable to prevent command timeouts during reconnection
-  // Explicit timeout for BullMQ operations (longer than default for heavy operations)
+  enableReadyCheck: true,
+  enableOfflineQueue: true,
   commandTimeout: parseInt(process.env.REDIS_COMMAND_TIMEOUT || '60000', 10),
-  // Connection pooling settings
   connectTimeout: parseInt(process.env.REDIS_CONNECT_TIMEOUT || '30000', 10),
   keepAlive: parseInt(process.env.REDIS_KEEPALIVE || '30000', 10)
-})
+}
+
+// Queue connection (non-blocking operations)
+const queueConnection = createRedisConnection(bullmqConnectionOptions)
 
-if (!connection) {
-  queueLogger.error('Failed to create Redis connection for BullMQ')
+if (!queueConnection) {
+  queueLogger.error('Failed to create Redis connection for BullMQ Queue')
   throw new Error('Redis configuration required for queue operations')
 }
 
+// Worker connection (blocking operations - MUST be separate)
+// BullMQ uses BRPOPLPUSH which blocks the connection
+const createWorkerConnection = () => {
+  const conn = createRedisConnection(bullmqConnectionOptions)
+  if (!conn) {
+    queueLogger.error('Failed to create Redis connection for BullMQ Worker')
+    throw new Error('Redis configuration required for worker operations')
+  }
+  return conn
+}
+
 // Main queue for storing documents
 export const StoreDocumentQueue = new Queue<StoreDocumentData>('store-documents', {
-  connection,
+  connection: queueConnection,
   defaultJobOptions: {
     attempts: 5, // Increased retry attempts for better reliability
     backoff: {
@@ -49,7 +60,7 @@ export const StoreDocumentQueue = new Queue<StoreDocumentData>('store-documents'
 
 // Dead Letter Queue for permanently failed jobs
 export const DeadLetterQueue = new Queue<StoreDocumentData>('store-documents-dlq', {
-  connection,
+  connection: queueConnection,
   defaultJobOptions: {
     removeOnComplete: {
       count: 500,
@@ -67,6 +78,8 @@ StoreDocumentQueue.on('error', (err: Error) => {
 // Worker to process document storage jobs
 export const createDocumentWorker = () => {
   const redisPublisher = getRedisPublisher()
+  // Worker MUST have dedicated connection (uses blocking commands)
+  const workerConnection = createWorkerConnection()
 
   const worker = new Worker<StoreDocumentData>(
     'store-documents',
@@ -119,7 +132,10 @@ export const createDocumentWorker = () => {
         })
 
         const duration = Date.now() - startTime
-        queueLogger.info({ jobId: job.id, duration: `${duration}ms` }, 'Document stored successfully')
+        queueLogger.info(
+          { jobId: job.id, duration: `${duration}ms` },
+          'Document stored successfully'
+        )
 
         // Publish save confirmation to document-specific Redis channel
         if (redisPublisher) {
@@ -152,7 +168,7 @@ export const createDocumentWorker = () => {
       }
     },
     {
-      connection,
+      connection: workerConnection,
       concurrency: parseInt(process.env.BULLMQ_CONCURRENCY || '5', 10),
       limiter: {
         max: parseInt(process.env.BULLMQ_RATE_LIMIT_MAX || '300', 10),

packages/hocuspocus.server/src/lib/redis.ts

@@ -25,7 +25,6 @@ const getRedisConfig = () => {
   const host = process.env.REDIS_HOST
   const port = process.env.REDIS_PORT
 
-
   if (!host || !port) {
     return null
   }
@@ -34,7 +33,6 @@ const getRedisConfig = () => {
     host,
     port: parseInt(port, 10),
 
-
     // Connection settings
     lazyConnect: false, // Connect immediately on creation
     connectTimeout: parseInt(process.env.REDIS_CONNECT_TIMEOUT || '20000', 10),
@@ -95,13 +93,15 @@ export const getRedisClient = (): RedisClient | null => {
   }
 
   if (!redis) {
-    redisLogger.info({
-      host: config.host,
-      port: config.port,
-      db: config.db,
-      lazyConnect: config.lazyConnect,
-      connectTimeout: config.connectTimeout
-    }, 'Creating new Redis client...')
+    redisLogger.info(
+      {
+        host: config.host,
+        port: config.port,
+        lazyConnect: config.lazyConnect,
+        connectTimeout: config.connectTimeout
+      },
+      'Creating new Redis client...'
+    )
 
     redis = new Redis(config)
 
@@ -111,7 +111,7 @@ export const getRedisClient = (): RedisClient | null => {
     })
 
     redis.on('ready', () => {
-      redisLogger.info({ host: config.host, port: config.port, db: config.db }, 'Redis ready')
+      redisLogger.info({ host: config.host, port: config.port }, 'Redis ready')
     })
 
     redis.on('error', (err: Error) => {
@@ -140,7 +140,10 @@ export const getRedisClient = (): RedisClient | null => {
 }
 
 // Wait for Redis client to be ready (with auto-connect, it should be connecting already)
-export const waitForRedisReady = async (client: RedisClient, timeoutMs = 10000): Promise<boolean> => {
+export const waitForRedisReady = async (
+  client: RedisClient,
+  timeoutMs = 10000
+): Promise<boolean> => {
   if (client.status === 'ready') {
     return true
   }
@@ -238,7 +241,8 @@ export const createRedisConnection = (options: Partial<RedisOptions> = {}): Redi
 
   // BullMQ needs higher timeout for long-running operations
   // Use explicit timeout from options if provided, otherwise use env var or default
-  const commandTimeout = options.commandTimeout ?? parseInt(process.env.REDIS_COMMAND_TIMEOUT || '60000', 10)
+  const commandTimeout =
+    options.commandTimeout ?? parseInt(process.env.REDIS_COMMAND_TIMEOUT || '60000', 10)
 
   return new Redis({
     ...config,