From 4ff1052f7f77050c3dd40f523af4401a252f67d8 Mon Sep 17 00:00:00 2001 From: "@mrubens" <2600+mrubens@users.noreply.github.com> Date: Sat, 15 Aug 2026 03:55:55 +0000 Subject: [PATCH 1/3] fix: bound shared Redis connectivity failures --- packages/redis/src/__tests__/index.test.ts | 84 +++++++++++++++++++++- packages/redis/src/index.ts | 57 ++++++++++++++- 2 files changed, 137 insertions(+), 4 deletions(-) diff --git a/packages/redis/src/__tests__/index.test.ts b/packages/redis/src/__tests__/index.test.ts index be77b1d94..a350ebbb6 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,82 @@ 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('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..96ad25f56 100644 --- a/packages/redis/src/index.ts +++ b/packages/redis/src/index.ts @@ -55,6 +55,58 @@ export async function syncAutoStartChannelCacheBestEffort(params: { let redis: 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 // @roomote/env may already have snapshotted an earlier value. @@ -70,9 +122,12 @@ function resolveRedisUrl(): string { export const getRedis = () => { if (!redis) { redis = new Redis(resolveRedisUrl(), { - maxRetriesPerRequest: null, + maxRetriesPerRequest: REDIS_MAX_RETRIES_PER_REQUEST, connectTimeout: 5000, + retryStrategy: (attempt) => + Math.min(attempt * 50, REDIS_MAX_RECONNECT_DELAY_MS), }); + observeRedisConnectivity(redis); } return redis; From 2a809a9fb645aa872e3ee1655d522d2cb4904e02 Mon Sep 17 00:00:00 2001 From: "@mrubens" <2600+mrubens@users.noreply.github.com> Date: Sat, 15 Aug 2026 04:25:47 +0000 Subject: [PATCH 2/3] fix: preserve BullMQ blocking Redis retries --- packages/redis/src/__tests__/index.test.ts | 14 +++++++++ packages/redis/src/index.ts | 29 ++++++++++++++----- .../lib/docker-environment-validation.ts | 4 +-- .../task-runs/__tests__/enqueue-sleep.test.ts | 15 ++++++++-- .../src/server/lib/task-runs/enqueue-sleep.ts | 4 +-- 5 files changed, 53 insertions(+), 13 deletions(-) diff --git a/packages/redis/src/__tests__/index.test.ts b/packages/redis/src/__tests__/index.test.ts index a350ebbb6..d539b0cf6 100644 --- a/packages/redis/src/__tests__/index.test.ts +++ b/packages/redis/src/__tests__/index.test.ts @@ -66,6 +66,20 @@ describe('getRedis', () => { 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')); diff --git a/packages/redis/src/index.ts b/packages/redis/src/index.ts index 96ad25f56..30f17bcf1 100644 --- a/packages/redis/src/index.ts +++ b/packages/redis/src/index.ts @@ -54,6 +54,7 @@ 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; @@ -119,20 +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: REDIS_MAX_RETRIES_PER_REQUEST, - connectTimeout: 5000, - retryStrategy: (attempt) => - Math.min(attempt * 50, REDIS_MAX_RECONNECT_DELAY_MS), - }); - observeRedisConnectivity(redis); + 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..9e7e4b982 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,8 +7,8 @@ const { queueEventsConstructorMock, waitUntilFinishedMock, mockGetRedis, + mockGetBullMqRedis, } = vi.hoisted(() => { - // eslint-disable-next-line @typescript-eslint/no-explicit-any type AnyMock = Mock<(...args: any[]) => any>; const queueAddMock = vi.fn() as AnyMock; const queueGetJobMock = 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(), }); } From 1e97c822a5441e134c090b642edd6df8089d01fc Mon Sep 17 00:00:00 2001 From: "@mrubens" <2600+mrubens@users.noreply.github.com> Date: Sat, 15 Aug 2026 04:26:34 +0000 Subject: [PATCH 3/3] test: remove broad BullMQ mock types --- .../src/server/lib/task-runs/__tests__/enqueue-sleep.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 9e7e4b982..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 @@ -9,7 +9,7 @@ const { mockGetRedis, mockGetBullMqRedis, } = vi.hoisted(() => { - 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;