diff --git a/lib/collections/link-reachability.ts b/lib/collections/link-reachability.ts index d7bc8985..82d833c8 100644 --- a/lib/collections/link-reachability.ts +++ b/lib/collections/link-reachability.ts @@ -109,7 +109,10 @@ export async function consumeProbeBudget( return { allowed: true, retryAfterMs: 0 }; } - const redis = await getReadyRedisClient(); + const redis = await getReadyRedisClient({ + caller: "library:link-reachability", + probes: amount, + }); if (!redis) { if (isRedisConfigured()) { log.warn("Link probe budget unavailable; refusing probes", { diff --git a/lib/common/redis.test.ts b/lib/common/redis.test.ts index 613dbe7f..43ccf1a6 100644 --- a/lib/common/redis.test.ts +++ b/lib/common/redis.test.ts @@ -1,4 +1,17 @@ -import { afterAll, describe, expect, mock, test } from "bun:test"; +import { + afterAll, + afterEach, + beforeEach, + describe, + expect, + mock, + spyOn, + test, +} from "bun:test"; +import { Logger } from "@/lib/common/logs/console/logger"; + +const REFUSAL_MESSAGE = + "Redis not ready after waiting for the connect; caller fails closed"; type RedisHandler = (...args: unknown[]) => void; @@ -31,14 +44,53 @@ const redis: typeof import("./redis") = await import( `${import.meta.dir}/redis.ts?isolation` ); +// A second private copy starts with no client, so its connect stays pending +// and the ready wait runs to its bound. +const redisWithPendingConnect: typeof import("./redis") = await import( + `${import.meta.dir}/redis.ts?isolation-pending` +); + +// The test environment drops every log record, so capture the calls the +// module makes to the logger instead of the output. +interface WarnCall { + message: string; + meta: unknown; +} + +const warnCalls: WarnCall[] = []; +let restoreWarnSpy: () => void; + +beforeEach(() => { + warnCalls.length = 0; + const spy = spyOn(Logger.prototype, "warn").mockImplementation( + (message, meta) => { + warnCalls.push({ message, meta }); + } + ); + restoreWarnSpy = () => spy.mockRestore(); +}); + +afterEach(() => { + restoreWarnSpy(); +}); + afterAll(() => { mock.restore(); delete process.env.REDIS_URL; }); +function refusalCalls(): unknown[] { + return warnCalls + .filter((call) => call.message === REFUSAL_MESSAGE) + .map((call) => call.meta); +} + describe("getReadyRedisClient", () => { test("waits for the cold-start connect instead of reporting an outage", async () => { - const pending = redis.getReadyRedisClient(); + const pending = redis.getReadyRedisClient({ + bucket: "read", + caller: "mcp.rate-limit", + }); let didSettle = false; pending.then(() => { @@ -52,11 +104,64 @@ describe("getReadyRedisClient", () => { resolveInitialConnect?.(); expect((await pending)?.isReady).toBe(true); + expect(warnCalls).toHaveLength(0); }); test("reports unavailable once an established connection is lost", async () => { redisClient.isReady = false; - expect(await redis.getReadyRedisClient()).toBeNull(); + expect( + await redis.getReadyRedisClient({ + bucket: "write", + caller: "mcp.rate-limit", + }) + ).toBeNull(); + }); + + test("logs the wait outcome when it refuses a caller", async () => { + // Restore the ready state so the once-per-episode gate reopens. + redisClient.isReady = true; + handlers.get("ready")?.(); + redisClient.isReady = false; + + expect( + await redis.getReadyRedisClient({ + bucket: "write", + caller: "mcp.rate-limit", + }) + ).toBeNull(); + + const refusals = refusalCalls(); + expect(refusals).toHaveLength(1); + expect(refusals[0]).toEqual({ + bucket: "write", + caller: "mcp.rate-limit", + hasRedisConnected: true, + waitDurationMs: expect.any(Number), + waitTimedOut: false, + }); + }); + + test("records the expired wait bound when the connect never settles", async () => { + const startedAt = Date.now(); + + expect( + await redisWithPendingConnect.getReadyRedisClient({ + caller: "library:link-reachability", + probes: 12, + }) + ).toBeNull(); + + const waitedMs = Date.now() - startedAt; + const refusals = refusalCalls(); + expect(refusals).toHaveLength(1); + expect(refusals[0]).toEqual({ + caller: "library:link-reachability", + hasRedisConnected: false, + probes: 12, + waitDurationMs: expect.any(Number), + waitTimedOut: true, + }); + expect(waitedMs).toBeGreaterThanOrEqual(1000); }); }); diff --git a/lib/common/redis.ts b/lib/common/redis.ts index 490332ec..a7bd119f 100644 --- a/lib/common/redis.ts +++ b/lib/common/redis.ts @@ -15,6 +15,24 @@ const RedisConnectionError = NamedError.create( ); export type RedisConnectionError = InstanceType; +/** + * Identifies the caller waiting for the connect, so a refusal is attributable + * during an incident. The count the caller was about to spend (the MCP bucket + * or the probe batch) sizes the blast radius. + */ +export type RedisReadyCaller = + | { readonly bucket: string; readonly caller: "mcp.rate-limit" } + | { readonly caller: "library:link-reachability"; readonly probes: number }; + +/** + * Outcome of {@link waitForRedisReady}: how long the connect stayed pending, + * and whether the wait bound expired before it settled. + */ +interface RedisReadyWait { + readonly durationMs: number; + readonly timedOut: boolean; +} + /** * How long an abuse-bounding caller waits for an in-flight connection before * it treats a configured Redis as unavailable. @@ -24,6 +42,7 @@ const REDIS_READY_WAIT_TIMEOUT_MS = 1000; let globalRedisClient: RedisClientType | null = null; let redisConnectPromise: Promise | null = null; let didWarnRedisUnavailable = false; +let didWarnRedisReadyRefusal = false; let hasRedisConnected = false; /** @@ -47,30 +66,56 @@ function warnRedisUnavailableOnce(): void { ); } +/** + * Report, once per outage episode, that a caller waited for the connect and + * was refused. Records the wait outcome so an operator can tell a cold start + * that outran the bound from a Redis instance that never connected, and can + * attribute the refusal to the caller it hit. + */ +function warnRedisReadyRefusalOnce( + caller: RedisReadyCaller, + wait: RedisReadyWait +): void { + if (didWarnRedisReadyRefusal) { + return; + } + didWarnRedisReadyRefusal = true; + log.warn( + "Redis not ready after waiting for the connect; caller fails closed", + { + ...caller, + hasRedisConnected, + waitDurationMs: wait.durationMs, + waitTimedOut: wait.timedOut, + } + ); +} + /** * Wait for the connect started by {@link getRedisClient} to settle, up to a * bound. A client that is still connecting on a cold start resolves here in * milliseconds; a client whose socket cannot connect keeps its connect * pending across reconnects, so the timeout keeps the caller from hanging. + * + * Reports how long the connect stayed pending and whether the bound expired, + * so the caller can attribute and classify the refusal. */ -function waitForRedisReady(timeoutMs: number): Promise { +function waitForRedisReady(timeoutMs: number): Promise { const connecting = redisConnectPromise; if (!connecting) { - return Promise.resolve(); + return Promise.resolve({ durationMs: 0, timedOut: false }); } - return new Promise((resolve) => { - const timer = setTimeout(() => resolve(), timeoutMs); - connecting.then( - () => { - clearTimeout(timer); - resolve(); - }, - () => { - clearTimeout(timer); - resolve(); - } - ); + const startedAt = Date.now(); + return new Promise((resolve) => { + const timer = setTimeout(() => { + resolve({ durationMs: Date.now() - startedAt, timedOut: true }); + }, timeoutMs); + const settle = () => { + clearTimeout(timer); + resolve({ durationMs: Date.now() - startedAt, timedOut: false }); + }; + connecting.then(settle, settle); }); } @@ -135,6 +180,7 @@ export function getRedisClient(): RedisClientType | null { globalRedisClient.on("ready", () => { hasRedisConnected = true; didWarnRedisUnavailable = false; + didWarnRedisReadyRefusal = false; log.info("Redis connection established"); }); @@ -172,7 +218,9 @@ export function getRedisClient(): RedisClientType | null { * genuinely down (or reconnecting) still reports null after the bounded wait. * A deployment with no `REDIS_URL` returns null immediately. */ -export async function getReadyRedisClient(): Promise { +export async function getReadyRedisClient( + caller: RedisReadyCaller +): Promise { const client = getRedisClient(); if (client) { return client; @@ -181,11 +229,11 @@ export async function getReadyRedisClient(): Promise { return null; } - await waitForRedisReady(REDIS_READY_WAIT_TIMEOUT_MS); + const wait = await waitForRedisReady(REDIS_READY_WAIT_TIMEOUT_MS); const readyClient = getRedisClient(); if (!readyClient) { - warnRedisUnavailableOnce(); + warnRedisReadyRefusalOnce(caller, wait); } return readyClient; } diff --git a/lib/integrations/mcp/rate-limit.ts b/lib/integrations/mcp/rate-limit.ts index f1e4bae4..7c7d1264 100644 --- a/lib/integrations/mcp/rate-limit.ts +++ b/lib/integrations/mcp/rate-limit.ts @@ -61,7 +61,10 @@ export async function checkMcpRateLimit( userId: string, bucket: Bucket ): Promise { - const redis = await getReadyRedisClient(); + const redis = await getReadyRedisClient({ + bucket: bucket.name, + caller: "mcp.rate-limit", + }); if (!redis) { return isRedisConfigured() ? { status: "unavailable" }