fix(redis): size cold-connection waits to survive a dead handshake

A dead Redis handshake costs two command deadlines, not one. ioredis sends
CLIENT SETNAME/SETINFO on connect and dispatches the INFO ready check only
once those settle — on a socket that never answers, they settle by timing
out. Only then does the ready check start its own deadline, fail, and tear the
socket down for retryStrategy to reconnect. Every readiness wait in the
codebase was sized to a single deadline, so it expired while the first attempt
was still being diagnosed and a configured retry could never be the thing that
rescued it.

Introduces `coldConnectionBudgetMs`, which derives a wait from the command
deadline and the reconnect delays a caller's own retryStrategy would return,
and states the 2x in one place. Both waits now derive from it:

- The execution-signal subscriber's readiness budget was the same constant as
  its commandTimeout (5s and 5s), so ioredis's reconnect was decorative on the
  cold path that failed in production. It now has a 2s command deadline — this
  client only issues SUBSCRIBE/UNSUBSCRIBE, always after ready — and a budget
  with room for two dead attempts and a healthy one.
- The shared client's warm-up budget was 10s against a 5s deadline, exactly
  the moment ioredis would first tear a stalled socket down, so on the case
  warming exists for it gave up just before recovery could land.

Warms the execution-signal subscriber alongside the shared client at process
start, in parallel, never throwing: it is the second connection a run opens
and was paying its handshake inside its own readiness budget.

The shared redis-config mock mirrors the pure budget helper, since consumers
now evaluate it at module load and would otherwise fail to import under the
global mock.

Author: waleedlatif1

apps/sim/instrumentation-node.ts

@@ -404,9 +404,13 @@ export async function register() {
   const { startMemoryTelemetry } = await import('./lib/monitoring/memory-telemetry')
   startMemoryTelemetry()
 
-  // Not awaited: the connection is warmed in the background so the first request
-  // that needs Redis does not pay the handshake inside its own command deadline,
-  // but boot never waits on Redis to serve requests that do not touch it.
-  const { warmRedisConnection } = await import('./lib/core/config/redis')
+  // Not awaited: both Redis connections are warmed in the background so the
+  // first request that needs one does not pay the handshake inside its own
+  // deadline, but boot never waits on Redis to serve requests that do not touch it.
+  const [{ warmRedisConnection }, { warmExecutionSignalHub }] = await Promise.all([
+    import('./lib/core/config/redis'),
+    import('./lib/execution/execution-signal'),
+  ])
   void warmRedisConnection()
+  void warmExecutionSignalHub()
 }

apps/sim/lib/core/config/redis.test.ts

