fix(jobs): survive Redis Cloud maintenance reconnects without crashing API (#775)
* fix(jobs): survive Redis Cloud maintenance reconnects BullMQ duplicates ioredis connections without inheriting error listeners, so Redis Cloud maintenance (ECONNABORTED) crashed the API process. Attach error handlers on duplicates, harden reconnect options for failover, and add API process-level safety nets. Co-authored-by: Rotimi Best <[email protected]> * refactor(jobs): address review feedback on Redis resilience fix - reconnectOnError returns 2 for READONLY/LOADING so the failed command is resent against the newly promoted primary after failover - export prepareRedisConnection and test the real implementation instead of a reimplemented copy; add a no-double-listener regression case - API uncaughtException now reports to Sentry and fails fast (flush + exit 1) so the orchestrator restarts cleanly instead of running in an undefined state Co-authored-by: Rotimi Best <[email protected]> * fix(api): don't fail-fast on transient Redis errors in process guard End-to-end testing (restarting Redis to simulate Cloud maintenance) revealed node-redis surfaces reconnect failures as an uncaught AggregateError [ECONNREFUSED], which the fail-fast uncaughtException guard turned into a process exit. Classify transient Redis/socket errors (incl. AggregateError sub-errors) and log-and-continue for those; still fail fast on unknown errors. Co-authored-by: Rotimi Best <[email protected]> * fix(api): address PR review on Redis process guards and logging - Fail fast on non-transient unhandledRejection (Greptile) - Filter Sentry OnUncaughtException/OnUnhandledRejection integrations to avoid double-capture; keep custom guards with transient Redis filtering - Make wrapDuplicate idempotent; throttle reconnect/error logs during maintenance - Drop READONLY/LOADING from node-redis transient classifier (command-level only) Co-authored-by: Rotimi Best <[email protected]> * fix(api): rate-limit transient Redis warnings in process guards Reuse logTransientRedisEvent from core so maintenance bursts do not flood logs from uncaughtException/unhandledRejection handlers. Co-authored-by: Rotimi Best <[email protected]> --------- Co-authored-by: Cursor Agent <[email protected]> Co-authored-by: Rotimi Best <[email protected]>
This commit is contained in:
Rotimi Best
co-authored by
Rotimi Best
Cursor Agent
Rotimi Best
Cursor Agent
parent
1a44983210
commit
2a53d675f6
@@ -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<void> {
|
||||
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;
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user