fix(redis): give every readiness waiter its own deadline and correct the budget model

Three corrections from an independent review of the branch.

The shared readiness wait carried one timer, so a subscribe that joined an
in-flight warm-up inherited only the remainder of that warm-up's budget — down
to nothing — where before this branch it was guaranteed a full wait of its own.
Waiters now share one ready/end listener pair but each runs its own deadline,
and the signal is torn down when its last waiter gives up.

The budget model claimed a dead handshake always costs two command deadlines.
That is true only without a password: with one, ioredis treats a timed-out
AUTH as fatal and tears the socket down after a single deadline, and a connect
that never completes is bounded by connectTimeout, which the formula had no
term for. `coldConnectionBudgetMs` now charges the largest of those so it holds
for either URL shape, and moves to @sim/utils/retry: a pure helper evaluated at
module load must not live in a module the test setup replaces wholesale, which
had forced the shared mock to carry a verbatim mirror of the arithmetic. The
subscriber derives its budget in its constructor from the exact options it is
built with; the shared client's reconnect delay is a pure function used by
both retryStrategy and the warm-up budget.

The Next server no longer opens the signal subscriber at boot: the execution
entry point already warms it on intent, and an eagerly opened subscriber that
cannot reach Redis would reconnect on a five-second cadence in every idle
replica for the life of the process.

Author: waleedlatif1

apps/sim/instrumentation-node.ts

@@ -404,13 +404,9 @@ export async function register() {
   const { startMemoryTelemetry } = await import('./lib/monitoring/memory-telemetry')
   startMemoryTelemetry()
 
-  // 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'),
-  ])
+  // 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')
   void warmRedisConnection()
-  void warmExecutionSignalHub()
 }

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

