From 2e58e0dc48467b47e62de3aaa9437424f1f54e70 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 27 Aug 2026 20:33:50 -0700 Subject: [PATCH] feat(observability): record Redis connection state on failed slot operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Redis command that never gets a reply fails identically whether the connection was still being established, was reconnecting with the command parked in the offline queue, or was a socket that had silently died. ioredis reports all three the same way — `Error: Command timed out` with only its own timer frames in the stack, no app frame naming the call, and no lifecycle event saying which happened. Nothing recorded anywhere distinguishes them, so the cause can only be inferred from timing. Adds `describeRedisConnection()`: client status, connection and ready ages, offline-queue depth, lifecycle counters, and whether the configured host is an IP or a DNS name. `status` alone usually decides it; queue depth confirms, since a parked command was waiting on connection setup while one written to a `ready` socket that never answered means the socket died unreported. Host kind rules DNS resolution in or out, which no server-side telemetry can see. Attaches it to the usage-reservation slot operations — the first Redis calls a queued workflow makes, so an unusable connection surfaces there first. Connect and ready now log elapsed-since-construction, making the wait before a connection becomes usable directly measurable; today it is spent inside a command's deadline, where it reads as a command timeout rather than as connection latency. Purely additive. Redis call arguments are unchanged, the wrapper rethrows the original error object, and `describeRedisConnection` never throws — it runs inside catch blocks where a throw would replace the real failure. All three have tests that fail if the behavior is removed. Only non-sensitive facts are derived from REDIS_URL, which carries the AUTH token and is never logged. --- .../calculations/usage-reservation.test.ts | 7 + .../billing/calculations/usage-reservation.ts | 64 ++++++-- apps/sim/lib/core/config/redis.test.ts | 110 +++++++++++++ apps/sim/lib/core/config/redis.ts | 146 +++++++++++++++++- .../testing/src/mocks/redis-config.mock.ts | 26 ++++ 5 files changed, 337 insertions(+), 16 deletions(-) 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, }