💓 fix: Detect Silently Dead Redis Sockets With Deadline-Bounded Heartbeats (#15539)

* 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

@@ -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)

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,

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),

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 ';

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<void> => 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();
+  });
+});

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<void>;
 };
@@ -83,7 +85,7 @@ function parseFrames(buffer: Buffer): { frames: string[][]; rest: Buffer } {
 export async function startRespServer(port = 0): Promise<RespServer> {
   const store = new Map<string, string>();
   const sockets = new Set<Socket>();
-  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<RespServer> {
     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<RespServer> {
       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<RespServer> {
     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<void>((resolve) => {

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) */

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<unknown>;
+  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<HeartbeatClient>();
+  let stopped = false;
+
+  const probe = async (target: HeartbeatClient): Promise<void> => {
+    if (stopped || inFlight.has(target) || target.status !== 'ready') {
+      return;
+    }
+    inFlight.add(target);
+    let deadline: ReturnType<typeof setTimeout> | undefined;
+    const expired = new Promise<ProbeOutcome>((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;
+}

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',
+    });
   }
 }
 

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.
  *

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');
       }