From 519e64769a74a0bd74ee90e106c9922c0a8544f8 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 2 Sep 2026 08:57:40 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A2=20fix:=20Reconnect=20Keyv=20Redis?= =?UTF-8?q?=20After=20Replica=20Failover=20(#15502)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 🔁 fix: Recover Keyv Redis Client From READONLY Replica After Failover When a Sentinel failover demotes the master a pod is connected to, the ioredis client recovers through `reconnectOnError`, but the Keyv/node-redis client never does: `socket.reconnectStrategy` only runs when the socket closes, and a demoted replica keeps existing sockets open while rejecting every write with `READONLY`. Every cache write through the Keyv client (USER_PRINCIPALS, MCP catalog, server configs) then fails until restart. - Add a READONLY recovery hook for the standalone node-redis client that tears the socket down and reconnects, re-resolving to the current master - Debounce attempts with `REDIS_READONLY_RECOVERY_INTERVAL` (5s default) and retry on the next error, since a reconnect issued before the topology settles can land on the demoted node again - Route all three surfaces where READONLY replies appear through the hook: the client error event, the Keyv cache error funnel, and Lua scripts evaluated via `evalKeyvRedisScript`, which reject straight to callers - Skip cluster clients, which follow MOVED to the owning master already - Cover the hook with a fake RESP server so real node-redis and Keyv code paths are exercised against a node whose role flips mid-connection Fixes #15500 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01WNbgMeR2aZDyhsqAy4L8gF * 🛟 fix: Harden READONLY Recovery Retry, Clear Path, and Clock Address review findings on the READONLY failover recovery: - Retry on the next error of any kind after a failed reconnect. A failed `connect()` leaves the client closed, so later failures surface as ClientClosedError rather than READONLY and the hook never fired again - Route the namespace `clear()` override through recovery. It deletes through the raw client, bypassing the Keyv error funnel, so a READONLY rejection there previously left the pod stranded - Debounce on an injectable monotonic clock (`performance.now`) instead of `Date.now`, so wall-clock steps cannot suppress or duplicate attempts; the interval test drives the clock instead of sleeping - Reuse `isReadonlyReplicaError` in the ioredis `reconnectOnError` hook so READONLY detection has one definition - Document the in-flight command rejection that a socket teardown implies Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01WNbgMeR2aZDyhsqAy4L8gF * 🛟 fix: Gate READONLY Recovery on Client State Address the second review pass on the READONLY failover recovery: - Act only on a client the hook can reason about: a READONLY reply on a ready socket, or a client its own failed reconnect left closed. While node-redis runs its own reconnect loop (open but not ready) the hook stays out, since a second `connect()` would start a concurrent loop that leaks sockets into one shared reply decoder - Clear the pending retry whenever an error arrives on a ready client, so a client that Keyv auto-connected after a failed attempt is never torn down by a later unrelated error - Drop the client error-event wiring: node-redis rejects command error replies straight to the command promise and emits `error` only from its socket and connect loop, so that listener could never see a READONLY reply and only fed connect-loop errors back into the hook - Cover the connect-loop, self-recovered, and Keyv auto-connect outage cases, and a rejected namespace clear, with a SCAN-aware RESP server - Add the new export to the `redisClients` mocks in the cacheFactory specs --- ...tandardCache.in_memory_memoization.spec.ts | 1 + .../standardCache.namespace_isolation.spec.ts | 1 + .../api/src/cache/__tests__/recovery.spec.ts | 246 ++++++++++++++++++ .../api/src/cache/__tests__/resp.helper.ts | 167 ++++++++++++ packages/api/src/cache/cacheConfig.ts | 4 + packages/api/src/cache/cacheFactory.ts | 54 ++-- packages/api/src/cache/recovery.ts | 116 +++++++++ packages/api/src/cache/redisClients.ts | 69 +++-- 8 files changed, 618 insertions(+), 40 deletions(-) create mode 100644 packages/api/src/cache/__tests__/recovery.spec.ts create mode 100644 packages/api/src/cache/__tests__/resp.helper.ts create mode 100644 packages/api/src/cache/recovery.ts diff --git a/packages/api/src/cache/__tests__/cacheFactory/standardCache.in_memory_memoization.spec.ts b/packages/api/src/cache/__tests__/cacheFactory/standardCache.in_memory_memoization.spec.ts index 16b016f9bd..19f87e6d22 100644 --- a/packages/api/src/cache/__tests__/cacheFactory/standardCache.in_memory_memoization.spec.ts +++ b/packages/api/src/cache/__tests__/cacheFactory/standardCache.in_memory_memoization.spec.ts @@ -5,6 +5,7 @@ jest.mock('@keyv/redis', () => ({ })); jest.mock('../../redisClients', () => ({ + handleKeyvRedisError: jest.fn(), keyvRedisClient: null, ioredisClient: null, })); diff --git a/packages/api/src/cache/__tests__/cacheFactory/standardCache.namespace_isolation.spec.ts b/packages/api/src/cache/__tests__/cacheFactory/standardCache.namespace_isolation.spec.ts index 9a8b4ff3bf..32a1492555 100644 --- a/packages/api/src/cache/__tests__/cacheFactory/standardCache.namespace_isolation.spec.ts +++ b/packages/api/src/cache/__tests__/cacheFactory/standardCache.namespace_isolation.spec.ts @@ -15,6 +15,7 @@ jest.mock('@keyv/redis', () => ({ const mockKeyvRedisClient = { scanIterator: jest.fn() }; jest.mock('../../redisClients', () => ({ + handleKeyvRedisError: jest.fn(), keyvRedisClient: mockKeyvRedisClient, ioredisClient: null, })); diff --git a/packages/api/src/cache/__tests__/recovery.spec.ts b/packages/api/src/cache/__tests__/recovery.spec.ts new file mode 100644 index 0000000000..1cbd3eb232 --- /dev/null +++ b/packages/api/src/cache/__tests__/recovery.spec.ts @@ -0,0 +1,246 @@ +import { createClient } from '@keyv/redis'; +import type { RedisClientType } from '@redis/client'; +import type { RespServer } from './resp.helper'; +import { createReadonlyRecovery, isReadonlyReplicaError } from '~/cache/recovery'; +import { closeRedisClients } from './redisClients.helper'; +import { startRespServer, waitFor } from './resp.helper'; + +const READONLY_MESSAGE = "READONLY You can't write against a read only replica."; + +const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); + +async function readonlyWriteError(client: RedisClientType): Promise { + try { + await client.set('key', 'value'); + } catch (error) { + return error; + } + throw new Error('Expected the write to be rejected'); +} + +describe('isReadonlyReplicaError', () => { + it('recognizes READONLY replies and nothing else', () => { + expect(isReadonlyReplicaError(new Error(READONLY_MESSAGE))).toBe(true); + expect(isReadonlyReplicaError(READONLY_MESSAGE)).toBe(true); + expect(isReadonlyReplicaError(new Error('ECONNRESET'))).toBe(false); + expect(isReadonlyReplicaError(undefined)).toBe(false); + }); +}); + +describe('createReadonlyRecovery', () => { + let server: RespServer; + let client: RedisClientType; + + const connectClient = async ( + reconnectStrategy: false | ((retries: number) => number), + ): Promise => { + client = createClient({ url: server.url, socket: { reconnectStrategy } }) as RedisClientType; + client.on('error', () => undefined); + await client.connect(); + }; + + beforeEach(async () => { + server = await startRespServer(); + await connectClient(false); + }); + + afterEach(async () => { + if (client.isOpen) { + client.destroy(); + } + await server.close(); + }); + + it('reconnects once per interval and keeps retrying while writes stay READONLY', async () => { + let clock = 0; + const recover = createReadonlyRecovery({ client, minIntervalMs: 200, now: () => clock }); + server.readonly = true; + const error = await readonlyWriteError(client); + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toBe(READONLY_MESSAGE); + + expect(recover(error)).toBe(true); + for (let i = 0; i < 50; i++) { + expect(recover(error)).toBe(false); + } + await waitFor(() => server.connections === 2 && client.isReady); + + expect(recover(await readonlyWriteError(client))).toBe(false); + clock = 200; + expect(recover(await readonlyWriteError(client))).toBe(true); + await waitFor(() => server.connections === 3 && client.isReady); + + server.readonly = false; + await expect(client.set('key', 'value')).resolves.toBe('OK'); + await expect(client.get('key')).resolves.toBe('value'); + }); + + it('ignores errors that are not READONLY replies', async () => { + const recover = createReadonlyRecovery({ client, minIntervalMs: 0 }); + expect(recover(new Error('ECONNRESET'))).toBe(false); + expect(recover(undefined)).toBe(false); + await sleep(50); + expect(server.connections).toBe(1); + }); + + it('retries on the next error of any kind after a failed reconnect', async () => { + const recover = createReadonlyRecovery({ client, minIntervalMs: 0 }); + const { port } = server; + await server.close(); + await waitFor(() => !client.isOpen); + expect(recover(new Error('The client is closed'))).toBe(false); + + server = await startRespServer(port); + await client.connect(); + server.readonly = true; + expect(recover(await readonlyWriteError(client))).toBe(true); + await server.close(); + await waitFor(() => !client.isOpen && server.connections === 1); + + expect(recover(new Error('ECONNRESET'))).toBe(true); + await waitFor(() => !client.isOpen); + + server = await startRespServer(port); + expect(recover(new Error('The client is closed'))).toBe(true); + await waitFor(() => server.connections === 1 && client.isReady); + await expect(client.set('key', 'value')).resolves.toBe('OK'); + }); + + it('stays out of the way while node-redis runs its own reconnect loop', async () => { + client.destroy(); + await connectClient(() => 20); + const recover = createReadonlyRecovery({ client, minIntervalMs: 0 }); + const { port } = server; + await server.close(); + await waitFor(() => client.isOpen && !client.isReady); + + expect(recover(new Error(READONLY_MESSAGE))).toBe(false); + expect(recover(new Error('ECONNRESET'))).toBe(false); + + server = await startRespServer(port); + await waitFor(() => client.isReady); + await sleep(50); + expect(server.connections).toBe(1); + }); + + it('does not tear down a client that reconnected on its own after a failed attempt', async () => { + const recover = createReadonlyRecovery({ client, minIntervalMs: 0 }); + const { port } = server; + server.readonly = true; + const error = await readonlyWriteError(client); + await server.close(); + expect(recover(error)).toBe(true); + await waitFor(() => !client.isOpen); + + server = await startRespServer(port); + await client.connect(); + expect( + recover(new Error('WRONGTYPE Operation against a key holding the wrong kind of value')), + ).toBe(false); + await sleep(50); + expect(server.connections).toBe(1); + expect(client.isReady).toBe(true); + }); +}); + +describe('standalone Keyv Redis client READONLY recovery', () => { + let originalEnv: NodeJS.ProcessEnv; + let server: RespServer; + let clients: typeof import('~/cache/redisClients'); + let cacheFactory: typeof import('~/cache/cacheFactory'); + const keyvClient = (): RedisClientType => clients.keyvRedisClient as RedisClientType; + + beforeAll(async () => { + originalEnv = { ...process.env }; + server = await startRespServer(); + process.env.USE_REDIS = 'true'; + process.env.USE_REDIS_CLUSTER = 'false'; + process.env.REDIS_URI = server.url; + process.env.REDIS_PING_INTERVAL = '0'; + process.env.REDIS_KEY_PREFIX = 'readonly-recovery'; + process.env.REDIS_READONLY_RECOVERY_INTERVAL = '100'; + process.env.REDIS_RETRY_MAX_ATTEMPTS = '2'; + process.env.REDIS_RETRY_MAX_DELAY = '50'; + jest.resetModules(); + clients = await import('~/cache/redisClients'); + cacheFactory = await import('~/cache/cacheFactory'); + await clients.keyvRedisClientReady; + }); + + afterAll(async () => { + await closeRedisClients(clients); + await server.close(); + process.env = originalEnv; + }); + + it('reconnects when a Keyv write is rejected with READONLY', async () => { + const connectionsBefore = server.connections; + const cache = cacheFactory.standardCache('readonly-recovery-test'); + server.readonly = true; + + await cache.set('key', 'value'); + await waitFor(() => server.connections === connectionsBefore + 1 && keyvClient().isReady); + + server.readonly = false; + await expect(cache.set('key', 'value')).resolves.toBe(true); + await expect(cache.get('key')).resolves.toBe('value'); + }); + + it('reconnects when a Lua script is rejected with READONLY', async () => { + await sleep(100); + const connectionsBefore = server.connections; + server.readonly = true; + + await expect( + clients.evalKeyvRedisScript('return 1', { keys: ['lock'], arguments: [] }), + ).rejects.toThrow(READONLY_MESSAGE); + await waitFor(() => server.connections === connectionsBefore + 1 && keyvClient().isReady); + + server.readonly = false; + await expect( + clients.evalKeyvRedisScript('return 1', { keys: ['lock'], arguments: [] }), + ).resolves.toBe(1); + }); + + it('reconnects when a namespace clear is rejected with READONLY', async () => { + await sleep(100); + const connectionsBefore = server.connections; + const cache = cacheFactory.standardCache('readonly-clear-test'); + await cache.set('key', 'value'); + server.readonly = true; + + await expect(cache.clear()).rejects.toThrow(READONLY_MESSAGE); + await waitFor(() => server.connections === connectionsBefore + 1 && keyvClient().isReady); + server.readonly = false; + }); + + it('routes errors handed to handleKeyvRedisError through the same recovery', async () => { + await sleep(100); + const connectionsBefore = server.connections; + expect(clients.handleKeyvRedisError(new Error(READONLY_MESSAGE))).toBe(true); + await waitFor(() => server.connections === connectionsBefore + 1 && keyvClient().isReady); + }); + + it('recovers through Keyv auto-connect after a failed reconnect without duplicate connections', async () => { + await sleep(100); + const cache = cacheFactory.standardCache('readonly-outage-test'); + const { port } = server; + server.readonly = true; + + await expect( + clients.evalKeyvRedisScript('return 1', { keys: ['lock'], arguments: [] }), + ).rejects.toThrow(READONLY_MESSAGE); + await server.close(); + await waitFor(() => !keyvClient().isOpen, 5000); + + await cache.set('key', 'value'); + await waitFor(() => !keyvClient().isOpen, 5000); + + server = await startRespServer(port); + await expect(cache.set('key', 'value')).resolves.toBe(true); + await waitFor(() => keyvClient().isReady); + expect(server.connections).toBe(1); + await expect(cache.get('key')).resolves.toBe('value'); + expect(server.connections).toBe(1); + }); +}); diff --git a/packages/api/src/cache/__tests__/resp.helper.ts b/packages/api/src/cache/__tests__/resp.helper.ts new file mode 100644 index 0000000000..4da642c173 --- /dev/null +++ b/packages/api/src/cache/__tests__/resp.helper.ts @@ -0,0 +1,167 @@ +import net from 'node:net'; +import type { AddressInfo, Socket } from 'node:net'; + +/** + * Minimal RESP2 server standing in for a Redis node whose role can be flipped + * between master and demoted replica. Real node-redis and ioredis clients speak + * to it over TCP, which is what a READONLY failover scenario needs: the socket + * stays open and healthy while every write is rejected. + */ +export type RespServer = { + url: string; + port: number; + /** Total sockets accepted since start; a reconnect shows up as a new one. */ + connections: number; + /** When true, write commands are rejected with the READONLY reply. */ + readonly: boolean; + commands: string[][]; + close(): Promise; +}; + +const READONLY_REPLY = "-READONLY You can't write against a read only replica.\r\n"; +const WRITE_COMMANDS = new Set(['SET', 'DEL', 'UNLINK', 'EVAL', 'EVALSHA', 'INCR', 'EXPIRE']); + +function encodeBulk(value: string | null): string { + return value == null ? '$-1\r\n' : `$${Buffer.byteLength(value)}\r\n${value}\r\n`; +} + +function globToRegExp(pattern: string): RegExp { + const escaped = pattern + .replace(/[.+^${}()|[\]\\]/g, '\\$&') + .replace(/\*/g, '.*') + .replace(/\?/g, '.'); + return new RegExp(`^${escaped}$`); +} + +function encodeScan(store: Map, args: string[]): string { + const matchIndex = args.findIndex((arg) => arg.toUpperCase() === 'MATCH'); + const matcher = matchIndex === -1 ? /^/ : globToRegExp(args[matchIndex + 1]); + const keys = [...store.keys()].filter((key) => matcher.test(key)); + return `*2\r\n$1\r\n0\r\n*${keys.length}\r\n${keys.map(encodeBulk).join('')}`; +} + +function parseFrames(buffer: Buffer): { frames: string[][]; rest: Buffer } { + const frames: string[][] = []; + let offset = 0; + while (offset < buffer.length) { + if (buffer[offset] !== 0x2a) { + break; + } + const headerEnd = buffer.indexOf('\r\n', offset); + if (headerEnd === -1) { + break; + } + const count = Number(buffer.subarray(offset + 1, headerEnd).toString()); + let cursor = headerEnd + 2; + const args: string[] = []; + let complete = true; + for (let i = 0; i < count; i++) { + const lengthEnd = buffer.indexOf('\r\n', cursor); + if (lengthEnd === -1 || buffer[cursor] !== 0x24) { + complete = false; + break; + } + const length = Number(buffer.subarray(cursor + 1, lengthEnd).toString()); + const valueStart = lengthEnd + 2; + const valueEnd = valueStart + length; + if (buffer.length < valueEnd + 2) { + complete = false; + break; + } + args.push(buffer.subarray(valueStart, valueEnd).toString()); + cursor = valueEnd + 2; + } + if (!complete) { + break; + } + frames.push(args); + offset = cursor; + } + return { frames, rest: buffer.subarray(offset) }; +} + +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 reply = (args: string[]): string => { + const command = args[0]?.toUpperCase() ?? ''; + if (state.readonly && WRITE_COMMANDS.has(command)) { + return READONLY_REPLY; + } + switch (command) { + case 'PING': + return '+PONG\r\n'; + case 'INFO': + return encodeBulk('# Server\r\nredis_version:7.2.4\r\nloading:0\r\n'); + case 'GET': + return encodeBulk(store.get(args[1]) ?? null); + case 'SET': + store.set(args[1], args[2]); + return '+OK\r\n'; + case 'DEL': + case 'UNLINK': + return `:${args.slice(1).filter((key) => store.delete(key)).length}\r\n`; + case 'EVAL': + case 'EVALSHA': + return ':1\r\n'; + case 'SCAN': + return encodeScan(store, args); + default: + return '+OK\r\n'; + } + }; + + const server = net.createServer((socket) => { + state.connections += 1; + sockets.add(socket); + let pending: Buffer = Buffer.alloc(0); + socket.on('data', (chunk) => { + const { frames, rest } = parseFrames(Buffer.concat([pending, chunk])); + pending = rest; + for (const frame of frames) { + state.commands.push(frame); + socket.write(reply(frame)); + } + }); + socket.on('close', () => sockets.delete(socket)); + socket.on('error', () => undefined); + }); + + await new Promise((resolve) => server.listen(port, '127.0.0.1', resolve)); + const { port: boundPort } = server.address() as AddressInfo; + + return { + url: `redis://127.0.0.1:${boundPort}`, + port: boundPort, + get connections() { + return state.connections; + }, + get readonly() { + return state.readonly; + }, + set readonly(value: boolean) { + state.readonly = value; + }, + commands: state.commands, + close: () => + new Promise((resolve) => { + for (const socket of sockets) { + socket.destroy(); + } + server.close(() => resolve()); + }), + }; +} + +/** Polls until `predicate` holds, failing after `timeoutMs`. */ +export async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) { + throw new Error('Timed out waiting for condition'); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} diff --git a/packages/api/src/cache/cacheConfig.ts b/packages/api/src/cache/cacheConfig.ts index 303fb968c8..619662c416 100644 --- a/packages/api/src/cache/cacheConfig.ts +++ b/packages/api/src/cache/cacheConfig.ts @@ -94,6 +94,8 @@ const cacheConfig: { REDIS_RETRY_MAX_ATTEMPTS: number; /** Connection timeout in ms */ REDIS_CONNECT_TIMEOUT: number; + /** Min spacing in ms between reconnects forced by READONLY replies after a failover */ + REDIS_READONLY_RECOVERY_INTERVAL: number; /** Queue commands when disconnected */ REDIS_ENABLE_OFFLINE_QUEUE: boolean; /** flag to modify redis connection by adding dnsLookup this is required when connecting to elasticache for ioredis @@ -171,6 +173,8 @@ const cacheConfig: { REDIS_RETRY_MAX_ATTEMPTS: math(process.env.REDIS_RETRY_MAX_ATTEMPTS, 10), /** Connection timeout in ms */ REDIS_CONNECT_TIMEOUT: math(process.env.REDIS_CONNECT_TIMEOUT, 10000), + /** Min spacing in ms between reconnects forced by READONLY replies after a failover */ + REDIS_READONLY_RECOVERY_INTERVAL: math(process.env.REDIS_READONLY_RECOVERY_INTERVAL, 5000), /** Queue commands when disconnected */ REDIS_ENABLE_OFFLINE_QUEUE: isEnabled(process.env.REDIS_ENABLE_OFFLINE_QUEUE ?? 'true'), /** flag to modify redis connection by adding dnsLookup this is required when connecting to elasticache for ioredis diff --git a/packages/api/src/cache/cacheFactory.ts b/packages/api/src/cache/cacheFactory.ts index 4e760058e8..5a45e2578c 100644 --- a/packages/api/src/cache/cacheFactory.ts +++ b/packages/api/src/cache/cacheFactory.ts @@ -14,7 +14,7 @@ import session, { MemoryStore } from 'express-session'; import { Time, CacheKeys } from 'librechat-data-provider'; import { RedisStore as ConnectRedis } from 'connect-redis'; import type { SendCommandFn } from 'rate-limit-redis'; -import { keyvRedisClient, ioredisClient } from './redisClients'; +import { keyvRedisClient, ioredisClient, handleKeyvRedisError } from './redisClients'; import { batchDeleteKeys, scanKeys } from './redisUtils'; import { instrumentIORedisClient, @@ -35,6 +35,34 @@ import { violationFile } from './keyvFiles'; */ const inMemoryCacheMap = new Map(); +/** + * Deletes every key under a namespace through the raw client, which is the one + * write path that bypasses the Keyv error funnel; READONLY rejections are routed + * to failover recovery before propagating. + */ +async function clearRedisNamespace(namespace: string): Promise { + if (!keyvRedisClient || !('scanIterator' in keyvRedisClient)) { + logger.warn(`Cannot clear namespace ${namespace}: Redis scanIterator not available`); + return; + } + + const pattern = cacheConfig.REDIS_KEY_PREFIX + ? `${cacheConfig.REDIS_KEY_PREFIX}${cacheConfig.GLOBAL_PREFIX_SEPARATOR}${namespace}:*` + : `${namespace}:*`; + + try { + const keysToDelete = await scanKeys(keyvRedisClient, pattern); + if (keysToDelete.length === 0) { + return; + } + await batchDeleteKeys(keyvRedisClient, keysToDelete); + logger.debug(`Cleared ${keysToDelete.length} keys from namespace ${namespace}`); + } catch (error) { + handleKeyvRedisError(error); + throw error; + } +} + /** * Creates a cache instance using Redis or a fallback store. Suitable for general caching needs. * @@ -57,33 +85,13 @@ export const standardCache = (namespace: string, ttl?: number, fallbackStore?: o cache.on('error', (err) => { logger.error(`Cache error in namespace ${namespace}:`, err); + handleKeyvRedisError(err); }); // Override clear() to handle namespace-aware deletion // The default Keyv clear() doesn't respect namespace due to the workaround above // Workaround for issue #10487 https://github.com/danny-avila/LibreChat/issues/10487 - cache.clear = async () => { - // Type-safe check for Redis client with scanIterator support - if (!keyvRedisClient || !('scanIterator' in keyvRedisClient)) { - logger.warn(`Cannot clear namespace ${namespace}: Redis scanIterator not available`); - return; - } - - // Build pattern: globalPrefix::namespace:* or namespace:* - const pattern = cacheConfig.REDIS_KEY_PREFIX - ? `${cacheConfig.REDIS_KEY_PREFIX}${cacheConfig.GLOBAL_PREFIX_SEPARATOR}${namespace}:*` - : `${namespace}:*`; - - // Use utility functions for efficient scan and parallel deletion - const keysToDelete = await scanKeys(keyvRedisClient, pattern); - - if (keysToDelete.length === 0) { - return; - } - - await batchDeleteKeys(keyvRedisClient, keysToDelete); - logger.debug(`Cleared ${keysToDelete.length} keys from namespace ${namespace}`); - }; + cache.clear = () => clearRedisNamespace(namespace); return instrumentRedisCache(cache, namespace); } catch (err) { diff --git a/packages/api/src/cache/recovery.ts b/packages/api/src/cache/recovery.ts new file mode 100644 index 0000000000..86a4d6a9ce --- /dev/null +++ b/packages/api/src/cache/recovery.ts @@ -0,0 +1,116 @@ +import { logger } from '@librechat/data-schemas'; + +/** Reply prefix Redis returns for writes sent to a demoted replica. */ +export const READONLY_ERROR_PREFIX = 'READONLY'; + +export type ReconnectableClient = { + readonly isOpen: boolean; + readonly isReady: boolean; + destroy(): void; + connect(): Promise; +}; + +export type ReadonlyRecoveryOptions = { + client: ReconnectableClient; + /** + * Minimum spacing between reconnect attempts. A failover produces READONLY + * replies by the hundred per second, and a reconnect issued before the + * topology settles can land on the demoted node again, so attempts are + * spaced out and retried on the next error instead of given up on. + */ + minIntervalMs: number; + /** Monotonic clock in milliseconds; defaults to `performance.now`. */ + now?: () => number; + /** Client label used in log lines. */ + label?: string; +}; + +/** Handler returning whether the error started a reconnect attempt. */ +export type ReadonlyRecoveryHandler = (error: unknown) => boolean; + +function errorMessage(error: unknown): string { + if (error instanceof Error) { + return error.message; + } + return typeof error === 'string' ? error : ''; +} + +/** Whether an error is a READONLY reply from a Redis node that was demoted to replica. */ +export function isReadonlyReplicaError(error: unknown): boolean { + return errorMessage(error).includes(READONLY_ERROR_PREFIX); +} + +/** + * Creates the READONLY recovery hook for a standalone node-redis client. + * + * node-redis only reconnects when its socket closes, and a demoted replica keeps + * existing sockets open while rejecting every write, so a READONLY reply never + * reaches `socket.reconnectStrategy`. The returned handler mirrors what ioredis + * does through `reconnectOnError`: it tears the socket down and reconnects, which + * re-resolves the connection to whatever the master address now points at. + * + * Tearing the socket down rejects the commands queued or in flight at that + * moment (reads included); node-redis cannot replay them the way ioredis does. + * That is a bounded, once-per-attempt cost, traded against every write failing + * until the process restarts. + * + * READONLY replies reject straight to the command's promise and never reach the + * client's error event, so the handler is meant to be called from every error + * funnel that sees command failures: the Keyv cache error event, Lua scripts + * evaluated directly against the client, and namespace clears. It debounces + * internally and never throws. + * + * The hook only acts on a client it can reason about: a READONLY reply on a + * ready socket, or a client its own failed reconnect left closed. While + * node-redis runs its own reconnect loop (open but not ready) the hook stays + * out of the way, since a second `connect()` would start a concurrent loop + * that leaks sockets into one shared reply decoder. + */ +export function createReadonlyRecovery(options: ReadonlyRecoveryOptions): ReadonlyRecoveryHandler { + const { client, minIntervalMs, label = '@keyv/redis' } = options; + const now = options.now ?? (() => performance.now()); + let inFlight = false; + let retryPending = false; + let lastAttemptAt = Number.NEGATIVE_INFINITY; + + const reconnect = async (): Promise => { + logger.warn(`${label} client reconnecting due to READONLY error`); + if (client.isOpen) { + client.destroy(); + } + await client.connect(); + logger.info(`${label} client reconnected after READONLY error`); + }; + + const shouldReconnect = (error: unknown): boolean => { + if (client.isReady) { + retryPending = false; + return isReadonlyReplicaError(error); + } + return !client.isOpen && retryPending; + }; + + return function recoverIfReadonly(error: unknown): boolean { + if (inFlight || !shouldReconnect(error)) { + return false; + } + const attemptAt = now(); + if (attemptAt - lastAttemptAt < minIntervalMs) { + return false; + } + inFlight = true; + lastAttemptAt = attemptAt; + void reconnect() + .then(() => { + retryPending = false; + }) + .catch((reconnectError: unknown) => { + retryPending = true; + logger.error(`${label} client reconnect after READONLY error failed:`, reconnectError); + }) + .finally(() => { + inFlight = false; + }); + return true; + }; +} diff --git a/packages/api/src/cache/redisClients.ts b/packages/api/src/cache/redisClients.ts index c6f6dcbef0..076d73d99e 100644 --- a/packages/api/src/cache/redisClients.ts +++ b/packages/api/src/cache/redisClients.ts @@ -5,6 +5,8 @@ import { createClient, createCluster } from '@keyv/redis'; import type { ScanOptions } from '@redis/client/dist/lib/commands/SCAN'; import type { RedisClientType, RedisClusterType } from '@redis/client'; import type { Redis, Cluster } from 'ioredis'; +import type { ReadonlyRecoveryHandler } from './recovery'; +import { createReadonlyRecovery, isReadonlyReplicaError } from './recovery'; import { cacheConfig } from './cacheConfig'; const urls = cacheConfig.REDIS_URI?.split(',').map((uri) => new URL(uri)) || []; @@ -62,8 +64,7 @@ if (cacheConfig.USE_REDIS) { return delay; }, reconnectOnError: (err: Error) => { - const targetError = 'READONLY'; - if (err.message.includes(targetError)) { + if (isReadonlyReplicaError(err)) { logger.warn('ioredis reconnecting due to READONLY error'); return 2; // Return retry delay instead of boolean } @@ -151,33 +152,59 @@ if (cacheConfig.USE_REDIS) { } let keyvRedisClient: RedisClientType | RedisClusterType | null = null; +let recoverKeyvRedisClient: ReadonlyRecoveryHandler | undefined; + +/** + * Routes a Keyv Redis client error through READONLY failover recovery. + * Cluster clients follow MOVED to the owning master on their own, so the hook + * only exists for the standalone client. Returns whether a reconnect was started. + */ +function handleKeyvRedisError(error: unknown): boolean { + return recoverKeyvRedisClient?.(error) ?? false; +} type RedisEvalOptions = { keys: string[]; arguments: string[] }; -/** - * Runs a Lua script on the master that owns its keys. Node Redis can execute a - * cluster EVAL through an arbitrary node while the slot map is settling, which - * leaks a MOVED reply instead of following it. Catalog scripts are deliberately - * single-slot, so selecting the owning master also makes that invariant explicit. - */ -async function evalKeyvRedisScript(script: string, options: RedisEvalOptions): Promise { - await waitForKeyvRedisClient(); - if (!keyvRedisClient) { - throw new Error('Keyv Redis client is not configured'); - } - if (!('masters' in keyvRedisClient) || options.keys.length === 0) { - return keyvRedisClient.eval(script, options); +async function runKeyvRedisScript( + client: RedisClientType | RedisClusterType, + script: string, + options: RedisEvalOptions, +): Promise { + if (!('masters' in client) || options.keys.length === 0) { + return client.eval(script, options); } const slot = calculateSlot(options.keys[0]); if (options.keys.some((key) => calculateSlot(key) !== slot)) { throw new Error('Redis catalog script keys must share one cluster slot'); } - const master = keyvRedisClient.getSlotMaster(slot); - const nodeClient = await keyvRedisClient.nodeClient(master); + const master = client.getSlotMaster(slot); + const nodeClient = await client.nodeClient(master); return nodeClient.eval(script, options); } +/** + * Runs a Lua script on the master that owns its keys. Node Redis can execute a + * cluster EVAL through an arbitrary node while the slot map is settling, which + * leaks a MOVED reply instead of following it. Catalog scripts are deliberately + * single-slot, so selecting the owning master also makes that invariant explicit. + * + * Script failures reject straight to the caller without passing through any + * client error event, so READONLY replies are routed to failover recovery here. + */ +async function evalKeyvRedisScript(script: string, options: RedisEvalOptions): Promise { + await waitForKeyvRedisClient(); + if (!keyvRedisClient) { + throw new Error('Keyv Redis client is not configured'); + } + try { + return await runKeyvRedisScript(keyvRedisClient, script, options); + } catch (error) { + handleKeyvRedisError(error); + throw error; + } +} + if (cacheConfig.USE_REDIS) { /** * ** WARNING ** Keyv Redis client does not support Prefix like ioredis above. @@ -238,6 +265,13 @@ if (cacheConfig.USE_REDIS) { keyvRedisClient.setMaxListeners(cacheConfig.REDIS_MAX_LISTENERS); + if (!isRedisCluster) { + recoverKeyvRedisClient = createReadonlyRecovery({ + client: keyvRedisClient as RedisClientType, + minIntervalMs: cacheConfig.REDIS_READONLY_RECOVERY_INTERVAL, + }); + } + keyvRedisClient.on('error', (err) => { logger.error('@keyv/redis client error:', err); }); @@ -272,4 +306,5 @@ export { keyvRedisClientReady, waitForKeyvRedisClient, evalKeyvRedisScript, + handleKeyvRedisError, };