diff --git a/packages/redis/src/__tests__/index.test.ts b/packages/redis/src/__tests__/index.test.ts index be77b1d94..d539b0cf6 100644 --- a/packages/redis/src/__tests__/index.test.ts +++ b/packages/redis/src/__tests__/index.test.ts @@ -1,8 +1,9 @@ -const { envMock, redisConstructorMock } = vi.hoisted(() => ({ +const { envMock, redisConstructorMock, redisOnMock } = vi.hoisted(() => ({ envMock: { REDIS_URL: 'redis://from-env-object:6379', }, redisConstructorMock: vi.fn(), + redisOnMock: vi.fn(), })); vi.mock('@roomote/env', async (importOriginal) => { @@ -19,6 +20,11 @@ vi.mock('ioredis', () => ({ constructor(...args: unknown[]) { redisConstructorMock(...args); } + + on(...args: unknown[]) { + redisOnMock(...args); + return this; + } }, })); @@ -26,6 +32,7 @@ describe('getRedis', () => { beforeEach(() => { vi.resetModules(); vi.clearAllMocks(); + vi.useRealTimers(); delete process.env.REDIS_URL; envMock.REDIS_URL = 'redis://from-env-object:6379'; }); @@ -39,11 +46,96 @@ describe('getRedis', () => { expect(redisConstructorMock).toHaveBeenCalledWith( 'redis://from-process-env:6379', - { - maxRetriesPerRequest: null, + expect.objectContaining({ + maxRetriesPerRequest: 3, connectTimeout: 5000, + retryStrategy: expect.any(Function), + }), + ); + }); + + it('caps reconnect delays while allowing the client to recover', async () => { + const { getRedis } = await import('../index'); + + getRedis(); + + const options = redisConstructorMock.mock.calls[0]?.[1] as { + retryStrategy: (attempt: number) => number; + }; + expect(options.retryStrategy(1)).toBe(50); + expect(options.retryStrategy(100)).toBe(2_000); + }); + + it('provides a separate BullMQ-compatible blocking client', async () => { + const { getBullMqRedis, getRedis } = await import('../index'); + + const sharedClient = getRedis(); + const bullMqClient = getBullMqRedis(); + + expect(bullMqClient).not.toBe(sharedClient); + expect(redisConstructorMock).toHaveBeenNthCalledWith( + 2, + 'redis://from-env-object:6379', + expect.objectContaining({ maxRetriesPerRequest: null }), + ); + }); + + it('rate-limits connection errors and summarizes recovery', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-08-15T03:41:59.000Z')); + const consoleErrorSpy = vi + .spyOn(console, 'error') + .mockImplementation(() => undefined); + const consoleInfoSpy = vi + .spyOn(console, 'info') + .mockImplementation(() => undefined); + const { getRedis } = await import('../index'); + + getRedis(); + + const handlers = Object.fromEntries(redisOnMock.mock.calls) as Record< + string, + (...args: unknown[]) => void + >; + const dnsError = Object.assign( + new Error('getaddrinfo ENOTFOUND redis.internal'), + { code: 'ENOTFOUND' }, + ); + + handlers.error?.(dnsError); + handlers.error?.(dnsError); + expect(consoleErrorSpy).toHaveBeenCalledTimes(1); + expect(consoleErrorSpy).toHaveBeenLastCalledWith( + '[redis] connection degraded; dependent operations may fail', + { + code: 'ENOTFOUND', + message: 'getaddrinfo ENOTFOUND redis.internal', + suppressedErrors: 0, }, ); + + vi.advanceTimersByTime(30_000); + handlers.error?.(dnsError); + expect(consoleErrorSpy).toHaveBeenCalledTimes(2); + expect(consoleErrorSpy).toHaveBeenLastCalledWith( + '[redis] connection degraded; dependent operations may fail', + { + code: 'ENOTFOUND', + message: 'getaddrinfo ENOTFOUND redis.internal', + suppressedErrors: 1, + }, + ); + + vi.advanceTimersByTime(5_000); + handlers.ready?.(); + expect(consoleInfoSpy).toHaveBeenCalledWith('[redis] connection restored', { + outageDurationMs: 35_000, + totalErrors: 3, + suppressedErrors: 0, + }); + + consoleErrorSpy.mockRestore(); + consoleInfoSpy.mockRestore(); }); it('throws when REDIS_URL is unavailable', async () => { diff --git a/packages/redis/src/index.ts b/packages/redis/src/index.ts index b0a0433a9..30f17bcf1 100644 --- a/packages/redis/src/index.ts +++ b/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, diff --git a/packages/sdk/src/server/lib/docker-environment-validation.ts b/packages/sdk/src/server/lib/docker-environment-validation.ts index 9568f803b..a2993506b 100644 --- a/packages/sdk/src/server/lib/docker-environment-validation.ts +++ b/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(), }); } diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/enqueue-sleep.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/enqueue-sleep.test.ts index 508e70941..4a428a826 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/enqueue-sleep.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/enqueue-sleep.test.ts @@ -7,9 +7,9 @@ const { queueEventsConstructorMock, waitUntilFinishedMock, mockGetRedis, + mockGetBullMqRedis, } = vi.hoisted(() => { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - type AnyMock = Mock<(...args: any[]) => any>; + type AnyMock = Mock<(...args: unknown[]) => unknown>; const queueAddMock = vi.fn() as AnyMock; const queueGetJobMock = vi.fn() as AnyMock; const waitUntilFinishedMock = vi.fn() as AnyMock; @@ -26,6 +26,7 @@ const { }) as AnyMock, waitUntilFinishedMock, mockGetRedis: vi.fn(() => ({ status: 'ready' })) as AnyMock, + mockGetBullMqRedis: vi.fn(() => ({ status: 'blocking-ready' })) as AnyMock, }; }); @@ -33,7 +34,10 @@ vi.mock('bullmq', () => ({ Queue: queueConstructorMock, QueueEvents: queueEventsConstructorMock, })); -vi.mock('@roomote/redis', () => ({ getRedis: mockGetRedis })); +vi.mock('@roomote/redis', () => ({ + getRedis: mockGetRedis, + getBullMqRedis: mockGetBullMqRedis, +})); describe('enqueueTaskSleep', () => { beforeEach(() => { @@ -61,6 +65,13 @@ describe('enqueueTaskSleep', () => { { kind: 'queue-events' }, 60_000, ); + expect(queueConstructorMock).toHaveBeenCalledWith( + 'task-sleep-jobs', + expect.objectContaining({ connection: { status: 'ready' } }), + ); + expect(queueEventsConstructorMock).toHaveBeenCalledWith('task-sleep-jobs', { + connection: { status: 'blocking-ready' }, + }); }); it('deduplicates a sleep action that is still waiting', async () => { diff --git a/packages/sdk/src/server/lib/task-runs/enqueue-sleep.ts b/packages/sdk/src/server/lib/task-runs/enqueue-sleep.ts index 48f238ad8..83c788603 100644 --- a/packages/sdk/src/server/lib/task-runs/enqueue-sleep.ts +++ b/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 { function getTaskSleepQueueEvents(): QueueEvents { if (!taskSleepQueueEvents) { taskSleepQueueEvents = new QueueEvents(TASK_SLEEP_QUEUE_NAME, { - connection: getRedis(), + connection: getBullMqRedis(), }); }