@@ -48,10 +48,12 @@ vi.mock('ioredis', () => ({
 import {
   acquireLock,
   closeRedisConnection,
+  coldConnectionBudgetMs,
   describeRedisConnection,
   extendLock,
   getRedisClient,
   onRedisReconnect,
+  REDIS_WARMUP_TIMEOUT_MS,
   resetForTesting,
   warmRedisConnection,
 } from '@/lib/core/config/redis'
@@ -471,7 +473,44 @@ describe('redis config', () => {
     })
   })
 
+  describe('coldConnectionBudgetMs', () => {
+    it('charges two command deadlines per dead handshake, plus each reconnect delay', () => {
+      // SETNAME/SETINFO gate the INFO ready check, and on a dead socket they
+      // settle only by timing out — so a dead attempt costs 2x, not 1x.
+      expect(coldConnectionBudgetMs({ commandTimeoutMs: 2_000, retryDelaysMs: [500, 1_000] })).toBe(
+        2 * 2_000 + 500 + 2 * 2_000 + 1_000 + 1_000
+      )
+    })
+
+    it('leaves only the healthy-handshake allowance when no dead attempts are tolerated', () => {
+      expect(coldConnectionBudgetMs({ commandTimeoutMs: 5_000, retryDelaysMs: [] })).toBe(1_000)
+    })
+  })
+
   describe('warmRedisConnection', () => {
+    it('outlasts one dead handshake so the reconnect can be what warms it', async () => {
+      mockRedisInstance.status = 'connecting'
+      const warm = warmRedisConnection()
+
+      // Two command deadlines to diagnose the dead socket, then the longest
+      // first reconnect delay: the moment a healthy second attempt can begin.
+      await vi.advanceTimersByTimeAsync(2 * 5_000 + 1_300)
+      const client = getRedisClient()
+      Object.assign(client ?? {}, { status: 'ready' })
+      client?.emit('ready')
+
+      await expect(warm).resolves.toBe(true)
+    })
+
+    it('still gives up once the budget is spent', async () => {
+      mockRedisInstance.status = 'connecting'
+      const warm = warmRedisConnection()
+
+      await vi.advanceTimersByTimeAsync(REDIS_WARMUP_TIMEOUT_MS)
+
+      await expect(warm).resolves.toBe(false)
+    })
+
     it('resolves immediately when the connection is already usable', async () => {
       mockRedisInstance.status = 'ready'
 

apps/sim/lib/core/config/redis.ts

@@ -203,12 +203,49 @@ export function describeRedisConnection(
 
 const PING_INTERVAL_MS = 15_000
 const MAX_PING_FAILURES = 2
+const SHARED_COMMAND_TIMEOUT_MS = 5_000
+const RECONNECT_BASE_MS = 1_000
+const RECONNECT_MAX_BASE_MS = 10_000
+const RECONNECT_JITTER_RATIO = 0.3
+/** Generous room for the healthy handshake that follows a recovered stall; a real one takes tens of milliseconds. */
+const HEALTHY_HANDSHAKE_ALLOWANCE_MS = 1_000
+
+/**
+ * How long a wait for a cold connection must allow before giving up, if it is
+ * to survive `retryDelaysMs.length` dead handshakes and still see a healthy one
+ * land.
+ *
+ * A dead handshake costs **two** command deadlines, not one. ioredis sends
+ * `CLIENT SETNAME`/`CLIENT SETINFO` on connect and dispatches the `INFO` ready
+ * check only once those settle — and on a socket that never answers they
+ * settle by timing out. Only then does the ready check start its own deadline,
+ * fail, and tear the socket down for `retryStrategy` to reconnect. A budget
+ * sized to one deadline expires while the first attempt is still being
+ * diagnosed, so a configured retry can never be the thing that saves it.
+ *
+ * `retryDelaysMs` lists the reconnect delay after each dead attempt, in order,
+ * so a caller states exactly what its own `retryStrategy` would return.
+ */
+export function coldConnectionBudgetMs(options: {
+  commandTimeoutMs: number
+  retryDelaysMs: readonly number[]
+}): number {
+  const recovery = options.retryDelaysMs.reduce(
+    (total, retryDelayMs) => total + 2 * options.commandTimeoutMs + retryDelayMs,
+    0
+  )
+  return recovery + HEALTHY_HANDSHAKE_ALLOWANCE_MS
+}
+
 /**
- * Warm-up budget. Sized to outlast a slow handshake rather than a fast one,
- * because giving up early just returns the handshake to the first command's
- * deadline, which is the thing this exists to avoid.
+ * Warm-up budget: one dead handshake, its longest possible first reconnect
+ * delay, then a healthy attempt. Giving up sooner returns the handshake to the
+ * first command's deadline, which is the thing warming exists to avoid.
  */
-const REDIS_WARMUP_TIMEOUT_MS = 10_000
+export const REDIS_WARMUP_TIMEOUT_MS = coldConnectionBudgetMs({
+  commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS,
+  retryDelaysMs: [RECONNECT_BASE_MS * (1 + RECONNECT_JITTER_RATIO)],
+})
 
 export function getConfiguredRedisUrl(): string | null {
   if (getConfiguredCacheProvider() === 'database') return null
@@ -296,16 +333,16 @@ export function getRedisClient(): Redis | null {
 
     state.client = new Redis(redisUrl, {
       ...defaults,
-      commandTimeout: 5000,
+      commandTimeout: SHARED_COMMAND_TIMEOUT_MS,
       maxRetriesPerRequest: 5,
 
       retryStrategy: (times) => {
         if (times > 10) {
           logger.error(`Redis reconnection attempt ${times}`, { nextRetryMs: 30000 })
           return 30000
         }
-        const base = Math.min(1000 * 2 ** (times - 1), 10000)
-        const jitter = randomFloat() * base * 0.3
+        const base = Math.min(RECONNECT_BASE_MS * 2 ** (times - 1), RECONNECT_MAX_BASE_MS)
+        const jitter = randomFloat() * base * RECONNECT_JITTER_RATIO
         const delay = Math.round(base + jitter)
         state.reconnects++
         logger.warn('Redis reconnecting', { attempt: times, nextRetryMs: delay })

apps/sim/lib/execution/execution-signal.test.ts

@@ -4,21 +4,31 @@
 import { EventEmitter } from 'node:events'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { connection, mockRedisUrl, mockSubscribe, mockUnsubscribe } = vi.hoisted(() => ({
-  connection: {
-    status: 'ready',
-    client: undefined as EventEmitter | undefined,
-  },
-  mockRedisUrl: { value: 'redis://localhost:6379' as string | undefined },
-  mockSubscribe: vi.fn(),
-  mockUnsubscribe: vi.fn(),
-}))
+const { connection, mockRedisUrl, mockSubscribe, mockUnsubscribe, budgetInputs } = vi.hoisted(
+  () => ({
+    connection: {
+      status: 'ready',
+      client: undefined as EventEmitter | undefined,
+      options: undefined as Record<string, unknown> | undefined,
+    },
+    mockRedisUrl: {
+      value: 'redis://localhost:6379' as string | undefined,
+      error: undefined as Error | undefined,
+    },
+    mockSubscribe: vi.fn(),
+    mockUnsubscribe: vi.fn(),
+    // A plain object, not a spy: the budget is computed once at module load, and
+    // `vi.clearAllMocks()` in beforeEach would erase a spy's record of that call.
+    budgetInputs: { value: undefined as unknown },
+  })
+)
 
 vi.mock('ioredis', () => ({
   default: class extends EventEmitter {
-    constructor() {
+    constructor(_url: string, options: Record<string, unknown>) {
       super()
       connection.client = this
+      connection.options = options
     }
 
     get status() {
@@ -31,13 +41,24 @@ vi.mock('ioredis', () => ({
 }))
 
 vi.mock('@/lib/core/config/redis', () => ({
-  getConfiguredRedisUrl: () => mockRedisUrl.value,
+  getConfiguredRedisUrl: () => {
+    if (mockRedisUrl.error) throw mockRedisUrl.error
+    return mockRedisUrl.value
+  },
   getRedisConnectionDefaults: () => ({}),
+  // The budget arithmetic itself is covered in redis.test.ts; here only the
+  // resulting number matters, and the readiness test reads it back by name.
+  coldConnectionBudgetMs: (inputs: unknown) => {
+    budgetInputs.value = inputs
+    return 10_000
+  },
 }))
 
 import {
   getExecutionSignalHub,
   publishLocalExecutionSignal,
+  SUBSCRIBER_READY_TIMEOUT_MS,
+  warmExecutionSignalHub,
 } from '@/lib/execution/execution-signal'
 
 describe('ExecutionSignalHub', () => {
@@ -48,6 +69,7 @@ describe('ExecutionSignalHub', () => {
     mockSubscribe.mockResolvedValue(1)
     mockUnsubscribe.mockResolvedValue(0)
     mockRedisUrl.value = 'redis://localhost:6379'
+    mockRedisUrl.error = undefined
     const signalGlobal = globalThis as typeof globalThis & { _executionSignalHub?: unknown }
     signalGlobal._executionSignalHub = undefined
   })
@@ -241,7 +263,7 @@ describe('ExecutionSignalHub', () => {
         'Timed out waiting for Redis subscriber readiness'
       )
 
-      const timeout = vi.advanceTimersByTimeAsync(4000).then(() => {
+      const timeout = vi.advanceTimersByTimeAsync(SUBSCRIBER_READY_TIMEOUT_MS - 1000).then(() => {
         connection.client?.emit('error', new Error('ECONNREFUSED'))
         return vi.advanceTimersByTimeAsync(1000)
       })
@@ -356,6 +378,63 @@ describe('ExecutionSignalHub', () => {
     expect(replacement).not.toHaveBeenCalledWith('unavailable')
   })
 
+  it('derives the readiness budget from the exact options the subscriber is built with', () => {
+    getExecutionSignalHub()
+    const options = connection.options as {
+      commandTimeout: number
+      retryStrategy: (attempt: number) => number
+    }
+
+    // Two dead handshakes, so the budget states the reconnect delays after
+    // attempt 1 and attempt 2 as the client's own retryStrategy would return them.
+    expect(budgetInputs.value).toEqual({
+      commandTimeoutMs: options.commandTimeout,
+      retryDelaysMs: [options.retryStrategy(1), options.retryStrategy(2)],
+    })
+    // And the wait must outlast at least one full dead attempt, or the retry is decorative.
+    expect(SUBSCRIBER_READY_TIMEOUT_MS).toBeGreaterThan(
+      2 * options.commandTimeout + options.retryStrategy(1)
+    )
+  })
+
+  it('warms to true once the subscriber becomes ready', async () => {
+    connection.status = 'connecting'
+    const warm = warmExecutionSignalHub()
+
+    connection.status = 'ready'
+    connection.client?.emit('ready')
+
+    await expect(warm).resolves.toBe(true)
+  })
+
+  it('warms to false, not a rejection, when readiness never arrives', async () => {
+    vi.useFakeTimers()
+    try {
+      connection.status = 'connect'
+      const warm = warmExecutionSignalHub()
+
+      await vi.advanceTimersByTimeAsync(SUBSCRIBER_READY_TIMEOUT_MS)
+
+      await expect(warm).resolves.toBe(false)
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('reports not-warm instead of throwing when Redis is misconfigured', async () => {
+    // Start-up hooks call this; a throw there would fail the run attempt.
+    mockRedisUrl.error = new Error('Cache capability selected Redis but REDIS_URL is missing')
+
+    await expect(warmExecutionSignalHub()).resolves.toBe(false)
+  })
+
+  it('is trivially warm when signals are process-local', async () => {
+    mockRedisUrl.value = undefined
+
+    await expect(warmExecutionSignalHub()).resolves.toBe(true)
+  })
+
   it('uses a process-local signal hub when Redis is not configured', async () => {
     mockRedisUrl.value = undefined
     const handler = vi.fn()