feat(observability): record Redis connection state on failed slot operations

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.

Author: waleedlatif1

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)

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<T>(
+  operation: string,
+  reservationId: string,
+  run: () => Promise<T>
+): Promise<T> {
+  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(

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,8 @@ describe('redis config', () => {
     vi.clearAllMocks()
     vi.useFakeTimers()
     resetForTesting()
+    mockRedisInstance.status = 'ready'
+    Object.assign(mockRedisInstance, { offlineQueue: undefined })
     mockEnv.REDIS_URL = 'redis://localhost:6379'
     mockEnv.REDIS_TLS_SERVERNAME = undefined
     MockRedisConstructor.mockImplementation(
@@ -159,6 +162,121 @@ 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('surfaces the offline queue depth that proves a command was waiting on the connection', () => {
+      Object.assign(mockRedisInstance, { offlineQueue: { length: 3 } })
+      getRedisClient()
+
+      expect(describeRedisConnection().queuedCommands).toBe(3)
+    })
+
+    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()