Fix Redis connection for Railway: improve timeout, retry logic, and connection handling

Author: StephaneWamba

services/knowledge-base/src/config/redis.ts

@@ -10,57 +10,84 @@ const logger = createLogger('knowledge-base-service:redis')
 
 let redis: Redis | null = null
 
-export function getRedis(): Redis | null {
+export function getRedis(): Redis {
   if (!redis) {
     const redisUrl = process.env.REDIS_URL
     if (!redisUrl) {
-      logger.warn('REDIS_URL not configured, Redis features will be unavailable. Using in-memory queue fallback.')
-      return null
+      throw new Error('REDIS_URL environment variable is required')
     }
 
     const isTLS = redisUrl.startsWith('rediss://') || process.env.REDIS_TLS === 'true'
 
-    // Parse URL to extract host and port for family option
+    // Parse URL to extract connection details
     const urlObj = new URL(redisUrl)
     const host = urlObj.hostname
     const port = parseInt(urlObj.port || '6379', 10)
+    const password = urlObj.password || (urlObj.searchParams.get('password') || undefined)
+    
+    // For Railway internal URLs, ensure proper connection
+    const isRailwayInternal = host.includes('railway.internal')
 
     redis = new Redis({
       host,
       port,
+      password,
       family: 4, // Force IPv4 to avoid IPv6 issues
       maxRetriesPerRequest: null, // Required by BullMQ for blocking operations
       tls: isTLS ? {
         rejectUnauthorized: false,
         checkServerIdentity: () => undefined,
       } : undefined,
-      connectTimeout: 10000,
-      enableReadyCheck: false, // Disable ready check - we'll verify with PING
-      enableOfflineQueue: true,
-      lazyConnect: true, // Connect explicitly after setup
+      connectTimeout: 30000, // Increased timeout for Railway internal network
+      enableReadyCheck: true, // Enable ready check for better connection validation
+      enableOfflineQueue: false, // Disable offline queue to fail fast if not connected
+      lazyConnect: false, // Connect immediately to catch errors early
       retryStrategy: (times) => {
-        const delay = Math.min(times * 50, 2000)
-        if (times > 10) {
+        const delay = Math.min(times * 100, 3000)
+        if (times > 30) { // More retries for Railway network
+          logger.warn('Redis connection retry limit reached', { attempts: times })
           return null
         }
         return delay
       },
       reconnectOnError: (err) => {
-        const targetError = 'READONLY'
-        if (err.message.includes(targetError)) {
+        // Reconnect on network errors, but not on auth errors
+        const targetErrors = ['ECONNREFUSED', 'ETIMEDOUT', 'ENOTFOUND', 'ECONNRESET']
+        if (targetErrors.some(error => err.message.includes(error))) {
           return true
         }
+        // Don't reconnect on auth errors
+        if (err.message.includes('NOAUTH') || err.message.includes('invalid password')) {
+          logger.error('Redis authentication failed', { error: err.message })
+          return false
+        }
         return false
       },
     })
 
     redis.on('error', (error) => {
-      logger.error('Redis connection error', { error: error.message })
+      logger.error('Redis connection error', { 
+        error: error.message,
+        host,
+        port,
+        isRailwayInternal 
+      })
+    })
+
+    redis.on('connect', () => {
+      logger.info('Redis connecting', { host, port, isRailwayInternal })
+    })
+
+    redis.on('ready', () => {
+      logger.info('Redis connection ready', { host, port, isRailwayInternal })
+    })
+
+    redis.on('close', () => {
+      logger.warn('Redis connection closed', { host, port })
     })
 
-    // Connect explicitly after event listeners are set up
-    redis.connect().catch((err) => {
-      logger.error('Failed to connect to Redis', { error: err.message })
+    redis.on('reconnecting', (delay) => {
+      logger.info('Redis reconnecting', { host, port, delay })
     })
   }
 

services/knowledge-base/src/services/queue.ts

@@ -13,34 +13,55 @@ const logger = createLogger('knowledge-base-service:queue')
 // Queue name
 const QUEUE_NAME = 'document-processing'
 
-/**
- * Get Redis connection (may be null if Redis is not configured)
- */
-function getRedisConnection(): ReturnType<typeof getRedis> {
-  try {
-    return getRedis()
-  } catch (error) {
-    logger.warn('Redis connection failed, will use in-memory queue', {
-      error: error instanceof Error ? error.message : String(error)
-    })
-    return null
-  }
-}
+// Get Redis connection
+const redisConnection = getRedis()
 
 /**
  * Wait for Redis to be ready by testing with PING
  */
-async function waitForRedisReady(redisConnection: NonNullable<ReturnType<typeof getRedis>>): Promise<void> {
-  const maxAttempts = 20
+async function waitForRedisReady(): Promise<void> {
+  const maxAttempts = 60 // Increased for Railway network
   const delayMs = 500
 
+  // First, wait for connection status to be 'ready' or 'connecting'
+  let statusCheckAttempts = 0
+  while (redisConnection.status !== 'ready' && redisConnection.status !== 'connecting' && statusCheckAttempts < 10) {
+    await new Promise(resolve => setTimeout(resolve, 100))
+    statusCheckAttempts++
+  }
+  
+  // Then test with PING
   for (let attempt = 1; attempt <= maxAttempts; attempt++) {
     try {
-      await redisConnection.ping()
-      return
+      const result = await Promise.race([
+        redisConnection.ping(),
+        new Promise<never>((_, reject) => 
+          setTimeout(() => reject(new Error('PING timeout')), 5000)
+        )
+      ])
+      if (result === 'PONG') {
+        logger.info('Redis connection verified', { 
+          status: redisConnection.status,
+          attempt 
+        })
+        return
+      }
     } catch (error) {
+      const errorMessage = error instanceof Error ? error.message : String(error)
       if (attempt === maxAttempts) {
-        throw new Error(`Redis connection timeout after ${maxAttempts} attempts. Status: ${redisConnection.status}`)
+        logger.error('Redis connection timeout', {
+          attempts: maxAttempts,
+          status: redisConnection.status,
+          error: errorMessage
+        })
+        throw new Error(`Redis connection timeout after ${maxAttempts} attempts. Status: ${redisConnection.status}. Error: ${errorMessage}`)
+      }
+      if (attempt % 10 === 0) {
+        logger.warn('Redis connection still waiting', { 
+          attempt, 
+          maxAttempts,
+          status: redisConnection.status 
+        })
       }
       await new Promise(resolve => setTimeout(resolve, delayMs))
     }
@@ -58,22 +79,7 @@ async function initializeQueue(): Promise<void> {
   }
 
   initializationPromise = (async () => {
-    const redisConnection = getRedisConnection()
-    
-    // If Redis is not available, skip BullMQ initialization
-    if (!redisConnection) {
-      logger.info('Redis not available, using in-memory queue only')
-      return
-    }
-    
-    try {
-      await waitForRedisReady(redisConnection)
-    } catch (error) {
-      logger.warn('Failed to connect to Redis, will use in-memory queue', {
-        error: error instanceof Error ? error.message : String(error)
-      })
-      return
-    }
+    await waitForRedisReady()
 
     if (!documentQueue) {
       documentQueue = new Queue(QUEUE_NAME, {
@@ -115,7 +121,7 @@ async function initializeQueue(): Promise<void> {
           }
         },
         {
-          connection: redisConnection!,
+          connection: redisConnection,
           concurrency: 2,
         }
       )
@@ -187,11 +193,6 @@ export async function enqueueDocument(documentId: string): Promise<Job | { id: s
       )
     ])
 
-    // Check if queue was successfully initialized (Redis available)
-    if (!documentQueue) {
-      throw new Error('Queue not initialized - Redis unavailable')
-    }
-    
     const queue = getDocumentQueue()
 
     const job = await Promise.race([
@@ -211,7 +212,7 @@ export async function enqueueDocument(documentId: string): Promise<Job | { id: s
     return job
   } catch (error) {
     logger.warn(`BullMQ queue unavailable, using in-memory queue for document ${documentId}`, {
-      error: error instanceof Error ? error.message : String(error)
+      error
     })
 
     if (inMemoryQueue.some(item => item.documentId === documentId)) {
@@ -237,20 +238,6 @@ export async function enqueueDocument(documentId: string): Promise<Job | { id: s
 export async function getQueueStats() {
   try {
     await initializeQueue()
-    
-    // If queue is not initialized (Redis unavailable), return in-memory stats only
-    if (!documentQueue) {
-      return {
-        waiting: 0,
-        active: 0,
-        completed: 0,
-        failed: 0,
-        delayed: 0,
-        inMemoryQueue: inMemoryQueue.length,
-        mode: 'in-memory',
-      }
-    }
-    
     const queue = getDocumentQueue()
 
     const [waiting, active, completed, failed, delayed] = await Promise.all([
@@ -268,7 +255,6 @@ export async function getQueueStats() {
       failed,
       delayed,
       inMemoryQueue: inMemoryQueue.length,
-      mode: 'bullmq',
     }
   } catch (error) {
     return {
@@ -278,7 +264,6 @@ export async function getQueueStats() {
       failed: 0,
       delayed: 0,
       inMemoryQueue: inMemoryQueue.length,
-      mode: 'in-memory',
     }
   }
 }