@@ -1,4 +1,5 @@
 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,13 +49,12 @@ vi.mock('ioredis', () => ({
 import {
   acquireLock,
   closeRedisConnection,
-  coldConnectionBudgetMs,
   describeRedisConnection,
   extendLock,
   getRedisClient,
   onRedisReconnect,
-  REDIS_WARMUP_TIMEOUT_MS,
   resetForTesting,
+  sharedReconnectDelayMs,
   warmRedisConnection,
 } from '@/lib/core/config/redis'
 
@@ -473,17 +473,13 @@ 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('sharedReconnectDelayMs', () => {
+    it('grows exponentially from the base and caps, with upward-only jitter', () => {
+      expect(sharedReconnectDelayMs(1, 0)).toBe(1_000)
+      expect(sharedReconnectDelayMs(2, 0)).toBe(2_000)
+      expect(sharedReconnectDelayMs(5, 0)).toBe(10_000)
+      expect(sharedReconnectDelayMs(6, 0)).toBe(10_000)
+      expect(sharedReconnectDelayMs(1, 1)).toBe(1_300)
     })
   })
 
@@ -492,9 +488,9 @@ describe('redis config', () => {
       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)
+      // 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))
       const client = getRedisClient()
       Object.assign(client ?? {}, { status: 'ready' })
       client?.emit('ready')
@@ -506,7 +502,13 @@ describe('redis config', () => {
       mockRedisInstance.status = 'connecting'
       const warm = warmRedisConnection()
 
-      await vi.advanceTimersByTimeAsync(REDIS_WARMUP_TIMEOUT_MS)
+      await vi.advanceTimersByTimeAsync(
+        coldConnectionBudgetMs({
+          connectTimeoutMs: 10_000,
+          commandTimeoutMs: 5_000,
+          reconnectDelayMs: sharedReconnectDelayMs(1, 1),
+        })
+      )
 
       await expect(warm).resolves.toBe(false)
     })

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

@@ -2,7 +2,8 @@ import { isIP } from 'node:net'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { randomFloat } from '@sim/utils/random'
-import Redis, { type RedisOptions } from 'ioredis'
+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'
 
@@ -42,13 +43,20 @@ function resolveRedisTlsOptions(url: string | undefined): { servername: string }
  * and TLS SNI when REDIS_URL targets an IP. Every Redis client we open should
  * spread this; callers add their own retry / timeout policy on top.
  */
-export function getRedisConnectionDefaults(
-  url: string | undefined
-): Pick<RedisOptions, 'keepAlive' | 'connectTimeout' | 'enableOfflineQueue' | 'tls'> {
+export interface RedisConnectionDefaults {
+  keepAlive: number
+  connectTimeout: number
+  enableOfflineQueue: boolean
+  tls?: { servername: string }
+}
+
+const CONNECT_TIMEOUT_MS = 10_000
+
+export function getRedisConnectionDefaults(url: string | undefined): RedisConnectionDefaults {
   const tls = resolveRedisTlsOptions(url)
   return {
     keepAlive: 1000,
-    connectTimeout: 10000,
+    connectTimeout: CONNECT_TIMEOUT_MS,
     enableOfflineQueue: true,
     ...(tls ? { tls } : {}),
   }
@@ -207,44 +215,28 @@ 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.
+ * The shared client's reconnect delay for attempt `times`, with `jitter` in
+ * `[0, 1]` scaling the upward-only jitter band. Pure so the same formula can
+ * be evaluated for a budget — `sharedReconnectDelayMs(1, 1)` is the longest
+ * possible first reconnect — as well as from `retryStrategy`, which adds the
+ * bookkeeping around it.
  */
-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
+export function sharedReconnectDelayMs(times: number, jitter: number): number {
+  const base = Math.min(RECONNECT_BASE_MS * 2 ** (times - 1), RECONNECT_MAX_BASE_MS)
+  return Math.round(base + jitter * base * RECONNECT_JITTER_RATIO)
 }
 
 /**
- * Warm-up budget: one dead handshake, its longest possible first reconnect
+ * Warm-up budget: one dead attempt, 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.
  */
-export const REDIS_WARMUP_TIMEOUT_MS = coldConnectionBudgetMs({
+const REDIS_WARMUP_TIMEOUT_MS = coldConnectionBudgetMs({
+  connectTimeoutMs: CONNECT_TIMEOUT_MS,
   commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS,
-  retryDelaysMs: [RECONNECT_BASE_MS * (1 + RECONNECT_JITTER_RATIO)],
+  reconnectDelayMs: sharedReconnectDelayMs(1, 1),
 })
 
 export function getConfiguredRedisUrl(): string | null {
@@ -341,9 +333,7 @@ export function getRedisClient(): Redis | null {
           logger.error(`Redis reconnection attempt ${times}`, { nextRetryMs: 30000 })
           return 30000
         }
-        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)
+        const delay = sharedReconnectDelayMs(times, randomFloat())
         state.reconnects++
         logger.warn('Redis reconnecting', { attempt: times, nextRetryMs: delay })
         return delay

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

@@ -2,26 +2,22 @@
  * @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, 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 },
-  })
-)
+const { connection, mockRedisUrl, mockSubscribe, mockUnsubscribe } = 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(),
+}))
 
 vi.mock('ioredis', () => ({
   default: class extends EventEmitter {
@@ -40,32 +36,35 @@ vi.mock('ioredis', () => ({
   },
 }))
 
-vi.mock('@/lib/core/config/redis', async () => {
-  const { redisConfigMock } = await import('@sim/testing')
-  return {
-    getConfiguredRedisUrl: () => {
-      if (mockRedisUrl.error) throw mockRedisUrl.error
-      return mockRedisUrl.value
-    },
-    getRedisConnectionDefaults: () => ({}),
-    // Records the inputs, then answers with the shared mirror of the real
-    // arithmetic so the derived budget here is the number production derives.
-    coldConnectionBudgetMs: (
-      inputs: Parameters<typeof redisConfigMock.coldConnectionBudgetMs>[0]
-    ) => {
-      budgetInputs.value = inputs
-      return redisConfigMock.coldConnectionBudgetMs(inputs)
-    },
-  }
-})
+vi.mock('@/lib/core/config/redis', () => ({
+  getConfiguredRedisUrl: () => {
+    if (mockRedisUrl.error) throw mockRedisUrl.error
+    return mockRedisUrl.value
+  },
+  // Realistic defaults: the readiness budget has a connect-deadline term.
+  getRedisConnectionDefaults: () => ({ connectTimeout: 10_000 }),
+}))
 
 import {
   getExecutionSignalHub,
   publishLocalExecutionSignal,
-  SUBSCRIBER_READY_TIMEOUT_MS,
   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
+    retryStrategy: (attempt: number) => number
+  }
+  return coldConnectionBudgetMs({
+    connectTimeoutMs: options.connectTimeout,
+    commandTimeoutMs: options.commandTimeout,
+    reconnectDelayMs: options.retryStrategy(1),
+  })
+}
+
 describe('ExecutionSignalHub', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -268,7 +267,7 @@ describe('ExecutionSignalHub', () => {
         'Timed out waiting for Redis subscriber readiness'
       )
 
-      const timeout = vi.advanceTimersByTimeAsync(SUBSCRIBER_READY_TIMEOUT_MS - 1000).then(() => {
+      const timeout = vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000).then(() => {
         connection.client?.emit('error', new Error('ECONNREFUSED'))
         return vi.advanceTimersByTimeAsync(1000)
       })
@@ -383,23 +382,49 @@ 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
+  it('waits exactly the budget derived from the options the subscriber is built with', async () => {
+    vi.useFakeTimers()
+    try {
+      connection.status = 'connect'
+      const hub = getExecutionSignalHub()
+      const subscription = hub.subscribe('execution-1', vi.fn())
+      const settled = vi.fn()
+      void subscription.then(settled, settled)
+
+      await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1)
+      expect(settled).not.toHaveBeenCalled()
+
+      await vi.advanceTimersByTimeAsync(1)
+      await expect(subscription).rejects.toThrow('Timed out waiting for Redis subscriber readiness')
+    } finally {
+      vi.useRealTimers()
     }
+  })
 
-    // One dead handshake, so the budget states the reconnect delay after
-    // attempt 1 as the client's own retryStrategy would return it.
-    expect(budgetInputs.value).toEqual({
-      commandTimeoutMs: options.commandTimeout,
-      retryDelaysMs: [options.retryStrategy(1)],
-    })
-    // 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('gives a subscribe that joins an in-flight warm-up 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.
+      await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000)
+      const subscription = getExecutionSignalHub().subscribe('execution-1', vi.fn())
+      const settled = vi.fn()
+      void subscription.then(settled, settled)
+
+      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()
+      expect(connection.client?.listenerCount('ready')).toBe(2)
+
+      connection.status = 'ready'
+      connection.client?.emit('ready')
+      await subscription
+      expect(mockSubscribe).toHaveBeenCalledOnce()
+    } finally {
+      vi.useRealTimers()
+    }
   })
 
   it('warms to true once the subscriber becomes ready', async () => {
@@ -418,7 +443,7 @@ describe('ExecutionSignalHub', () => {
       connection.status = 'connect'
       const warm = warmExecutionSignalHub()
 
-      await vi.advanceTimersByTimeAsync(SUBSCRIBER_READY_TIMEOUT_MS)
+      await vi.advanceTimersByTimeAsync(readyBudgetMs())
 
       await expect(warm).resolves.toBe(false)
       expect(vi.getTimerCount()).toBe(0)