[Fix] Services stall during transient Redis outages

packages/redis/src/index.ts

@@ -54,6 +54,59 @@ export async function syncAutoStartChannelCacheBestEffort(params: {
 }
 
 let redis: Redis | null = null;
+let bullMqRedis: Redis | null = null;
+
+const REDIS_MAX_RETRIES_PER_REQUEST = 3;
+const REDIS_MAX_RECONNECT_DELAY_MS = 2_000;
+const REDIS_ERROR_LOG_INTERVAL_MS = 30_000;
+
+function observeRedisConnectivity(client: Redis): void {
+  let outageStartedAt: number | null = null;
+  let lastErrorLoggedAt = 0;
+  let errorsSinceLastLog = 0;
+  let totalErrors = 0;
+
+  client.on('error', (error: Error & { code?: string }) => {
+    const now = Date.now();
+    outageStartedAt ??= now;
+    errorsSinceLastLog += 1;
+    totalErrors += 1;
+
+    if (
+      totalErrors > 1 &&
+      now - lastErrorLoggedAt < REDIS_ERROR_LOG_INTERVAL_MS
+    ) {
+      return;
+    }
+
+    console.error(
+      '[redis] connection degraded; dependent operations may fail',
+      {
+        code: error.code,
+        message: error.message,
+        suppressedErrors: Math.max(0, errorsSinceLastLog - 1),
+      },
+    );
+    lastErrorLoggedAt = now;
+    errorsSinceLastLog = 0;
+  });
+
+  client.on('ready', () => {
+    if (outageStartedAt === null) {
+      return;
+    }
+
+    console.info('[redis] connection restored', {
+      outageDurationMs: Date.now() - outageStartedAt,
+      totalErrors,
+      suppressedErrors: errorsSinceLastLog,
+    });
+    outageStartedAt = null;
+    lastErrorLoggedAt = 0;
+    errorsSinceLastLog = 0;
+    totalErrors = 0;
+  });
+}
 
 function resolveRedisUrl(): string {
   // In apps/web on Vercel, dotenvx decrypts into process.env at runtime after
@@ -67,17 +120,34 @@ function resolveRedisUrl(): string {
   return redisUrl;
 }
 
+function createRedis(maxRetriesPerRequest: number | null): Redis {
+  const client = new Redis(resolveRedisUrl(), {
+    maxRetriesPerRequest,
+    connectTimeout: 5000,
+    retryStrategy: (attempt) =>
+      Math.min(attempt * 50, REDIS_MAX_RECONNECT_DELAY_MS),
+  });
+  observeRedisConnectivity(client);
+  return client;
+}
+
 export const getRedis = () => {
   if (!redis) {
-    redis = new Redis(resolveRedisUrl(), {
-      maxRetriesPerRequest: null,
-      connectTimeout: 5000,
-    });
+    redis = createRedis(REDIS_MAX_RETRIES_PER_REQUEST);
   }
 
   return redis;
 };
 
+export const getBullMqRedis = () => {
+  if (!bullMqRedis) {
+    // BullMQ blocking connections reject clients with a finite request retry limit.
+    bullMqRedis = createRedis(null);
+  }
+
+  return bullMqRedis;
+};
+
 export { acquireRedisLock, withRedisLock, withContention } from './lock';
 export type {
   RedisLockOptions,

packages/sdk/src/server/lib/docker-environment-validation.ts

@@ -1,6 +1,6 @@
 import { Queue, QueueEvents } from 'bullmq';
 
-import { getRedis } from '@roomote/redis';
+import { getBullMqRedis, getRedis } from '@roomote/redis';
 import { DOCKER_VALIDATION_QUEUE_NAME } from '@roomote/types';
 import type { DockerEnvironmentValidationResult } from '@roomote/compute-providers';
 
@@ -39,7 +39,7 @@ function getValidationQueue() {
 function getValidationQueueEvents(): QueueEvents {
   if (!validationQueueEvents) {
     validationQueueEvents = new QueueEvents(DOCKER_VALIDATION_QUEUE_NAME, {
-      connection: getRedis(),
+      connection: getBullMqRedis(),
     });
   }
 

packages/sdk/src/server/lib/task-runs/enqueue-sleep.ts

@@ -1,7 +1,7 @@
 import { Queue, QueueEvents } from 'bullmq';
 import { z } from 'zod';
 
-import { getRedis } from '@roomote/redis';
+import { getBullMqRedis, getRedis } from '@roomote/redis';
 
 export const TASK_SLEEP_QUEUE_NAME = 'task-sleep-jobs';
 const TASK_SLEEP_RESULT_TIMEOUT_MS = 60_000;
@@ -42,7 +42,7 @@ function getTaskSleepQueue(): Queue<TaskSleepRequest> {
 function getTaskSleepQueueEvents(): QueueEvents {
   if (!taskSleepQueueEvents) {
     taskSleepQueueEvents = new QueueEvents(TASK_SLEEP_QUEUE_NAME, {
-      connection: getRedis(),
+      connection: getBullMqRedis(),
     });
   }