diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 9e62c1c1..2c6e64f9 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -7,9 +7,12 @@ import { configureOpenAPI } from '@api/utils/openapi'; import { connectRedis } from '@cio/core/utils/redis/redis'; import { env } from '@cio/core/config/env'; import { preloadVerifiedCustomDomainOriginsRegistry } from '@api/utils/origins'; +import { registerProcessErrorGuards } from '@api/process-error-guards'; import { serve } from '@hono/node-server'; import { showRoutes } from 'hono/dev'; +registerProcessErrorGuards(); + // Start server async function startServer() { console.log('Starting server on port:', API_PORT); diff --git a/apps/api/src/instrument.ts b/apps/api/src/instrument.ts index ccf244af..58d61acd 100644 --- a/apps/api/src/instrument.ts +++ b/apps/api/src/instrument.ts @@ -15,6 +15,13 @@ if (dsn && isProd && !isSelfHosted) { dsn, environment: process.env.SENTRY_ENVIRONMENT?.trim() || 'production', tracesSampleRate: Number(process.env.SENTRY_TRACES_SAMPLE_RATE ?? 0), - sendDefaultPii: false + sendDefaultPii: false, + // process-error-guards.ts registers handlers with transient Redis filtering. + integrations(integrations) { + return integrations.filter( + (integration) => + integration.name !== 'OnUncaughtException' && integration.name !== 'OnUnhandledRejection' + ); + } }); } diff --git a/apps/api/src/process-error-guards.ts b/apps/api/src/process-error-guards.ts new file mode 100644 index 00000000..21eb9754 --- /dev/null +++ b/apps/api/src/process-error-guards.ts @@ -0,0 +1,34 @@ +import * as Sentry from '@sentry/node'; + +import { isTransientRedisError, logTransientRedisEvent } from '@cio/core/utils/redis/redis'; + +function reportFatalProcessError(kind: 'uncaughtException' | 'unhandledRejection', error: unknown): void { + console.error(kind === 'uncaughtException' ? 'Uncaught exception:' : 'Unhandled rejection:', error); + Sentry.captureException(error); + void Sentry.flush(2000).finally(() => process.exit(1)); +} + +/** + * Custom process guards (see instrument.ts: Sentry's default OnUncaughtException / + * OnUnhandledRejection integrations are filtered out when Sentry inits so these + * own flush/exit). Transient Redis maintenance errors must not take down the API. + */ +export function registerProcessErrorGuards(): void { + process.on('uncaughtException', (error) => { + if (isTransientRedisError(error)) { + logTransientRedisEvent('Ignoring transient Redis error (client will reconnect):', error); + return; + } + + reportFatalProcessError('uncaughtException', error); + }); + + process.on('unhandledRejection', (reason) => { + if (isTransientRedisError(reason)) { + logTransientRedisEvent('Ignoring transient Redis rejection (client will reconnect):', reason); + return; + } + + reportFatalProcessError('unhandledRejection', reason); + }); +} diff --git a/packages/core/src/utils/redis/redis.ts b/packages/core/src/utils/redis/redis.ts index b6e73786..8df11ffe 100644 --- a/packages/core/src/utils/redis/redis.ts +++ b/packages/core/src/utils/redis/redis.ts @@ -1,12 +1,34 @@ import { createClient, type RedisClientType } from 'redis'; import { env } from '../../config/env'; -// Create Redis client +/** + * General-purpose Redis client (rate limiting, cache, OAuth handoff). + * + * Configured for Redis Cloud maintenance / failover resilience: + * - automatic reconnect with exponential backoff + jitter + * - error listeners so socket drops never crash the process + * - fresh DNS on each reconnect (node-redis default) + * + * See https://redis.io/docs/latest/operate/rc/resilient-apps/ + * and https://redis.io/docs/latest/develop/clients/nodejs/connect/ + */ const client: RedisClientType = createClient({ - url: env.REDIS_URL + url: env.REDIS_URL, + socket: { + // Keep retrying through Redis Cloud maintenance windows; never return false. + reconnectStrategy(retries) { + const jitter = Math.floor(Math.random() * 100); + const delay = Math.min(2 ** retries * 50, 3000); + + return delay + jitter; + }, + keepAlive: 10_000, + connectTimeout: 10_000 + } }); let hasLoggedRedisUnavailable = false; +let isConnected = false; export function logRedisUnavailableOnce(message: string, error?: unknown): void { if (hasLoggedRedisUnavailable) { @@ -23,24 +45,105 @@ export function logRedisUnavailableOnce(message: string, error?: unknown): void console.warn(message); } +const TRANSIENT_REDIS_ERROR_CODES = new Set([ + 'ECONNABORTED', + 'ECONNRESET', + 'ECONNREFUSED', + 'ETIMEDOUT', + 'EPIPE', + 'ENOTFOUND', + 'EAI_AGAIN', + 'EHOSTUNREACH', + 'ENETUNREACH' +]); + +const TRANSIENT_REDIS_MESSAGE_PATTERN = + /ECONNABORTED|ECONNRESET|ECONNREFUSED|ETIMEDOUT|Connection is closed|Socket closed unexpectedly|Connection timeout|client is closed/i; + +const TRANSIENT_REDIS_LOG_INTERVAL_MS = 10_000; +let lastTransientRedisLogAt = 0; + +export function logTransientRedisEvent(message: string, error?: unknown): void { + const now = Date.now(); + if (now - lastTransientRedisLogAt < TRANSIENT_REDIS_LOG_INTERVAL_MS) { + return; + } + + lastTransientRedisLogAt = now; + + if (error) { + console.warn(message, error); + return; + } + + console.warn(message); +} + +/** + * True for the connection/socket errors Redis Cloud maintenance & failover + * throw (ioredis and node-redis alike). Handles `AggregateError` (node-redis + * surfaces multi-address connect failures this way, e.g. IPv6 + IPv4) by + * recursing into its sub-errors. Exported so the API's process-level + * `uncaughtException` guard can distinguish "Redis is briefly unreachable, + * the client will reconnect" from a genuinely fatal bug. + */ +export function isTransientRedisError(error: unknown): boolean { + if (!error || typeof error !== 'object') { + return false; + } + + const code = 'code' in error && typeof error.code === 'string' ? error.code : ''; + if (TRANSIENT_REDIS_ERROR_CODES.has(code)) { + return true; + } + + const name = 'name' in error && typeof error.name === 'string' ? error.name : ''; + const message = 'message' in error && typeof error.message === 'string' ? error.message : ''; + if (TRANSIENT_REDIS_MESSAGE_PATTERN.test(message) || TRANSIENT_REDIS_MESSAGE_PATTERN.test(name)) { + return true; + } + + if ('errors' in error && Array.isArray((error as AggregateError).errors)) { + return (error as AggregateError).errors.some((sub) => isTransientRedisError(sub)); + } + + return false; +} + // Handle connection events client.on('error', (err) => { + if (isTransientRedisError(err)) { + // Expected during Redis Cloud maintenance — client reconnects automatically. + logTransientRedisEvent('Redis socket error (reconnecting):', err); + return; + } + logRedisUnavailableOnce('Redis unavailable, Redis-backed features disabled', err); }); +client.on('reconnecting', () => { + logTransientRedisEvent('Redis client reconnecting'); +}); + client.on('connect', () => { console.log('Redis Client Connected'); }); -// Track connection state -let isConnected = false; +client.on('ready', () => { + console.log('Redis Client Ready'); + isConnected = true; +}); + +client.on('end', () => { + isConnected = false; +}); export async function connectRedis(): Promise { if (!env.REDIS_URL) { logRedisUnavailableOnce('REDIS_URL not set, Redis features disabled'); return; } - if (isConnected) return; + if (isConnected || client.isOpen) return; try { await client.connect(); isConnected = true; diff --git a/packages/jobs/src/connection.regression.test.ts b/packages/jobs/src/connection.regression.test.ts new file mode 100644 index 00000000..7f74f7b0 --- /dev/null +++ b/packages/jobs/src/connection.regression.test.ts @@ -0,0 +1,71 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import test from 'node:test'; + +import type { Redis } from 'ioredis'; + +import { prepareRedisConnection } from './connection'; + +/** + * Regression: BullMQ calls `duplicate()` for secondary clients. Those copies + * must get an `error` listener, otherwise Redis Cloud maintenance disconnects + * (`ECONNABORTED`) crash the Node process as unhandled EventEmitter errors. + * + * We drive the REAL `prepareRedisConnection` from `connection.ts` against a + * fake Redis EventEmitter so this stays runnable without a live Redis and + * actually protects the production code path. + */ + +type FakeRedis = EventEmitter & { + duplicate: (options?: object) => FakeRedis; +}; + +function createFakeRedis(): FakeRedis { + const connection = new EventEmitter() as FakeRedis; + + connection.duplicate = () => createFakeRedis(); + prepareRedisConnection(connection as unknown as Redis); + + return connection; +} + +test('duplicate connections get error listeners', () => { + const root = createFakeRedis(); + const firstDuplicate = root.duplicate(); + const nestedDuplicate = firstDuplicate.duplicate(); + + assert.ok(root.listenerCount('error') >= 1); + assert.ok(firstDuplicate.listenerCount('error') >= 1); + assert.ok(nestedDuplicate.listenerCount('error') >= 1); +}); + +test('preparing the same connection twice does not stack listeners', () => { + const root = createFakeRedis(); + prepareRedisConnection(root as unknown as Redis); + + assert.equal(root.listenerCount('error'), 1); +}); + +test('preparing the same connection twice does not re-wrap duplicate()', () => { + const root = createFakeRedis(); + const duplicateAfterFirst = root.duplicate; + + prepareRedisConnection(root as unknown as Redis); + + assert.equal(root.duplicate, duplicateAfterFirst); +}); + +test('emitting error on duplicates does not throw', () => { + const root = createFakeRedis(); + const duplicated = root.duplicate(); + const abortError = Object.assign(new Error('read ECONNABORTED'), { + code: 'ECONNABORTED', + errno: -103, + syscall: 'read' + }); + + assert.doesNotThrow(() => { + root.emit('error', abortError); + duplicated.emit('error', abortError); + }); +}); diff --git a/packages/jobs/src/connection.ts b/packages/jobs/src/connection.ts index 7f230fac..0e010369 100644 --- a/packages/jobs/src/connection.ts +++ b/packages/jobs/src/connection.ts @@ -11,14 +11,63 @@ import IORedis, { type Redis, type RedisOptions } from 'ioredis'; * Self-hosted note: Redis is required to enqueue or process jobs. Callers * should fail loudly when `REDIS_URL` is not configured rather than silently * dropping work. + * + * Redis Cloud note: scheduled maintenance / failover commonly surfaces as + * `ECONNABORTED` / `ECONNRESET` / `READONLY`. Clients must keep reconnecting + * (with DNS refresh) and must attach `error` listeners on every connection — + * including BullMQ `duplicate()` copies — or Node exits on unhandled errors. + * See https://redis.io/docs/latest/operate/rc/resilient-apps/ */ const DEFAULT_OPTIONS: RedisOptions = { // Required by BullMQ - blocking commands need this disabled to work. maxRetriesPerRequest: null, - enableReadyCheck: false + enableReadyCheck: false, + // Keep reconnecting through Redis Cloud maintenance windows (minutes, not seconds). + // Caps delay at 2s; always returns a number so ioredis never gives up. + retryStrategy(times) { + return Math.min(times * 50, 2000); + }, + // After a replica promotion, Redis may reply READONLY until we reconnect to the new primary. + reconnectOnError(error) { + const message = error.message.toUpperCase(); + if (message.includes('READONLY') || message.includes('LOADING')) { + // 2 = reconnect AND resend the failed command against the new primary. + return 2; + } + + return false; + }, + // Detect half-open sockets during provider maintenance (0 = OS default; 10s is safer for cloud). + keepAlive: 10_000, + // Queue commands while reconnecting so producers/workers ride out brief handoffs. + enableOfflineQueue: true, + autoResendUnfulfilledCommands: true }; +/** Marker so we don't stack duplicate error listeners when wrapping `duplicate()`. */ +const ERROR_HANDLER_ATTACHED = Symbol.for('cio.bullmq.redisErrorHandler'); +const DUPLICATE_WRAPPED = Symbol.for('cio.bullmq.duplicateWrapped'); + +const BULLMQ_REDIS_LOG_INTERVAL_MS = 10_000; +let lastBullMqRedisLogAt = 0; + +function logBullMqRedisEvent(message: string, error?: unknown): void { + const now = Date.now(); + if (now - lastBullMqRedisLogAt < BULLMQ_REDIS_LOG_INTERVAL_MS) { + return; + } + + lastBullMqRedisLogAt = now; + + if (error) { + console.error(message, error); + return; + } + + console.warn(message); +} + export interface RedisConnectionOptions extends RedisOptions { /** Override the URL; falls back to REDIS_URL env. */ url?: string; @@ -43,6 +92,65 @@ function resolveUrl(override?: string): string { return url; } +function attachRedisLifecycleHandlers(connection: Redis): void { + const flagged = connection as Redis & { [ERROR_HANDLER_ATTACHED]?: boolean }; + + if (flagged[ERROR_HANDLER_ATTACHED]) { + return; + } + + flagged[ERROR_HANDLER_ATTACHED] = true; + + connection.on('error', (error) => { + // Transient disconnects (ECONNABORTED / ECONNRESET / ETIMEDOUT) are expected + // during Redis Cloud maintenance; ioredis reconnects automatically. Log only. + logBullMqRedisEvent('BullMQ Redis connection error:', error); + }); + + connection.on('reconnecting', (delay: number) => { + logBullMqRedisEvent(`BullMQ Redis reconnecting in ${delay}ms`); + }); + + connection.on('ready', () => { + console.log('BullMQ Redis connection ready'); + }); +} + +/** + * BullMQ calls `connection.duplicate()` for blocking / secondary clients. + * Duplicates inherit options but NOT event listeners. An `error` event on a + * listener-less ioredis client becomes an uncaught exception and exits the + * Node process (seen in production as cascading `read ECONNABORTED` crashes + * during Redis Cloud maintenance). + */ +function wrapDuplicate(connection: Redis): void { + const flagged = connection as Redis & { [DUPLICATE_WRAPPED]?: boolean }; + + if (flagged[DUPLICATE_WRAPPED]) { + return; + } + + flagged[DUPLICATE_WRAPPED] = true; + + const originalDuplicate = connection.duplicate.bind(connection); + + connection.duplicate = ((options?: object) => { + const duplicated = originalDuplicate(options); + prepareRedisConnection(duplicated); + return duplicated; + }) as typeof connection.duplicate; +} + +/** + * Attach lifecycle handlers and recursively wrap `duplicate()` so every + * connection BullMQ derives from this one is crash-safe. Exported so tests can + * assert against the real implementation instead of a copy. + */ +export function prepareRedisConnection(connection: Redis): void { + attachRedisLifecycleHandlers(connection); + wrapDuplicate(connection); +} + /** * Create a brand new ioredis connection. Use this for each Worker process. */ @@ -51,9 +159,7 @@ export function createRedisConnection(options: RedisConnectionOptions = {}): Red const resolved = resolveUrl(url); const connection = new IORedis(resolved, { ...DEFAULT_OPTIONS, ...rest }); - connection.on('error', (error) => { - console.error('BullMQ Redis connection error:', error); - }); + prepareRedisConnection(connection); return connection; } diff --git a/packages/jobs/src/queues/factories.ts b/packages/jobs/src/queues/factories.ts index 2cd0fb7f..39ec1565 100644 --- a/packages/jobs/src/queues/factories.ts +++ b/packages/jobs/src/queues/factories.ts @@ -20,6 +20,9 @@ export function getQueue(name: QueueName): Queue { ...buildQueueOptions(name), connection: getSharedRedisConnection() }); + queue.on('error', (error) => { + console.error(`BullMQ queue error (${name}):`, error); + }); queues.set(name, queue); } @@ -29,6 +32,9 @@ export function getQueue(name: QueueName): Queue { export function getFlowProducer(): FlowProducer { if (!flowProducer) { flowProducer = new FlowProducer({ connection: getSharedRedisConnection() }); + flowProducer.on('error', (error) => { + console.error('BullMQ flow producer error:', error); + }); } return flowProducer;