From 2d794a04390397f51d2f2f7b509e3c0be11010b3 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Thu, 3 Sep 2026 09:45:16 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=92=93=20fix:=20Detect=20Silently=20Dead?= =?UTF-8?q?=20Redis=20Sockets=20With=20Deadline-Bounded=20Heartbeats=20(#1?= =?UTF-8?q?5539)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: Detect silently dead Redis sockets with deadline-bounded heartbeats A Redis peer can vanish without a FIN or RST (a dropped NAT entry, a proxy failover, a migrated VM). The socket stays `ready` from ioredis's point of view, no error fires, and every command on it waits for the kernel's retransmission timeout — about fifteen minutes at Linux defaults. Observed in production: while the main client's socket was dead, POST /api/agents hung until Cloudflare returned a 524; while the subscriber's socket was dead, every attach failed with "Timed out synchronizing Redis subscription" for sixteen minutes across five conversations. `REDIS_PING_INTERVAL` could not help — it awaits ping() with no deadline, so the probe just joins everything else that is waiting. - Add startRedisHeartbeat: races a PING against a deadline and destroys the socket when the probe goes unanswered, so ioredis reconnects, replays queued commands, and re-subscribes a subscriber's channels. - Destroy the stream rather than disconnect(true), whose FIN a vanished peer never acknowledges; cluster clients fall back to the regular reconnect. - Add createIoRedisSubscriber so every dedicated subscriber gets an error listener and its own heartbeat, and route the generation-stream and both subagent-routing subscribers through it. - Set ioredis keepAlive from REDIS_KEEP_ALIVE (default 10s). - Add REDIS_PING_TIMEOUT, REDIS_SUBSCRIBER_PING_INTERVAL and REDIS_KEEP_ALIVE, documented in .env.example. - Give the fake RESP server a silent mode and a SUBSCRIBE reply so the dead-peer case can be tested against real clients. * fix: Probe each cluster node and refuse a non-positive heartbeat deadline A cluster-level PING is routed to one node, so a healthy reply proved nothing about the other sockets. The heartbeat now resolves its targets each tick — every node for a cluster, the client itself otherwise — and races an independent deadline per target, tearing down only the socket that stops answering. A non-positive deadline disables the heartbeat with a warning instead of forcing a reconnect on every tick, and the single-probe test no longer depends on wall-clock slack. --- .env.example | 12 + .../Endpoints/agents/subagentThreadStore.js | 11 +- .../agents/subagentThreadStore.spec.js | 8 +- .../src/cache/__tests__/cacheConfig.spec.ts | 23 ++ .../api/src/cache/__tests__/heartbeat.spec.ts | 213 ++++++++++++++++++ .../api/src/cache/__tests__/resp.helper.ts | 16 +- packages/api/src/cache/cacheConfig.ts | 12 + packages/api/src/cache/heartbeat.ts | 140 ++++++++++++ packages/api/src/cache/redisClients.ts | 27 +-- packages/api/src/cache/redisUtils.ts | 25 ++ .../api/src/stream/createStreamServices.ts | 3 +- 11 files changed, 466 insertions(+), 24 deletions(-) create mode 100644 packages/api/src/cache/__tests__/heartbeat.spec.ts create mode 100644 packages/api/src/cache/heartbeat.ts diff --git a/.env.example b/.env.example index b78fe8363c..5edbdb117f 100644 --- a/.env.example +++ b/.env.example @@ -1258,6 +1258,18 @@ HELP_AND_FAQ_URL=https://librechat.ai # When unset or 0, no pinging is performed (recommended for most use cases) # REDIS_PING_INTERVAL=300 +# Milliseconds a heartbeat PING may go unanswered before the socket is presumed dead and +# reconnected (default: 5000). Applies to REDIS_PING_INTERVAL and the subscriber heartbeat. +# REDIS_PING_TIMEOUT=5000 + +# Heartbeat interval in seconds for dedicated pub/sub subscriber connections (default: 15, 0 = disabled) +# Subscribers carry no traffic between generations, so a peer that vanished without closing the +# socket would otherwise go unnoticed until the kernel gives up (about 15 minutes at Linux defaults) +# REDIS_SUBSCRIBER_PING_INTERVAL=15 + +# TCP keepalive idle delay in milliseconds for ioredis sockets (default: 10000, 0 = kernel default) +# REDIS_KEEP_ALIVE=10000 + # Force specific cache namespaces to use in-memory storage even when Redis is enabled # Comma-separated list of CacheKeys # Defaults to CONFIG_STORE,APP_CONFIG so YAML-derived config stays per-container (safe for blue/green deployments) diff --git a/api/server/services/Endpoints/agents/subagentThreadStore.js b/api/server/services/Endpoints/agents/subagentThreadStore.js index 27620d3b2c..b33b7df541 100644 --- a/api/server/services/Endpoints/agents/subagentThreadStore.js +++ b/api/server/services/Endpoints/agents/subagentThreadStore.js @@ -3,6 +3,7 @@ const { ioredisClient, registerShutdownTask, duplicateIoRedisClient, + createIoRedisSubscriber, createSubagentThreadTaskStore, createSubagentCompletionWakeupHandler, GenerationJobManager, @@ -110,13 +111,19 @@ async function configureSubagentTaskRouting() { if (ioredisClient == null || typeof ioredisClient.duplicate !== 'function') { throw new Error('Redis subagent task routing requires a dedicated subscriber connection.'); } - const subscriber = ioredisClient.duplicate(); + const subscriber = createIoRedisSubscriber( + ioredisClient, + '[SubagentTaskRouting] task subscriber', + ); /** A dedicated publisher without the offline queue: the shared client would hold a * command issued during a disconnect and deliver it after the caller gave up, so a * steer the caller was told had failed could still reach the child. Failing fast * turns that into the honest `unavailable` the caller already handles. */ const publisher = duplicateIoRedisClient(ioredisClient, { enableOfflineQueue: false }); - const activitySubscriber = ioredisClient.duplicate(); + const activitySubscriber = createIoRedisSubscriber( + ioredisClient, + '[SubagentTaskRouting] activity subscriber', + ); const activityPublisher = duplicateIoRedisClient(ioredisClient, { enableOfflineQueue: false }); const transport = new RedisSubagentTaskControlTransport(publisher, subscriber, { namespace: cacheConfig.REDIS_KEY_PREFIX, diff --git a/api/server/services/Endpoints/agents/subagentThreadStore.spec.js b/api/server/services/Endpoints/agents/subagentThreadStore.spec.js index 3883f2ada4..bae8a2cdeb 100644 --- a/api/server/services/Endpoints/agents/subagentThreadStore.spec.js +++ b/api/server/services/Endpoints/agents/subagentThreadStore.spec.js @@ -12,6 +12,7 @@ jest.mock('@librechat/api', () => ({ ioredisClient: { duplicate: jest.fn() }, registerShutdownTask: jest.fn(), duplicateIoRedisClient: jest.fn(), + createIoRedisSubscriber: jest.fn(), createSubagentThreadTaskStore: jest.fn(() => mockTaskStore), createSubagentCompletionWakeupHandler: jest.fn(() => mockCompletionWakeupHandler), RedisSubagentTaskControlTransport: jest.fn(), @@ -50,6 +51,7 @@ const { ioredisClient, registerShutdownTask, duplicateIoRedisClient, + createIoRedisSubscriber, createSubagentThreadTaskStore, } = require('@librechat/api'); const subagentThreadTaskStore = require('./subagentThreadStore'); @@ -91,7 +93,7 @@ describe('subagent thread Redis lifecycle', () => { const activitySubscriber = { disconnect: jest.fn() }; const taskPublisher = { disconnect: jest.fn() }; const activityPublisher = { disconnect: jest.fn() }; - ioredisClient.duplicate + createIoRedisSubscriber .mockReturnValueOnce(taskSubscriber) .mockReturnValueOnce(activitySubscriber); duplicateIoRedisClient @@ -100,6 +102,10 @@ describe('subagent thread Redis lifecycle', () => { await configureSubagentTaskRouting(); + expect(createIoRedisSubscriber.mock.calls).toEqual([ + [ioredisClient, '[SubagentTaskRouting] task subscriber'], + [ioredisClient, '[SubagentTaskRouting] activity subscriber'], + ]); expect(activityPrepareRegistration).toEqual([ 'subagent activity streams prepare', expect.any(Function), diff --git a/packages/api/src/cache/__tests__/cacheConfig.spec.ts b/packages/api/src/cache/__tests__/cacheConfig.spec.ts index 0ef09a2710..a9f1f71e8b 100644 --- a/packages/api/src/cache/__tests__/cacheConfig.spec.ts +++ b/packages/api/src/cache/__tests__/cacheConfig.spec.ts @@ -14,6 +14,9 @@ describe('cacheConfig', () => { delete process.env.USE_REDIS_CLUSTER; delete process.env.REDIS_CLUSTER_SAFE_DELETE; delete process.env.REDIS_PING_INTERVAL; + delete process.env.REDIS_PING_TIMEOUT; + delete process.env.REDIS_SUBSCRIBER_PING_INTERVAL; + delete process.env.REDIS_KEEP_ALIVE; delete process.env.FORCED_IN_MEMORY_CACHE_NAMESPACES; delete process.env.VIOLATION_SCORE_TTL; @@ -222,6 +225,26 @@ describe('cacheConfig', () => { }); }); + describe('Redis dead-socket detection configuration', () => { + test('defaults the heartbeat deadline, subscriber interval, and keepalive', async () => { + const { cacheConfig } = await import('../cacheConfig'); + expect(cacheConfig.REDIS_PING_TIMEOUT).toBe(5000); + expect(cacheConfig.REDIS_SUBSCRIBER_PING_INTERVAL).toBe(15); + expect(cacheConfig.REDIS_KEEP_ALIVE).toBe(10000); + }); + + test('reads the provided values, including zero to disable', async () => { + process.env.REDIS_PING_TIMEOUT = '2500'; + process.env.REDIS_SUBSCRIBER_PING_INTERVAL = '0'; + process.env.REDIS_KEEP_ALIVE = '0'; + + const { cacheConfig } = await import('../cacheConfig'); + expect(cacheConfig.REDIS_PING_TIMEOUT).toBe(2500); + expect(cacheConfig.REDIS_SUBSCRIBER_PING_INTERVAL).toBe(0); + expect(cacheConfig.REDIS_KEEP_ALIVE).toBe(0); + }); + }); + describe('FORCED_IN_MEMORY_CACHE_NAMESPACES validation', () => { test('should parse comma-separated cache keys correctly', async () => { process.env.FORCED_IN_MEMORY_CACHE_NAMESPACES = ' ROLES, MESSAGES '; diff --git a/packages/api/src/cache/__tests__/heartbeat.spec.ts b/packages/api/src/cache/__tests__/heartbeat.spec.ts new file mode 100644 index 0000000000..79c28595b6 --- /dev/null +++ b/packages/api/src/cache/__tests__/heartbeat.spec.ts @@ -0,0 +1,213 @@ +import Redis from 'ioredis'; +import type { HeartbeatClient } from '~/cache/heartbeat'; +import type { RespServer } from './resp.helper'; +import { startRedisHeartbeat, forceRedisReconnect } from '~/cache/heartbeat'; +import { startRespServer, waitFor } from './resp.helper'; + +const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); + +const countCommands = (server: RespServer, name: string): number => + server.commands.filter((frame) => frame[0]?.toUpperCase() === name).length; + +describe('startRedisHeartbeat', () => { + let server: RespServer; + let client: Redis; + let stop: () => void = () => undefined; + + beforeEach(async () => { + server = await startRespServer(); + client = new Redis(server.url, { + lazyConnect: true, + maxRetriesPerRequest: null, + retryStrategy: () => 10, + }); + client.on('error', () => undefined); + await client.connect(); + }); + + afterEach(async () => { + stop(); + client.disconnect(); + await server.close(); + }); + + it('leaves an answering connection alone', async () => { + stop = startRedisHeartbeat({ client, intervalMs: 20, timeoutMs: 200, label: 'test' }); + await waitFor(() => countCommands(server, 'PING') >= 3); + expect(server.connections).toBe(1); + expect(client.status).toBe('ready'); + }); + + it('tears down a socket whose peer stops answering so ioredis reconnects and replays', async () => { + stop = startRedisHeartbeat({ client, intervalMs: 20, timeoutMs: 60, label: 'test' }); + server.silent = true; + const pending = client.get('key'); + + await waitFor(() => client.status !== 'ready'); + server.silent = false; + + await expect(pending).resolves.toBeNull(); + await waitFor(() => client.status === 'ready'); + expect(server.connections).toBe(2); + }); + + it('sends one probe at a time while a reply is outstanding', async () => { + stop = startRedisHeartbeat({ client, intervalMs: 20, timeoutMs: 5000, label: 'test' }); + server.silent = true; + await waitFor(() => countCommands(server, 'PING') >= 1); + await sleep(100); + expect(countCommands(server, 'PING')).toBe(1); + expect(server.connections).toBe(1); + }); + + it('re-subscribes a subscriber connection after forcing it to reconnect', async () => { + await client.subscribe('events'); + stop = startRedisHeartbeat({ client, intervalMs: 20, timeoutMs: 60, label: 'test' }); + await waitFor(() => countCommands(server, 'PING') >= 2); + expect(server.connections).toBe(1); + + server.silent = true; + await waitFor(() => client.status !== 'ready'); + server.silent = false; + + await waitFor(() => countCommands(server, 'SUBSCRIBE') >= 2); + await waitFor(() => client.status === 'ready'); + expect(server.connections).toBe(2); + }); + + it('stops probing once the client has ended', async () => { + stop = startRedisHeartbeat({ client, intervalMs: 20, timeoutMs: 200, label: 'test' }); + await waitFor(() => countCommands(server, 'PING') >= 1); + client.disconnect(); + await waitFor(() => client.status === 'end'); + const sent = countCommands(server, 'PING'); + await sleep(100); + expect(countCommands(server, 'PING')).toBe(sent); + }); + + it('does nothing when the interval is not positive', () => { + stop = startRedisHeartbeat({ client, intervalMs: 0, timeoutMs: 200, label: 'test' }); + expect(client.listenerCount('end')).toBe(0); + }); + + it('disables itself instead of reconnecting on every tick when the deadline is not positive', async () => { + stop = startRedisHeartbeat({ client, intervalMs: 20, timeoutMs: 0, label: 'test' }); + expect(client.listenerCount('end')).toBe(0); + await sleep(100); + expect(countCommands(server, 'PING')).toBe(0); + expect(server.connections).toBe(1); + }); +}); + +describe('startRedisHeartbeat on a cluster', () => { + type FakeNode = HeartbeatClient & { + destroy: jest.Mock; + pings: number; + pingsAtDestroy: number[]; + }; + + const fakeNode = (host: string, answers: boolean, status = 'ready'): FakeNode => { + const node: FakeNode = { + status, + options: { host, port: 6379 }, + pings: 0, + pingsAtDestroy: [], + destroy: jest.fn(() => { + node.pingsAtDestroy.push(node.pings); + }), + ping: () => { + node.pings += 1; + return answers ? Promise.resolve('PONG') : new Promise(() => undefined); + }, + on: () => undefined, + off: () => undefined, + disconnect: jest.fn(), + stream: { destroyed: false, destroy: (error?: Error) => node.destroy(error) }, + }; + return node; + }; + + const fakeCluster = (nodes: FakeNode[]): HeartbeatClient & { ping: jest.Mock } => ({ + status: 'ready', + ping: jest.fn(() => Promise.resolve('PONG')), + on: () => undefined, + off: () => undefined, + disconnect: jest.fn(), + nodes: () => nodes, + }); + + it('probes every node and tears down only the one that stops answering', async () => { + const healthy = fakeNode('10.0.0.1', true); + const dead = fakeNode('10.0.0.2', false); + const cluster = fakeCluster([healthy, dead]); + const stop = startRedisHeartbeat({ + client: cluster, + intervalMs: 10, + timeoutMs: 40, + label: 'cluster', + }); + + await waitFor(() => dead.destroy.mock.calls.length >= 1); + stop(); + + expect(cluster.ping).not.toHaveBeenCalled(); + expect(healthy.pings).toBeGreaterThanOrEqual(2); + expect(healthy.destroy).not.toHaveBeenCalled(); + expect(dead.pingsAtDestroy[0]).toBe(1); + expect((dead.destroy.mock.calls[0][0] as Error).message).toContain('10.0.0.2:6379'); + }); + + it('skips nodes that are not ready', async () => { + const ready = fakeNode('10.0.0.1', true); + const reconnecting = fakeNode('10.0.0.2', false, 'reconnecting'); + const stop = startRedisHeartbeat({ + client: fakeCluster([ready, reconnecting]), + intervalMs: 10, + timeoutMs: 40, + label: 'cluster', + }); + + await waitFor(() => ready.pings >= 2); + await sleep(60); + stop(); + + expect(reconnecting.pings).toBe(0); + expect(reconnecting.destroy).not.toHaveBeenCalled(); + }); +}); + +describe('forceRedisReconnect', () => { + it('falls back to a regular reconnect when the client exposes no socket', () => { + const disconnect = jest.fn(); + forceRedisReconnect( + { + status: 'ready', + ping: () => Promise.resolve(), + on: () => undefined, + off: () => undefined, + disconnect, + }, + 'test', + ); + expect(disconnect).toHaveBeenCalledWith(true); + }); + + it('destroys the live socket with the reason instead of ending it', () => { + const destroy = jest.fn(); + const disconnect = jest.fn(); + forceRedisReconnect( + { + status: 'ready', + ping: () => Promise.resolve(), + on: () => undefined, + off: () => undefined, + disconnect, + stream: { destroyed: false, destroy }, + }, + 'peer vanished', + ); + expect(destroy).toHaveBeenCalledTimes(1); + expect((destroy.mock.calls[0][0] as Error).message).toBe('peer vanished'); + expect(disconnect).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/cache/__tests__/resp.helper.ts b/packages/api/src/cache/__tests__/resp.helper.ts index 4da642c173..99bf497960 100644 --- a/packages/api/src/cache/__tests__/resp.helper.ts +++ b/packages/api/src/cache/__tests__/resp.helper.ts @@ -14,6 +14,8 @@ export type RespServer = { connections: number; /** When true, write commands are rejected with the READONLY reply. */ readonly: boolean; + /** When true, frames are recorded but never answered: a peer that has silently gone away. */ + silent: boolean; commands: string[][]; close(): Promise; }; @@ -83,7 +85,7 @@ function parseFrames(buffer: Buffer): { frames: string[][]; rest: Buffer } { export async function startRespServer(port = 0): Promise { const store = new Map(); const sockets = new Set(); - const state = { connections: 0, readonly: false, commands: [] as string[][] }; + const state = { connections: 0, readonly: false, silent: false, commands: [] as string[][] }; const reply = (args: string[]): string => { const command = args[0]?.toUpperCase() ?? ''; @@ -93,6 +95,8 @@ export async function startRespServer(port = 0): Promise { switch (command) { case 'PING': return '+PONG\r\n'; + case 'SUBSCRIBE': + return `*3\r\n$9\r\nsubscribe\r\n${encodeBulk(args[1])}:1\r\n`; case 'INFO': return encodeBulk('# Server\r\nredis_version:7.2.4\r\nloading:0\r\n'); case 'GET': @@ -122,7 +126,9 @@ export async function startRespServer(port = 0): Promise { pending = rest; for (const frame of frames) { state.commands.push(frame); - socket.write(reply(frame)); + if (!state.silent) { + socket.write(reply(frame)); + } } }); socket.on('close', () => sockets.delete(socket)); @@ -144,6 +150,12 @@ export async function startRespServer(port = 0): Promise { set readonly(value: boolean) { state.readonly = value; }, + get silent() { + return state.silent; + }, + set silent(value: boolean) { + state.silent = value; + }, commands: state.commands, close: () => new Promise((resolve) => { diff --git a/packages/api/src/cache/cacheConfig.ts b/packages/api/src/cache/cacheConfig.ts index 619662c416..00cff485cd 100644 --- a/packages/api/src/cache/cacheConfig.ts +++ b/packages/api/src/cache/cacheConfig.ts @@ -88,6 +88,12 @@ const cacheConfig: { GLOBAL_PREFIX_SEPARATOR: string; REDIS_MAX_LISTENERS: number; REDIS_PING_INTERVAL: number; + /** Milliseconds a heartbeat PING may go unanswered before the socket is presumed dead */ + REDIS_PING_TIMEOUT: number; + /** Heartbeat interval in seconds for dedicated pub/sub subscriber connections (0 = disabled) */ + REDIS_SUBSCRIBER_PING_INTERVAL: number; + /** TCP keepalive idle delay in ms for ioredis sockets (0 = kernel default idle time) */ + REDIS_KEEP_ALIVE: number; /** Max delay between reconnection attempts in ms */ REDIS_RETRY_MAX_DELAY: number; /** Max number of reconnection attempts (0 = infinite) */ @@ -167,6 +173,12 @@ const cacheConfig: { GLOBAL_PREFIX_SEPARATOR: '::', REDIS_MAX_LISTENERS: math(process.env.REDIS_MAX_LISTENERS, 40), REDIS_PING_INTERVAL: math(process.env.REDIS_PING_INTERVAL, 0), + /** Milliseconds a heartbeat PING may go unanswered before the socket is presumed dead */ + REDIS_PING_TIMEOUT: math(process.env.REDIS_PING_TIMEOUT, 5000), + /** Heartbeat interval in seconds for dedicated pub/sub subscriber connections (0 = disabled) */ + REDIS_SUBSCRIBER_PING_INTERVAL: math(process.env.REDIS_SUBSCRIBER_PING_INTERVAL, 15), + /** TCP keepalive idle delay in ms for ioredis sockets (0 = kernel default idle time) */ + REDIS_KEEP_ALIVE: math(process.env.REDIS_KEEP_ALIVE, 10000), /** Max delay between reconnection attempts in ms */ REDIS_RETRY_MAX_DELAY: math(process.env.REDIS_RETRY_MAX_DELAY, 3000), /** Max number of reconnection attempts (0 = infinite) */ diff --git a/packages/api/src/cache/heartbeat.ts b/packages/api/src/cache/heartbeat.ts new file mode 100644 index 0000000000..778bf21b2b --- /dev/null +++ b/packages/api/src/cache/heartbeat.ts @@ -0,0 +1,140 @@ +import { logger } from '@librechat/data-schemas'; + +/** The slice of an ioredis `Redis` or `Cluster` client the heartbeat needs. */ +export interface HeartbeatClient { + readonly status: string; + readonly options?: object; + ping(): Promise; + on(event: 'end', listener: () => void): unknown; + off(event: 'end', listener: () => void): unknown; + disconnect(reconnect?: boolean): void; + /** Present on a standalone client and on every cluster node; a cluster itself owns no socket. */ + stream?: { readonly destroyed: boolean; destroy(error?: Error): unknown }; + /** A cluster exposes its node connections, each with a socket of its own to probe. */ + nodes?(role: 'all'): HeartbeatClient[]; +} + +export interface RedisHeartbeatOptions { + client: HeartbeatClient; + /** Milliseconds between probes; a non-positive value disables the heartbeat. */ + intervalMs: number; + /** Milliseconds a probe may go unanswered before the socket is presumed dead. */ + timeoutMs: number; + /** Client label used in log lines. */ + label: string; +} + +type ProbeOutcome = 'pong' | 'rejected' | 'expired'; + +function describeTarget(target: HeartbeatClient, label: string): string { + const options = target.options; + if (options != null && 'host' in options && 'port' in options) { + return `${label} ${String(options.host)}:${String(options.port)}`; + } + return label; +} + +/** + * A cluster's sockets belong to its nodes, and a cluster-level PING is routed to one of + * them, so a healthy reply there says nothing about the others. A standalone client is + * its own single target. + */ +function probeTargets(client: HeartbeatClient): HeartbeatClient[] { + return client.nodes != null ? client.nodes('all') : [client]; +} + +/** + * Drops the current socket without waiting for the peer. `disconnect(true)` calls + * `stream.end()`, whose FIN a vanished peer never acknowledges, so ioredis would not + * observe `close` until the kernel retransmission timeout — the very wait a heartbeat + * exists to avoid. Destroying the stream raises `close` immediately, after which ioredis + * reconnects through its retry strategy, replays the commands it was holding, and + * re-subscribes a subscriber's channels. A cluster node whose socket closes is dropped + * from the pool and recreated on the next slot refresh, with its commands retried through + * the cluster's redirection path. Anything without a socket falls back to the regular + * reconnect. + */ +export function forceRedisReconnect(client: HeartbeatClient, reason: string): void { + const stream = client.stream; + if (stream != null && !stream.destroyed) { + stream.destroy(new Error(reason)); + return; + } + client.disconnect(true); +} + +/** + * Probes each connection with deadline-bounded PINGs and tears a socket down when its + * probe goes unanswered. A peer that disappears without a FIN or RST (a dropped NAT + * entry, a proxy failover, a migrated VM) leaves the socket "connected" from ioredis's + * point of view: no error fires, every command waits, and only the kernel's + * retransmission or keepalive timeout — about fifteen minutes at Linux defaults — ends + * the wait. Dedicated subscriber connections are the worst case, since nothing else + * ever writes to them. A cluster is probed node by node for the same reason. + * + * Only a probe that neither resolves nor rejects within `timeoutMs` counts: a rejected + * probe means ioredis already knows the connection state and is handling it. A target is + * skipped while it is not ready or its previous probe is still waiting, and the heartbeat + * ends with the client (`end` fires only once ioredis stops reconnecting). + */ +export function startRedisHeartbeat(options: RedisHeartbeatOptions): () => void { + const { client, intervalMs, timeoutMs, label } = options; + if (intervalMs <= 0) { + return () => undefined; + } + if (timeoutMs <= 0) { + logger.warn( + `${label} heartbeat disabled: the probe deadline must be positive, got ${timeoutMs}ms`, + ); + return () => undefined; + } + + const inFlight = new Set(); + let stopped = false; + + const probe = async (target: HeartbeatClient): Promise => { + if (stopped || inFlight.has(target) || target.status !== 'ready') { + return; + } + inFlight.add(target); + let deadline: ReturnType | undefined; + const expired = new Promise((resolve) => { + deadline = setTimeout(() => resolve('expired'), timeoutMs); + deadline.unref?.(); + }); + const answered = target.ping().then( + (): ProbeOutcome => 'pong', + (): ProbeOutcome => 'rejected', + ); + try { + const outcome = await Promise.race([answered, expired]); + if (outcome !== 'expired' || stopped) { + return; + } + const name = describeTarget(target, label); + logger.warn(`${name} heartbeat: no PING reply within ${timeoutMs}ms, reconnecting`); + forceRedisReconnect(target, `${name} heartbeat timed out after ${timeoutMs}ms`); + } finally { + clearTimeout(deadline); + inFlight.delete(target); + } + }; + + const timer = setInterval(() => { + for (const target of probeTargets(client)) { + void probe(target); + } + }, intervalMs); + timer.unref?.(); + + const stop = (): void => { + if (stopped) { + return; + } + stopped = true; + clearInterval(timer); + client.off('end', stop); + }; + client.on('end', stop); + return stop; +} diff --git a/packages/api/src/cache/redisClients.ts b/packages/api/src/cache/redisClients.ts index 076d73d99e..ebe87b7879 100644 --- a/packages/api/src/cache/redisClients.ts +++ b/packages/api/src/cache/redisClients.ts @@ -7,6 +7,7 @@ import type { RedisClientType, RedisClusterType } from '@redis/client'; import type { Redis, Cluster } from 'ioredis'; import type { ReadonlyRecoveryHandler } from './recovery'; import { createReadonlyRecovery, isReadonlyReplicaError } from './recovery'; +import { startRedisHeartbeat } from './heartbeat'; import { cacheConfig } from './cacheConfig'; const urls = cacheConfig.REDIS_URI?.split(',').map((uri) => new URL(uri)) || []; @@ -73,6 +74,7 @@ if (cacheConfig.USE_REDIS) { enableOfflineQueue: cacheConfig.REDIS_ENABLE_OFFLINE_QUEUE, connectTimeout: cacheConfig.REDIS_CONNECT_TIMEOUT, maxRetriesPerRequest: 3, + keepAlive: cacheConfig.REDIS_KEEP_ALIVE, }; ioredisClient = !isRedisCluster @@ -129,25 +131,14 @@ if (cacheConfig.USE_REDIS) { logger.warn('ioredis client connection closed'); }); - /** Ping Interval to keep the Redis server connection alive (if enabled) */ - let pingInterval: NodeJS.Timeout | null = null; - const clearPingInterval = () => { - if (pingInterval) { - clearInterval(pingInterval); - pingInterval = null; - } - }; - + /** Deadline-bounded keepalive PINGs; an unanswered probe forces a reconnect. */ if (cacheConfig.REDIS_PING_INTERVAL > 0) { - pingInterval = setInterval(() => { - if (ioredisClient && ioredisClient.status === 'ready') { - ioredisClient.ping().catch((err) => { - logger.error('ioredis ping failed:', err); - }); - } - }, cacheConfig.REDIS_PING_INTERVAL * 1000); - ioredisClient.on('close', clearPingInterval); - ioredisClient.on('end', clearPingInterval); + startRedisHeartbeat({ + client: ioredisClient, + intervalMs: cacheConfig.REDIS_PING_INTERVAL * 1000, + timeoutMs: cacheConfig.REDIS_PING_TIMEOUT, + label: 'ioredis client', + }); } } diff --git a/packages/api/src/cache/redisUtils.ts b/packages/api/src/cache/redisUtils.ts index 4903df8c08..91b23525ec 100644 --- a/packages/api/src/cache/redisUtils.ts +++ b/packages/api/src/cache/redisUtils.ts @@ -1,6 +1,7 @@ import { logger } from '@librechat/data-schemas'; import type { ClusterOptions, RedisOptions, Cluster, Redis } from 'ioredis'; import type { RedisClientType, RedisClusterType } from '@redis/client'; +import { startRedisHeartbeat } from './heartbeat'; import { cacheConfig } from './cacheConfig'; /** @@ -44,6 +45,30 @@ export function duplicateIoRedisClient( return (client as Redis).duplicate(options); } +/** + * Duplicates a client for a dedicated pub/sub subscriber. `duplicate()` copies options but + * not listeners, so a bare duplicate reports socket errors only through ioredis's + * "Unhandled error event" and, between generations, carries no traffic at all: a peer + * that vanishes without closing the socket stays undetected until the kernel gives up. + * The heartbeat is the traffic the shared client gets for free. + */ +export function createIoRedisSubscriber(client: Redis | Cluster, label: string): Redis | Cluster { + const subscriber = duplicateIoRedisClient(client); + subscriber.on('error', (error: Error) => { + logger.error(`${label} error:`, error); + }); + subscriber.on('ready', () => { + logger.info(`${label} ready`); + }); + startRedisHeartbeat({ + client: subscriber, + intervalMs: cacheConfig.REDIS_SUBSCRIBER_PING_INTERVAL * 1000, + timeoutMs: cacheConfig.REDIS_PING_TIMEOUT, + label, + }); + return subscriber; +} + /** * Efficiently deletes multiple Redis keys with support for both cluster and single-node modes. * diff --git a/packages/api/src/stream/createStreamServices.ts b/packages/api/src/stream/createStreamServices.ts index 942b0e95a3..73206de8e5 100644 --- a/packages/api/src/stream/createStreamServices.ts +++ b/packages/api/src/stream/createStreamServices.ts @@ -5,6 +5,7 @@ import { InMemoryEventTransport } from './implementations/InMemoryEventTransport import { RedisEventTransport } from './implementations/RedisEventTransport'; import { InMemoryJobStore } from './implementations/InMemoryJobStore'; import { RedisJobStore } from './implementations/RedisJobStore'; +import { createIoRedisSubscriber } from '~/cache/redisUtils'; import { ioredisClient } from '~/cache/redisClients'; import { cacheConfig } from '~/cache/cacheConfig'; @@ -82,7 +83,7 @@ export function createStreamServices(config: StreamServicesConfig = {}): StreamS let subscriber = redisSubscriber; if (!subscriber && 'duplicate' in redisClient) { - subscriber = (redisClient as Redis).duplicate(); + subscriber = createIoRedisSubscriber(redisClient, '[StreamServices] subscriber'); logger.info('[StreamServices] Duplicated Redis client for subscriber'); }