From 1bd4455c2d92b29fdd499a54a421e6afeacbc6c9 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 9 Aug 2026 06:59:29 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=AD=20fix:=20Make=20MCP=20Catalog=20Re?= =?UTF-8?q?dis=20Cluster-Safe=20(#14717)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: make MCP catalog Redis startup cluster-safe * fix: stabilize Redis readiness gate * fix: type Redis readiness export * style: apply canonical import order --- api/server/index.js | 2 + api/server/index.spec.js | 10 ++++ .../__tests__/getCachedTools.lock.spec.js | 45 ++++++++++++++ api/server/services/Config/getCachedTools.js | 5 +- package-lock.json | 4 +- packages/api/package.json | 3 +- .../redisClients.cache_integration.spec.ts | 20 +++++++ packages/api/src/cache/redisClients.ts | 59 ++++++++++++++++--- packages/api/src/mcp/catalog/store.ts | 36 +++++++---- 9 files changed, 160 insertions(+), 24 deletions(-) diff --git a/api/server/index.js b/api/server/index.js index dc9c456736..028ec7a824 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -37,6 +37,7 @@ const { updateInterfacePermissions, configureMessageFilterRegexValidator, configureFileConfigRegexEngine, + waitForKeyvRedisClient, } = require('@librechat/api'); const { connectDb, indexSync } = require('~/db'); const { @@ -115,6 +116,7 @@ const configureGenerationStreams = () => { }; const startServer = async () => { + await waitForKeyvRedisClient(); const { metricsMiddleware, metricsRouter } = createMetrics(); if (!process.env.METRICS_SECRET) { logger.warn('[metrics] METRICS_SECRET is not set - /metrics will return 401 for all requests'); diff --git a/api/server/index.spec.js b/api/server/index.spec.js index 6c5b3c20cd..bdc249bcc7 100644 --- a/api/server/index.spec.js +++ b/api/server/index.spec.js @@ -110,6 +110,16 @@ describe('Telemetry wiring', () => { describe('Startup readiness wiring', () => { const source = fs.readFileSync(path.join(__dirname, 'index.js'), 'utf8'); + it('awaits the shared Redis client before startup cache access', () => { + const redisReadyIndex = source.indexOf('await waitForKeyvRedisClient();'); + const connectDbIndex = source.indexOf('await connectDb();'); + const appConfigIndex = source.indexOf('await getAppConfig({ baseOnly: true });'); + + expect(redisReadyIndex).toBeGreaterThan(-1); + expect(connectDbIndex).toBeGreaterThan(redisReadyIndex); + expect(appConfigIndex).toBeGreaterThan(redisReadyIndex); + }); + it('configures generation streams before the server accepts requests', () => { const streamConfigIndex = source.indexOf('configureGenerationStreams();'); const listenIndex = source.indexOf('const server = app.listen'); diff --git a/api/server/services/Config/__tests__/getCachedTools.lock.spec.js b/api/server/services/Config/__tests__/getCachedTools.lock.spec.js index 9417bbfcba..6302d20cad 100644 --- a/api/server/services/Config/__tests__/getCachedTools.lock.spec.js +++ b/api/server/services/Config/__tests__/getCachedTools.lock.spec.js @@ -8,13 +8,20 @@ const mockRedisClient = { const mockKeyvRedisClient = { eval: jest.fn(), }; +let mockRedisReadyPromise = Promise.resolve(); +const mockRedisReady = { + then: (...args) => mockRedisReadyPromise.then(...args), +}; +const mockWaitForRedis = jest.fn(() => mockRedisReady); jest.mock('@librechat/api', () => ({ ...jest.requireActual('@librechat/api'), cacheConfig: { FORCED_IN_MEMORY_CACHE_NAMESPACES: [] }, + evalKeyvRedisScript: (...args) => mockKeyvRedisClient.eval(...args), mcpConfig: { USER_CONNECTION_IDLE_TIMEOUT: 15 * 60 * 1000 }, ioredisClient: mockRedisClient, keyvRedisClient: mockKeyvRedisClient, + waitForKeyvRedisClient: mockWaitForRedis, })); jest.mock('@librechat/data-schemas', () => ({ @@ -47,6 +54,7 @@ describe('global tool cache write lock', () => { beforeEach(() => { jest.clearAllMocks(); + mockRedisReadyPromise = Promise.resolve(); mockRedisClient.set.mockResolvedValue('OK'); mockRedisClient.eval.mockResolvedValue(1); mockKeyvRedisClient.eval.mockResolvedValue(1); @@ -54,6 +62,43 @@ describe('global tool cache write lock', () => { mockCache.delete.mockResolvedValue(true); }); + it('waits for the shared Redis client before accessing the catalog', async () => { + mockCache.get.mockResolvedValue(null); + let resolveRedisReady; + mockRedisReadyPromise = new Promise((resolve) => { + resolveRedisReady = resolve; + }); + + const update = updateCachedGlobalTools(() => ({ builtin: {} })); + await Promise.resolve(); + + expect(mockRedisClient.set).not.toHaveBeenCalled(); + expect(mockCache.get).not.toHaveBeenCalled(); + + resolveRedisReady(); + await expect(update).resolves.toBeUndefined(); + + expect(mockRedisClient.set).toHaveBeenCalledTimes(1); + expect(mockCache.get).toHaveBeenCalledWith('tools:global'); + }); + + it('waits for the shared Redis client before reading the catalog', async () => { + mockCache.get.mockResolvedValue({ builtin: {} }); + let resolveRedisReady; + mockRedisReadyPromise = new Promise((resolve) => { + resolveRedisReady = resolve; + }); + + const read = getCachedTools(); + await Promise.resolve(); + + expect(mockCache.get).not.toHaveBeenCalled(); + + resolveRedisReady(); + await expect(read).resolves.toEqual({ builtin: {} }); + expect(mockCache.get).toHaveBeenCalledWith('tools:global'); + }); + it('acquires and safely releases the Redis lock around an aggregate update', async () => { const operation = jest.fn().mockResolvedValue('updated'); diff --git a/api/server/services/Config/getCachedTools.js b/api/server/services/Config/getCachedTools.js index 0dc0da0897..dc929b085e 100644 --- a/api/server/services/Config/getCachedTools.js +++ b/api/server/services/Config/getCachedTools.js @@ -1,8 +1,10 @@ const { CacheKeys } = require('librechat-data-provider'); const { cacheConfig, + evalKeyvRedisScript, ioredisClient, keyvRedisClient, + waitForKeyvRedisClient, mcpConfig, ToolCacheKeys, createMCPCatalogStore, @@ -12,7 +14,8 @@ const getLogStores = require('~/cache/getLogStores'); const store = createMCPCatalogStore({ cacheConfig, ioredisClient, - keyvRedisClient, + keyvRedisClient: keyvRedisClient ? { eval: evalKeyvRedisScript } : null, + waitForRedis: waitForKeyvRedisClient, userConnectionIdleTimeout: mcpConfig.USER_CONNECTION_IDLE_TIMEOUT, getCache: () => getLogStores(CacheKeys.TOOL_CACHE), }); diff --git a/package-lock.json b/package-lock.json index 179d3fde49..a2c52cc939 100644 --- a/package-lock.json +++ b/package-lock.json @@ -22751,6 +22751,7 @@ "version": "1.1.2", "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz", "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==", + "license": "Apache-2.0", "engines": { "node": ">=0.10.0" } @@ -42654,7 +42655,8 @@ "license": "ISC", "dependencies": { "@langchain/langgraph-checkpoint": "^1.1.2", - "@langchain/langgraph-checkpoint-mongodb": "^1.4.0" + "@langchain/langgraph-checkpoint-mongodb": "^1.4.0", + "cluster-key-slot": "^1.1.2" }, "devDependencies": { "@babel/preset-env": "^7.29.5", diff --git a/packages/api/package.json b/packages/api/package.json index b6f683748e..ad8d7f6398 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -167,6 +167,7 @@ }, "dependencies": { "@langchain/langgraph-checkpoint": "^1.1.2", - "@langchain/langgraph-checkpoint-mongodb": "^1.4.0" + "@langchain/langgraph-checkpoint-mongodb": "^1.4.0", + "cluster-key-slot": "^1.1.2" } } diff --git a/packages/api/src/cache/__tests__/redisClients.cache_integration.spec.ts b/packages/api/src/cache/__tests__/redisClients.cache_integration.spec.ts index b3f1288117..56dad1ef35 100644 --- a/packages/api/src/cache/__tests__/redisClients.cache_integration.spec.ts +++ b/packages/api/src/cache/__tests__/redisClients.cache_integration.spec.ts @@ -145,6 +145,26 @@ describe('redisClients Integration Tests', () => { clients.keyvRedisClientReady!.then(() => undefined), ); }); + + test('should execute same-slot catalog scripts on the owning master', async () => { + process.env.USE_REDIS_CLUSTER = 'true'; + process.env.REDIS_URI = + 'redis://127.0.0.1:7001,redis://127.0.0.1:7002,redis://127.0.0.1:7003'; + + const clients = await import('../redisClients'); + keyvRedisClient = clients.keyvRedisClient; + const hashTag = `catalog-eval-${Date.now()}`; + const keys = [`catalog:revision:{${hashTag}}`, `catalog:tools:{${hashTag}}`]; + + await expect( + clients.evalKeyvRedisScript( + "redis.call('SET', KEYS[1], ARGV[1]); redis.call('SET', KEYS[2], ARGV[1]); return 1", + { keys, arguments: ['published'] }, + ), + ).resolves.toBe(1); + await expect(keyvRedisClient!.mGet(keys)).resolves.toEqual(['published', 'published']); + await keyvRedisClient!.del(keys); + }); }); }); }); diff --git a/packages/api/src/cache/redisClients.ts b/packages/api/src/cache/redisClients.ts index eef63689f5..253bec3a8d 100644 --- a/packages/api/src/cache/redisClients.ts +++ b/packages/api/src/cache/redisClients.ts @@ -1,4 +1,5 @@ import IoRedis from 'ioredis'; +import calculateSlot from 'cluster-key-slot'; import { logger } from '@librechat/data-schemas'; import { createClient, createCluster } from '@keyv/redis'; import type { ScanCommandOptions } from '@redis/client/dist/lib/commands/SCAN'; @@ -11,6 +12,20 @@ const username = urls?.[0]?.username || cacheConfig.REDIS_USERNAME; const password = urls?.[0]?.password || cacheConfig.REDIS_PASSWORD; const ca = cacheConfig.REDIS_CA; +let resolveKeyvRedisClientReady: (() => void) | undefined; +let rejectKeyvRedisClientReady: ((reason?: unknown) => void) | undefined; +const keyvRedisClientReady: Promise | null = cacheConfig.USE_REDIS + ? new Promise((resolve, reject) => { + resolveKeyvRedisClientReady = resolve; + rejectKeyvRedisClientReady = reject; + }) + : null; + +/** Waits for the stable shared Keyv Redis readiness gate. */ +async function waitForKeyvRedisClient(): Promise { + await keyvRedisClientReady; +} + let ioredisClient: Redis | Cluster | null = null; if (cacheConfig.USE_REDIS) { const redisOptions: Record = { @@ -126,10 +141,32 @@ if (cacheConfig.USE_REDIS) { } let keyvRedisClient: RedisClientType | RedisClusterType | null = null; -let keyvRedisClientReady: - | Promise - | Promise, Record, Record>> - | null = null; + +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); + } + + 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); + return nodeClient.eval(script, options); +} if (cacheConfig.USE_REDIS) { /** @@ -212,12 +249,18 @@ if (cacheConfig.USE_REDIS) { logger.warn('@keyv/redis client disconnected'); }); - // Start connection immediately - keyvRedisClientReady = keyvRedisClient.connect(); + // Start connection immediately and settle the gate created before client initialization. + void keyvRedisClient.connect().then(resolveKeyvRedisClientReady, rejectKeyvRedisClientReady); - keyvRedisClientReady.catch((err): void => { + void keyvRedisClientReady?.catch((err): void => { logger.error('@keyv/redis initial connection failed:', err); }); } -export { ioredisClient, keyvRedisClient, keyvRedisClientReady }; +export { + ioredisClient, + keyvRedisClient, + keyvRedisClientReady, + waitForKeyvRedisClient, + evalKeyvRedisScript, +}; diff --git a/packages/api/src/mcp/catalog/store.ts b/packages/api/src/mcp/catalog/store.ts index 0ce949bbd9..96cae3c75c 100644 --- a/packages/api/src/mcp/catalog/store.ts +++ b/packages/api/src/mcp/catalog/store.ts @@ -154,6 +154,7 @@ export interface CatalogStoreDeps { }; ioredisClient?: LockRedisClient | null; keyvRedisClient?: KeyvRedisClient | null; + waitForRedis?: () => Promise; userConnectionIdleTimeout?: number | string; } @@ -288,6 +289,13 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { deps.keyvRedisClient != null && !deps.cacheConfig.FORCED_IN_MEMORY_CACHE_NAMESPACES?.includes(CacheKeys.TOOL_CACHE); + const getReadyCache = async (): Promise => { + if (sharedRedis()) { + await deps.waitForRedis?.(); + } + return deps.getCache(); + }; + const rawKey = (key: string): string => { const namespaced = `${CacheKeys.TOOL_CACHE}:${key}`; return deps.cacheConfig.REDIS_KEY_PREFIX @@ -342,6 +350,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { if (!sharedRedis()) { return operation(); } + await deps.waitForRedis?.(); const redis = deps.ioredisClient!; const token = randomUUID(); const deadline = Date.now() + ttl + LOCK_RETRY_MS; @@ -515,7 +524,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { async function getCachedTools( options: CachedToolsOptions = {}, ): Promise { - const cache = deps.getCache(); + const cache = await getReadyCache(); const { userId, serverName, configGeneration } = options; if (!userId || !serverName) { const global = await cache.get(ToolCacheKeys.GLOBAL); @@ -593,7 +602,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { update: (tools: LCAvailableTools) => LCAvailableTools, ): Promise { await runWithGlobalCacheLock(async () => { - const cache = deps.getCache(); + const cache = await getReadyCache(); const current = await cache.get(ToolCacheKeys.GLOBAL); const currentTools = isTools(current) ? current : {}; const next = update(currentTools); @@ -610,7 +619,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { tools: LCAvailableTools, options: CachedToolsOptions = {}, ): Promise { - const cache = deps.getCache(); + const cache = await getReadyCache(); const ttl = options.ttl ?? Time.TWELVE_HOURS; if (options.userId && options.serverName) { return ( @@ -640,7 +649,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { userId: string; serverName: string; }): Promise { - const cache = deps.getCache(); + const cache = await getReadyCache(); const key = ToolCacheKeys.MCP_SERVER_GENERATION(scope.userId, scope.serverName); const existing = await cache.get(key); if (typeof existing === 'string' && existing.length > 0) return existing; @@ -690,7 +699,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { serverName: string; publicationGeneration: string; }): Promise { - const cache = deps.getCache(); + const cache = await getReadyCache(); return withUserQueue(scope.userId, scope.serverName, () => renewIfCurrent( cache, @@ -704,7 +713,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { tools: LCAvailableTools, options: GuardedToolsOptions, ): Promise { - const cache = deps.getCache(); + const cache = await getReadyCache(); return withUserQueue(options.userId, options.serverName, async () => { const generationKey = ToolCacheKeys.MCP_SERVER_GENERATION(options.userId, options.serverName); const guarded: GuardedEntry = { @@ -749,9 +758,8 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { serverName: string, configGeneration: string, ): Promise { - const value = await deps - .getCache() - .get(ToolCacheKeys.MCP_APP_SERVER(serverName, configGeneration)); + const cache = await getReadyCache(); + const value = await cache.get(ToolCacheKeys.MCP_APP_SERVER(serverName, configGeneration)); if (isAppToolsEntry(value)) { return value.tools; } @@ -764,7 +772,8 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { ): Promise { const toolsKey = ToolCacheKeys.MCP_APP_SERVER(serverName, configGeneration); const scope = JSON.stringify([serverName, configGeneration]); - const cached = await deps.getCache().get(toolsKey); + const cache = await getReadyCache(); + const cached = await cache.get(toolsKey); const cachedRevision = isAppToolsEntry(cached) ? parseAppToolsRevision(cached.publicationRevision) : 0; @@ -798,9 +807,10 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { tools, }; const scope = JSON.stringify([serverName, configGeneration]); + const cache = await getReadyCache(); if (!sharedRedis()) { return withAppWriteQueue(scope, async () => { - const cached = await deps.getCache().get(toolsKey); + const cached = await cache.get(toolsKey); const cachedRevision = isAppToolsEntry(cached) ? parseAppToolsRevision(cached.publicationRevision) : 0; @@ -808,7 +818,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { if (currentRevision > nextRevision) { return false; } - if ((await deps.getCache().set(toolsKey, entry, ttl)) === false) { + if ((await cache.set(toolsKey, entry, ttl)) === false) { throw new Error('App tool cache rejected the write'); } rememberAppRevision(appCommittedRevisions, scope, nextRevision); @@ -839,7 +849,7 @@ export function createMCPCatalogStore(deps: CatalogStoreDeps): MCPCatalogStore { invalidateGlobal?: boolean; } = {}, ): Promise { - const cache = deps.getCache(); + const cache = await getReadyCache(); if (options.invalidateGlobal) { await runWithGlobalCacheLock(() => deleteGlobalWithinLock(cache)); }