fix(redis): connect the signal subscriber on intent and complete the budget model

Findings from an independent review of the finished branch.

Constructing the execution-signal hub is what connects — ioredis dials in its
constructor — so the warm-up that observed that handshake and returned a
boolean its only caller discarded is gone. The execution entry point now
constructs the hub, synchronously and never throwing, and the subscribe that
follows waits on its own budget. With no observer to accommodate, the waiter
machinery collapses to a race between the shared readiness signal and a
per-waiter deadline, with a per-signal count so the last waiter detaches.

The budget's dead-attempt term was still short for an unanswered handshake:
after the ready check times out, ioredis half-closes the socket and destroys
it only after `disconnectTimeout` when a wedged peer never answers with a FIN.
Measured against such a peer: an unauthenticated connection reconnects at
12.5s, past the 11.5s the budget allowed. The term is now
`max(connectTimeout, 2 * commandTimeout + disconnectTimeout)`, with the
disconnect deadline stated in the shared connection defaults so the budget
derives from it. The model is specific to ioredis 5's handshake sequence, so it
lives beside the pinned client in `lib/core/config/redis-budget.ts` — a module
the global test mock does not replace — rather than in a generic package.

Corrects two doc claims: no caller warms the subscriber at server boot, and a
late joiner never had a full wait of its own before this branch either — the
memoized promise shared its single timer.

Author: waleedlatif1

packages/utils/src/retry.test.ts …sim/lib/core/config/redis-budget.test.tspackages/utils/src/retry.test.ts renamed to apps/sim/lib/core/config/redis-budget.test.ts packages/utils/src/retry.test.ts renamed to apps/sim/lib/core/config/redis-budget.test.ts

@@ -1,16 +1,21 @@
+/**
+ * @vitest-environment node
+ */
 import { describe, expect, it } from 'vitest'
-import { coldConnectionBudgetMs } from './retry'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 describe('coldConnectionBudgetMs', () => {
-  it('charges a dead handshake at two command deadlines when that exceeds the connect deadline', () => {
-    // Unauthenticated: SETNAME/SETINFO settle by timing out before INFO starts its own deadline.
+  it('charges an unanswered handshake at two command deadlines plus the half-close wait', () => {
+    // Unauthenticated: SETNAME/SETINFO time out, then INFO times out, then the
+    // half-closed socket waits for a FIN a wedged peer never sends.
     expect(
       coldConnectionBudgetMs({
         connectTimeoutMs: 1_000,
-        commandTimeoutMs: 2_000,
+        commandTimeoutMs: 5_000,
+        disconnectTimeoutMs: 2_000,
         reconnectDelayMs: 500,
       })
-    ).toBe(2 * 2_000 + 500 + 1_000)
+    ).toBe(2 * 5_000 + 2_000 + 500 + 1_000)
   })
 
   it('charges the connect deadline when a connection that never completes is the longer case', () => {
@@ -20,14 +25,20 @@ describe('coldConnectionBudgetMs', () => {
       coldConnectionBudgetMs({
         connectTimeoutMs: 10_000,
         commandTimeoutMs: 3_000,
+        disconnectTimeoutMs: 2_000,
         reconnectDelayMs: 500,
       })
     ).toBe(10_000 + 500 + 1_000)
   })
 
   it('always leaves room for the healthy attempt that follows the reconnect', () => {
     expect(
-      coldConnectionBudgetMs({ connectTimeoutMs: 0, commandTimeoutMs: 0, reconnectDelayMs: 0 })
+      coldConnectionBudgetMs({
+        connectTimeoutMs: 0,
+        commandTimeoutMs: 0,
+        disconnectTimeoutMs: 0,
+        reconnectDelayMs: 0,
+      })
     ).toBe(1_000)
   })
 })

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

@@ -0,0 +1,45 @@
+/** 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
+
+export interface ColdConnectionBudgetOptions {
+  /** The client's TCP connect deadline: bounds a connection that never completes at all. */
+  connectTimeoutMs: number
+  /** The client's per-command deadline: bounds a handshake command the server never answers. */
+  commandTimeoutMs: number
+  /** How long ioredis waits after half-closing a dead socket for the peer's FIN before destroying it. */
+  disconnectTimeoutMs: number
+  /** What the client's `retryStrategy` returns for its first reconnect. */
+  reconnectDelayMs: number
+}
+
+/**
+ * How long a wait for a cold ioredis connection must allow before giving up,
+ * if it is to survive one dead attempt and still see a healthy one land.
+ *
+ * A dead attempt is diagnosed by whichever deadline governs the phase it
+ * stalls in. A connect that never completes costs `connectTimeoutMs`, and the
+ * socket is destroyed outright. A connection that opens but whose handshake
+ * is never answered costs command deadlines — one with a password, because a
+ * timed-out `AUTH` is fatal; two without, because `CLIENT SETNAME`/`SETINFO`
+ * must settle, by timing out, before the `INFO` ready check starts its own —
+ * and then `disconnectTimeoutMs` more, because ioredis half-closes the socket
+ * and a peer that is wedged never answers with a FIN. The budget takes the
+ * larger phase so it holds for either URL shape without parsing it. Only then
+ * does `retryStrategy` run and a fresh attempt begin.
+ *
+ * A wait sized to a single deadline expires while the first attempt is still
+ * being diagnosed, so a configured retry can never be the thing that saves it.
+ *
+ * The guarantee is one dead attempt from a fresh or previously-ready client.
+ * ioredis feeds `retryStrategy` its running attempt count, so a wait that
+ * begins while the client is already deep in a reconnect loop faces larger
+ * delays this does not model. The handshake sequence is ioredis 5's; keep this
+ * beside the pinned client, not in a generic package.
+ */
+export function coldConnectionBudgetMs(options: ColdConnectionBudgetOptions): number {
+  const deadAttemptMs = Math.max(
+    options.connectTimeoutMs,
+    2 * options.commandTimeoutMs + options.disconnectTimeoutMs
+  )
+  return deadAttemptMs + options.reconnectDelayMs + HEALTHY_HANDSHAKE_ALLOWANCE_MS
+}

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

@@ -1,5 +1,4 @@
 import { createMockRedis } from '@sim/testing'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { mockEnv, MockRedisConstructor, mockLogger } = vi.hoisted(() => ({
@@ -48,15 +47,19 @@ vi.mock('ioredis', () => ({
 
 import {
   acquireLock,
+  CONNECT_TIMEOUT_MS,
   closeRedisConnection,
+  DISCONNECT_TIMEOUT_MS,
   describeRedisConnection,
   extendLock,
   getRedisClient,
   onRedisReconnect,
   resetForTesting,
+  SHARED_COMMAND_TIMEOUT_MS,
   sharedReconnectDelayMs,
   warmRedisConnection,
 } from '@/lib/core/config/redis'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 describe('redis config', () => {
   beforeEach(() => {
@@ -488,9 +491,12 @@ describe('redis config', () => {
       mockRedisInstance.status = 'connecting'
       const warm = warmRedisConnection()
 
-      // The dead attempt's own deadline, then the longest first reconnect
-      // delay: the moment a healthy second attempt can begin.
-      await vi.advanceTimersByTimeAsync(Math.max(10_000, 2 * 5_000) + sharedReconnectDelayMs(1, 1))
+      // The dead attempt's own diagnosis and half-close, then the longest first
+      // reconnect delay: the moment a healthy second attempt can begin.
+      await vi.advanceTimersByTimeAsync(
+        Math.max(CONNECT_TIMEOUT_MS, 2 * SHARED_COMMAND_TIMEOUT_MS + DISCONNECT_TIMEOUT_MS) +
+          sharedReconnectDelayMs(1, 1)
+      )
       const client = getRedisClient()
       Object.assign(client ?? {}, { status: 'ready' })
       client?.emit('ready')
@@ -504,8 +510,9 @@ describe('redis config', () => {
 
       await vi.advanceTimersByTimeAsync(
         coldConnectionBudgetMs({
-          connectTimeoutMs: 10_000,
-          commandTimeoutMs: 5_000,
+          connectTimeoutMs: CONNECT_TIMEOUT_MS,
+          commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS,
+          disconnectTimeoutMs: DISCONNECT_TIMEOUT_MS,
           reconnectDelayMs: sharedReconnectDelayMs(1, 1),
         })
       )

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

@@ -2,10 +2,10 @@ import { isIP } from 'node:net'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { randomFloat } from '@sim/utils/random'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import Redis from 'ioredis'
 import { env } from '@/lib/core/config/env'
 import { getConfiguredCacheProvider } from '@/lib/core/config/env-capabilities.server'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 const logger = createLogger('Redis')
 
@@ -46,17 +46,21 @@ function resolveRedisTlsOptions(url: string | undefined): { servername: string }
 export interface RedisConnectionDefaults {
   keepAlive: number
   connectTimeout: number
+  disconnectTimeout: number
   enableOfflineQueue: boolean
   tls?: { servername: string }
 }
 
-const CONNECT_TIMEOUT_MS = 10_000
+export const CONNECT_TIMEOUT_MS = 10_000
+/** ioredis's own default, stated so readiness budgets can be derived from it. */
+export const DISCONNECT_TIMEOUT_MS = 2_000
 
 export function getRedisConnectionDefaults(url: string | undefined): RedisConnectionDefaults {
   const tls = resolveRedisTlsOptions(url)
   return {
     keepAlive: 1000,
     connectTimeout: CONNECT_TIMEOUT_MS,
+    disconnectTimeout: DISCONNECT_TIMEOUT_MS,
     enableOfflineQueue: true,
     ...(tls ? { tls } : {}),
   }
@@ -211,7 +215,7 @@ export function describeRedisConnection(
 
 const PING_INTERVAL_MS = 15_000
 const MAX_PING_FAILURES = 2
-const SHARED_COMMAND_TIMEOUT_MS = 5_000
+export 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
@@ -236,6 +240,7 @@ export function sharedReconnectDelayMs(times: number, jitter: number): number {
 const REDIS_WARMUP_TIMEOUT_MS = coldConnectionBudgetMs({
   connectTimeoutMs: CONNECT_TIMEOUT_MS,
   commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS,
+  disconnectTimeoutMs: DISCONNECT_TIMEOUT_MS,
   reconnectDelayMs: sharedReconnectDelayMs(1, 1),
 })
 

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

@@ -2,7 +2,6 @@
  * @vitest-environment node
  */
 import { EventEmitter } from 'node:events'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { connection, mockRedisUrl, mockSubscribe, mockUnsubscribe } = vi.hoisted(() => ({
@@ -41,26 +40,30 @@ vi.mock('@/lib/core/config/redis', () => ({
     if (mockRedisUrl.error) throw mockRedisUrl.error
     return mockRedisUrl.value
   },
-  // Realistic defaults: the readiness budget has a connect-deadline term.
-  getRedisConnectionDefaults: () => ({ connectTimeout: 10_000 }),
+  // Realistic defaults: the readiness budget derives from these. The literals
+  // are inherent to mocking the module that exports the real constants.
+  getRedisConnectionDefaults: () => ({ connectTimeout: 10_000, disconnectTimeout: 2_000 }),
 }))
 
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 import {
+  connectExecutionSignalHub,
   getExecutionSignalHub,
   publishLocalExecutionSignal,
-  warmExecutionSignalHub,
 } from '@/lib/execution/execution-signal'
 
 /** The readiness budget production derives from the options the subscriber was built with. */
 function readyBudgetMs(): number {
   const options = connection.options as {
     connectTimeout: number
     commandTimeout: number
+    disconnectTimeout: number
     retryStrategy: (attempt: number) => number
   }
   return coldConnectionBudgetMs({
     connectTimeoutMs: options.connectTimeout,
     commandTimeoutMs: options.commandTimeout,
+    disconnectTimeoutMs: options.disconnectTimeout,
     reconnectDelayMs: options.retryStrategy(1),
   })
 }
@@ -401,27 +404,33 @@ describe('ExecutionSignalHub', () => {
     }
   })
 
-  it('gives a subscribe that joins an in-flight warm-up its own full budget', async () => {
+  it('gives a subscribe that joins another in-flight wait its own full budget', async () => {
     vi.useFakeTimers()
     try {
       connection.status = 'connect'
-      const warm = warmExecutionSignalHub()
-      // Late joiner: the warm-up has almost spent its budget when this subscribe begins.
+      const hub = getExecutionSignalHub()
+      const first = hub.subscribe('execution-first', vi.fn())
+      const firstSettled = vi.fn()
+      void first.then(firstSettled, firstSettled)
+      // Late joiner: the first wait has almost spent its budget when this one begins.
       await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000)
-      const subscription = getExecutionSignalHub().subscribe('execution-1', vi.fn())
-      const settled = vi.fn()
-      void subscription.then(settled, settled)
+      const second = hub.subscribe('execution-second', vi.fn())
+      const secondSettled = vi.fn()
+      void second.then(secondSettled, secondSettled)
 
       await vi.advanceTimersByTimeAsync(1000)
-      await expect(warm).resolves.toBe(false)
-      // The warm-up's deadline was its own; the subscribe is still waiting.
-      expect(settled).not.toHaveBeenCalled()
+      await expect(first).rejects.toThrow('Timed out waiting for Redis subscriber readiness')
+      // The first deadline was its own; the second is still waiting on the shared signal.
+      expect(secondSettled).not.toHaveBeenCalled()
       expect(connection.client?.listenerCount('ready')).toBe(2)
 
       connection.status = 'ready'
       connection.client?.emit('ready')
-      await subscription
-      expect(mockSubscribe).toHaveBeenCalledOnce()
+      await second
+      expect(mockSubscribe).toHaveBeenCalledWith(
+        'execution:signal:execution-second',
+        'execution:cancel'
+      )
     } finally {
       vi.useRealTimers()
     }
@@ -472,42 +481,21 @@ describe('ExecutionSignalHub', () => {
     }
   })
 
-  it('warms to true once the subscriber becomes ready', async () => {
+  it('begins connecting when asked to connect ahead of a subscription', () => {
     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(readyBudgetMs())
+    connectExecutionSignalHub()
 
-      await expect(warm).resolves.toBe(false)
-      expect(vi.getTimerCount()).toBe(0)
-    } finally {
-      vi.useRealTimers()
-    }
+    // Constructing the hub is what dials; the client exists before any subscribe.
+    expect(connection.client).toBeDefined()
+    expect(mockSubscribe).not.toHaveBeenCalled()
   })
 
-  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.
+  it('does not throw when Redis is misconfigured, leaving that to the first subscriber', () => {
     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)
+    expect(() => connectExecutionSignalHub()).not.toThrow()
+    expect(connection.client).toBeUndefined()
   })
 
   it('uses a process-local signal hub when Redis is not configured', async () => {