diff --git a/apps/sim/lib/billing/calculations/usage-reservation.test.ts b/apps/sim/lib/billing/calculations/usage-reservation.test.ts index daaefbd127d..ca71b7cd1e4 100644 --- a/apps/sim/lib/billing/calculations/usage-reservation.test.ts +++ b/apps/sim/lib/billing/calculations/usage-reservation.test.ts @@ -339,6 +339,13 @@ describe('usage-reservation', () => { }) describe('refreshExecutionSlotExpiry', () => { + it('rethrows the original error object rather than the diagnostic wrapper', async () => { + const original = Object.assign(new Error('Command timed out'), { code: 'ETIMEDOUT' }) + getMock.mockRejectedValueOnce(original) + + await expect(refreshExecutionSlotExpiry('exec-1', Date.now() + 60_000)).rejects.toBe(original) + }) + it('refreshes only the locally owned slot and matching pointer', async () => { evalMock.mockResolvedValueOnce(1).mockResolvedValueOnce(1) await reserveExecutionSlot(memberParams) diff --git a/apps/sim/lib/billing/calculations/usage-reservation.ts b/apps/sim/lib/billing/calculations/usage-reservation.ts index aa7caf3311b..89f9df98edf 100644 --- a/apps/sim/lib/billing/calculations/usage-reservation.ts +++ b/apps/sim/lib/billing/calculations/usage-reservation.ts @@ -9,7 +9,7 @@ import { type ReservationDenialReason, } from '@/lib/core/admission/transient-failure' import { isBillingEnabled, isHosted } from '@/lib/core/config/env-flags' -import { getRedisClient } from '@/lib/core/config/redis' +import { describeRedisConnection, getRedisClient } from '@/lib/core/config/redis' import { getExecutionReservationTtlMs } from '@/lib/core/execution-limits' const logger = createLogger('UsageReservation') @@ -425,6 +425,34 @@ export type ReserveExecutionSlotResult = reason: ReservationDenialReason } +/** + * Records connection state alongside a failed slot operation. + * + * These three functions are the first Redis calls a queued workflow makes, so + * when the connection is not usable they are where it surfaces — as an + * `Error: Command timed out` carrying no app frame and no indication of which + * of several very different causes applied. Pairing the failure with + * `describeRedisConnection()` is what makes the next occurrence self-diagnosing + * instead of another inference from timing alone. + */ +async function withReservationDiagnostics( + operation: string, + reservationId: string, + run: () => Promise +): Promise { + try { + return await run() + } catch (error) { + logger.error('Usage reservation Redis operation failed', { + operation, + reservationId, + error: toError(error).message, + redis: describeRedisConnection(), + }) + throw error + } +} + /** * Atomic admission reservation that closes the usage-cap check-then-use race. * @@ -510,6 +538,7 @@ export async function reserveExecutionSlot( error: toError(error).message, entityKey, reservationId, + redis: describeRedisConnection(), }) throw new UsageReservationUnavailableError( 'Usage admission is temporarily unavailable. Please retry.', @@ -622,7 +651,11 @@ export async function refreshExecutionSlotExpiry( const boundedReservationId = requireBoundedIdentifier(reservationId, 'reservation id') const pointerKey = `${POINTER_KEY_PREFIX}${boundedReservationId}` - const descriptorValue = await redis.get(pointerKey) + const descriptorValue = await withReservationDiagnostics( + 'refresh:read-pointer', + boundedReservationId, + () => redis.get(pointerKey) + ) if (!descriptorValue) return false const descriptor = parseDescriptor(descriptorValue) if (!descriptor) { @@ -633,22 +666,25 @@ export async function refreshExecutionSlotExpiry( const expiryAt = Math.min(expiresAt, now + getExecutionReservationTtlMs()) const keys = buildLocalKeys(descriptor, boundedReservationId) const keyArgs = localKeyArguments(keys) - const localResult = await redis.eval( - REFRESH_LOCAL_SCRIPT, - keyArgs.length, - ...keyArgs, + const localResult = await withReservationDiagnostics( + 'refresh:extend-local', boundedReservationId, - descriptorValue, - expiryAt.toString() + () => + redis.eval( + REFRESH_LOCAL_SCRIPT, + keyArgs.length, + ...keyArgs, + boundedReservationId, + descriptorValue, + expiryAt.toString() + ) ) if (localResult !== 1) return false - const pointerResult = await redis.eval( - REFRESH_POINTER_SCRIPT, - 1, - pointerKey, - descriptorValue, - expiryAt.toString() + const pointerResult = await withReservationDiagnostics( + 'refresh:extend-pointer', + boundedReservationId, + () => redis.eval(REFRESH_POINTER_SCRIPT, 1, pointerKey, descriptorValue, expiryAt.toString()) ) if (pointerResult !== 1) { throw new UsageReservationUnavailableError( diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts index 96b0c49e4b6..61c33523c49 100644 --- a/apps/sim/lib/core/config/redis.test.ts +++ b/apps/sim/lib/core/config/redis.test.ts @@ -27,6 +27,7 @@ vi.mock('ioredis', () => ({ import { acquireLock, closeRedisConnection, + describeRedisConnection, extendLock, getRedisClient, onRedisReconnect, @@ -38,6 +39,7 @@ describe('redis config', () => { vi.clearAllMocks() vi.useFakeTimers() resetForTesting() + mockRedisInstance.status = 'ready' mockEnv.REDIS_URL = 'redis://localhost:6379' mockEnv.REDIS_TLS_SERVERNAME = undefined MockRedisConstructor.mockImplementation( @@ -159,6 +161,114 @@ describe('redis config', () => { }) }) + describe('describeRedisConnection', () => { + it('reports no client before one is built', () => { + const d = describeRedisConnection() + + expect(d.status).toBe('no-client') + expect(d.clientAgeMs).toBeNull() + expect(d.readyAgeMs).toBeNull() + expect(d.connects).toBe(0) + }) + + it('separates a connecting client from a ready one', () => { + // The constructor copies the mock's fields, so each state has to be set + // before the client is built. + mockRedisInstance.status = 'connecting' + getRedisClient() + expect(describeRedisConnection().status).toBe('connecting') + + resetForTesting() + mockRedisInstance.status = 'ready' + getRedisClient() + expect(describeRedisConnection().status).toBe('ready') + }) + + it('counts lifecycle events so a reconnect is distinguishable from a first connect', async () => { + getRedisClient() + const handler = (event: string) => + mockRedisInstance.on.mock.calls.find((c: unknown[]) => c[0] === event)?.[1] as + | (() => void) + | undefined + + handler('connect')?.() + handler('ready')?.() + const afterConnect = describeRedisConnection() + expect(afterConnect.connects).toBe(1) + expect(afterConnect.readyAgeMs).not.toBeNull() + + const errorHandler = mockRedisInstance.on.mock.calls.find( + (c: unknown[]) => c[0] === 'error' + )?.[1] as ((e: Error) => void) | undefined + errorHandler?.(new Error('ECONNRESET')) + + const afterError = describeRedisConnection() + expect(afterError.errors).toBe(1) + expect(afterError.lastErrorMessage).toBe('ECONNRESET') + }) + + it('classifies the host without ever exposing the URL that carries the auth token', () => { + mockEnv.REDIS_URL = 'rediss://10.0.0.5:6379' + mockEnv.REDIS_TLS_SERVERNAME = 'primary.example.cache.amazonaws.com' + + const d = describeRedisConnection() + + expect(d).toMatchObject({ hostKind: 'ip', tls: true, sniOverride: true }) + expect(JSON.stringify(d)).not.toContain('10.0.0.5') + }) + + it('never throws, so it cannot mask the error it is describing', () => { + // Called from catch blocks: a throw here would replace the real failure. + mockEnv.REDIS_URL = undefined + expect(() => describeRedisConnection()).not.toThrow() + + mockEnv.REDIS_URL = 'not a url' + expect(() => describeRedisConnection()).not.toThrow() + expect(describeRedisConnection().hostKind).toBe('unknown') + + // rediss:// to a bare IP with no REDIS_TLS_SERVERNAME makes the URL + // resolution throw; the snapshot must still come back. + mockEnv.REDIS_URL = 'rediss://10.0.0.5:6379' + mockEnv.REDIS_TLS_SERVERNAME = undefined + expect(() => describeRedisConnection()).not.toThrow() + }) + + it('does not date a connection that has been discarded', async () => { + mockRedisInstance.status = 'ready' + getRedisClient() + expect(describeRedisConnection().clientAgeMs).not.toBeNull() + + // Two consecutive PING failures drop the cached client. + mockRedisInstance.ping.mockRejectedValue(new Error('ETIMEDOUT')) + await vi.advanceTimersByTimeAsync(15_000) + await vi.advanceTimersByTimeAsync(15_000) + + const d = describeRedisConnection() + expect(d.status).toBe('no-client') + expect(d.clientAgeMs).toBeNull() + expect(d.readyAgeMs).toBeNull() + expect(d.msSinceLastPingOk).toBeNull() + // Lifecycle counters stay cumulative for the process. + expect(d.reconnects).toBeGreaterThanOrEqual(0) + }) + + it('classifies an IPv6 literal as an IP, not a DNS name', () => { + mockEnv.REDIS_URL = 'rediss://[2600:1f18::1]:6379' + + const d = describeRedisConnection() + + expect(d.hostKind).toBe('ip') + // Mirrors resolveRedisTlsOptions, which applies the override for IPv4 only. + expect(d.sniOverride).toBe(false) + }) + + it('reports a DNS host so resolution latency can be ruled in or out', () => { + mockEnv.REDIS_URL = 'rediss://primary.example.cache.amazonaws.com:6379' + + expect(describeRedisConnection()).toMatchObject({ hostKind: 'dns', sniOverride: false }) + }) + }) + describe('closeRedisConnection', () => { it('should clear the PING interval', async () => { getRedisClient() diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts index 04ae9ae53b2..2926ea901db 100644 --- a/apps/sim/lib/core/config/redis.ts +++ b/apps/sim/lib/core/config/redis.ts @@ -1,3 +1,4 @@ +import { isIP } from 'node:net' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { randomFloat } from '@sim/utils/random' @@ -59,6 +60,13 @@ interface RedisState { pingInterval: NodeJS.Timeout | null pingInFlight: boolean reconnectListeners: Array<() => void> + clientCreatedAt: number | null + lastReadyAt: number | null + lastPingOkAt: number | null + connects: number + reconnects: number + errors: number + lastErrorMessage: string | null } const g = globalThis as typeof globalThis & { _redisState?: RedisState } @@ -69,10 +77,112 @@ if (!g._redisState) { pingInterval: null, pingInFlight: false, reconnectListeners: [], + clientCreatedAt: null, + lastReadyAt: null, + lastPingOkAt: null, + connects: 0, + reconnects: 0, + errors: 0, + lastErrorMessage: null, } } const state = g._redisState +/** + * A command that never gets a reply fails identically whichever of three states + * the client was in — still establishing its connection, reconnecting with the + * command parked in the offline queue, or holding a socket that has silently + * died. ioredis reports all three the same way: an `Error: Command timed out` + * whose stack contains only its own timer frames, with no app frame naming the + * call and no lifecycle event to say which happened. + * + * `status` is what separates them: `connecting` and `reconnecting` mean the + * command was parked waiting on connection setup, while `ready` means it was + * written to a live socket that never answered — a socket dead in a way nothing + * reported. + */ +export interface RedisConnectionDiagnostics { + status: string + /** Age of the client object — distinguishes one created for this unit of work from an inherited one. */ + clientAgeMs: number | null + /** Time since the connection last reached `ready`. */ + readyAgeMs: number | null + /** Time since the last PING round-trip actually completed; the health check runs every 15s. */ + msSinceLastPingOk: number | null + connects: number + reconnects: number + errors: number + lastErrorMessage: string | null + /** Whether REDIS_URL targets an IP literal or a DNS name. Names DNS resolution in or out. */ + hostKind: 'ip' | 'dns' | 'unknown' + tls: boolean + /** Whether the TLS SNI override is in play (set when the host is a bare IP). */ + sniOverride: boolean +} + +/** Milliseconds since a recorded instant, or null when it was never recorded. */ +function elapsedSince(at: number | null): number | null { + return at === null ? null : Date.now() - at +} + +function describeRedisUrl( + url: string | null +): Pick { + if (!url) return { hostKind: 'unknown', tls: false, sniOverride: false } + try { + const parsed = new URL(url) + // WHATWG keeps IPv6 literals bracketed in `hostname`; `isIP` wants them bare. + const host = parsed.hostname.replace(/^\[|\]$/g, '') + const tls = parsed.protocol === 'rediss:' + // `sniOverride` deliberately mirrors `resolveRedisTlsOptions`, which tests for + // IPv4 only. So an IPv6 literal over TLS reports `hostKind: 'ip'` with + // `sniOverride: false` — not a contradiction but the useful reading, since that + // combination is a connection whose certificate cannot verify. + return { + hostKind: isIP(host) === 0 ? 'dns' : 'ip', + tls, + sniOverride: tls && isIP(host) === 4, + } + } catch { + return { hostKind: 'unknown', tls: false, sniOverride: false } + } +} + +/** + * Connection state at a point in time, safe to attach to any log line. + * + * Derives only non-sensitive facts from REDIS_URL — never the URL itself, which + * carries the AUTH token. + */ +export function describeRedisConnection(): RedisConnectionDiagnostics { + let url: string | null = null + try { + url = getConfiguredRedisUrl() + } catch { + url = null + } + + const client = state.client + + // Ages describe the client currently held. A discarded client leaves its + // timestamps behind until the next `getRedisClient()` rebuilds them, and + // reporting those against `no-client` would date a connection that no longer + // exists. The counters below are deliberately cumulative for the process. + const ageOf = (at: number | null) => (client === null ? null : elapsedSince(at)) + + return { + status: client?.status ?? 'no-client', + clientAgeMs: ageOf(state.clientCreatedAt), + readyAgeMs: ageOf(state.lastReadyAt), + msSinceLastPingOk: ageOf(state.lastPingOkAt), + connects: state.connects, + reconnects: state.reconnects, + errors: state.errors, + lastErrorMessage: state.lastErrorMessage, + ...describeRedisUrl(url), + } +} + const PING_INTERVAL_MS = 15_000 const MAX_PING_FAILURES = 2 @@ -103,6 +213,7 @@ function startPingHealthCheck(redis: Redis): void { try { await redis.ping() state.pingFailures = 0 + state.lastPingOkAt = Date.now() } catch (error) { state.pingFailures++ logger.warn('Redis PING failed', { @@ -172,6 +283,7 @@ export function getRedisClient(): Redis | null { const base = Math.min(1000 * 2 ** (times - 1), 10000) const jitter = randomFloat() * base * 0.3 const delay = Math.round(base + jitter) + state.reconnects++ logger.warn('Redis reconnecting', { attempt: times, nextRetryMs: delay }) return delay }, @@ -182,9 +294,32 @@ export function getRedisClient(): Redis | null { }, }) - state.client.on('connect', () => logger.info('Redis connected')) - state.client.on('ready', () => logger.info('Redis ready')) + state.clientCreatedAt = Date.now() + state.lastReadyAt = null + state.lastPingOkAt = null + + state.client.on('connect', () => { + state.connects++ + // Elapsed since construction, because the wait before a connection becomes + // usable is the number this path has never been able to produce: it is + // spent inside a command's deadline, where it surfaces as a command + // timeout rather than as connection latency. + // `connectCount`, not `attempt` — the retryStrategy above already logs an + // `attempt`, meaning the retry number. This is how many times this client + // has successfully connected, which is what separates a first connect from + // a reconnect. + logger.info('Redis connected', { + elapsedMs: elapsedSince(state.clientCreatedAt), + connectCount: state.connects, + }) + }) + state.client.on('ready', () => { + state.lastReadyAt = Date.now() + logger.info('Redis ready', { elapsedMs: elapsedSince(state.clientCreatedAt) }) + }) state.client.on('error', (err: Error) => { + state.errors++ + state.lastErrorMessage = err.message logger.error('Redis error', { error: err.message, code: (err as any).code }) }) state.client.on('close', () => logger.warn('Redis connection closed')) @@ -355,4 +490,11 @@ export function resetForTesting(): void { state.pingFailures = 0 state.pingInFlight = false state.reconnectListeners.length = 0 + state.clientCreatedAt = null + state.lastReadyAt = null + state.lastPingOkAt = null + state.connects = 0 + state.reconnects = 0 + state.errors = 0 + state.lastErrorMessage = null } diff --git a/packages/testing/src/mocks/redis-config.mock.ts b/packages/testing/src/mocks/redis-config.mock.ts index 82c9a4d88d2..9e5e548f2c8 100644 --- a/packages/testing/src/mocks/redis-config.mock.ts +++ b/packages/testing/src/mocks/redis-config.mock.ts @@ -43,6 +43,27 @@ function getRedisConnectionDefaultsImpl(url?: string): { } } +/** + * Mirrors the real `describeRedisConnection` under its Redis-unavailable + * default: no client, no lifecycle history, and nothing derivable from an + * unset REDIS_URL. + */ +function describeRedisConnectionImpl() { + return { + status: 'no-client', + clientAgeMs: null, + readyAgeMs: null, + msSinceLastPingOk: null, + connects: 0, + reconnects: 0, + errors: 0, + lastErrorMessage: null, + hostKind: 'unknown' as const, + tls: false, + sniOverride: false, + } +} + /** * Controllable mock functions for `@/lib/core/config/redis`. * Default: `getConfiguredRedisUrl` and `getRedisClient` return `null` (tests @@ -69,6 +90,7 @@ export const redisConfigMockFns = { mockExtendLock: vi.fn().mockResolvedValue(true), mockCloseRedisConnection: vi.fn().mockResolvedValue(undefined), mockResetForTesting: vi.fn(), + mockDescribeRedisConnection: vi.fn(describeRedisConnectionImpl), } /** @@ -86,6 +108,9 @@ export function resetRedisConfigMock(): void { redisConfigMockFns.mockExtendLock.mockReset().mockResolvedValue(true) redisConfigMockFns.mockCloseRedisConnection.mockReset().mockResolvedValue(undefined) redisConfigMockFns.mockResetForTesting.mockReset() + redisConfigMockFns.mockDescribeRedisConnection + .mockReset() + .mockImplementation(describeRedisConnectionImpl) } /** @@ -107,4 +132,5 @@ export const redisConfigMock = { extendLock: redisConfigMockFns.mockExtendLock, closeRedisConnection: redisConfigMockFns.mockCloseRedisConnection, resetForTesting: redisConfigMockFns.mockResetForTesting, + describeRedisConnection: redisConfigMockFns.mockDescribeRedisConnection, }