mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-23 10:36:09 +00:00
🚢 fix: Reconnect Keyv Redis After Replica Failover (#15502)
* 🔁 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 <noreply@anthropic.com> 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 <noreply@anthropic.com> 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
This commit is contained in:
parent
b1c56239f9
commit
519e64769a
8 changed files with 618 additions and 40 deletions
|
|
@ -5,6 +5,7 @@ jest.mock('@keyv/redis', () => ({
|
|||
}));
|
||||
|
||||
jest.mock('../../redisClients', () => ({
|
||||
handleKeyvRedisError: jest.fn(),
|
||||
keyvRedisClient: null,
|
||||
ioredisClient: null,
|
||||
}));
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ jest.mock('@keyv/redis', () => ({
|
|||
const mockKeyvRedisClient = { scanIterator: jest.fn() };
|
||||
|
||||
jest.mock('../../redisClients', () => ({
|
||||
handleKeyvRedisError: jest.fn(),
|
||||
keyvRedisClient: mockKeyvRedisClient,
|
||||
ioredisClient: null,
|
||||
}));
|
||||
|
|
|
|||
246
packages/api/src/cache/__tests__/recovery.spec.ts
vendored
Normal file
246
packages/api/src/cache/__tests__/recovery.spec.ts
vendored
Normal file
|
|
@ -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<void> => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
async function readonlyWriteError(client: RedisClientType): Promise<unknown> {
|
||||
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<void> => {
|
||||
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);
|
||||
});
|
||||
});
|
||||
167
packages/api/src/cache/__tests__/resp.helper.ts
vendored
Normal file
167
packages/api/src/cache/__tests__/resp.helper.ts
vendored
Normal file
|
|
@ -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<void>;
|
||||
};
|
||||
|
||||
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<string, string>, 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<RespServer> {
|
||||
const store = new Map<string, string>();
|
||||
const sockets = new Set<Socket>();
|
||||
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<void>((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<void>((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<void> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
4
packages/api/src/cache/cacheConfig.ts
vendored
4
packages/api/src/cache/cacheConfig.ts
vendored
|
|
@ -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
|
||||
|
|
|
|||
54
packages/api/src/cache/cacheFactory.ts
vendored
54
packages/api/src/cache/cacheFactory.ts
vendored
|
|
@ -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<string, Keyv>();
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
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) {
|
||||
|
|
|
|||
116
packages/api/src/cache/recovery.ts
vendored
Normal file
116
packages/api/src/cache/recovery.ts
vendored
Normal file
|
|
@ -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<unknown>;
|
||||
};
|
||||
|
||||
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<void> => {
|
||||
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;
|
||||
};
|
||||
}
|
||||
69
packages/api/src/cache/redisClients.ts
vendored
69
packages/api/src/cache/redisClients.ts
vendored
|
|
@ -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<unknown> {
|
||||
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<unknown> {
|
||||
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<unknown> {
|
||||
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,
|
||||
};
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue