From 7e6d7a7f3e42a486dc7768fb46a2a7bc86db32ab Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Thu, 10 Sep 2026 23:02:12 -0700
Subject: [PATCH 01/10] fix(redis): size cold-connection waits to survive a
 dead handshake
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 apps/sim/instrumentation-node.ts              |  12 +-
 apps/sim/lib/core/config/redis.test.ts        |  39 +++++++
 apps/sim/lib/core/config/redis.ts             |  51 +++++++--
 .../lib/execution/execution-signal.test.ts    | 103 ++++++++++++++++--
 apps/sim/lib/execution/execution-signal.ts    |  72 +++++++++++-
 apps/sim/trigger.config.ts                    |  20 ++--
 .../src/mocks/redis-config.mock.test.ts       |  12 ++
 .../testing/src/mocks/redis-config.mock.ts    |  21 ++++
 8 files changed, 295 insertions(+), 35 deletions(-)

diff --git a/apps/sim/instrumentation-node.ts b/apps/sim/instrumentation-node.ts
index b35aa21c931..fe48ce4cc50 100644
--- a/apps/sim/instrumentation-node.ts
+++ b/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()
 }
diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts
index bf02a57e48d..9911a54f2e3 100644
--- a/apps/sim/lib/core/config/redis.test.ts
+++ b/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'
 
diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts
index 2c11595df32..f95115b1fdf 100644
--- a/apps/sim/lib/core/config/redis.ts
+++ b/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,7 +333,7 @@ export function getRedisClient(): Redis | null {
 
     state.client = new Redis(redisUrl, {
       ...defaults,
-      commandTimeout: 5000,
+      commandTimeout: SHARED_COMMAND_TIMEOUT_MS,
       maxRetriesPerRequest: 5,
 
       retryStrategy: (times) => {
@@ -304,8 +341,8 @@ export function getRedisClient(): Redis | null {
           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 })
diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index 4b73887d294..b8e0ef34ad7 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/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()
diff --git a/apps/sim/lib/execution/execution-signal.ts b/apps/sim/lib/execution/execution-signal.ts
index c335f0123e9..c6cdec80c91 100644
--- a/apps/sim/lib/execution/execution-signal.ts
+++ b/apps/sim/lib/execution/execution-signal.ts
@@ -2,11 +2,31 @@ import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
 import Redis, { type RedisOptions } from 'ioredis'
-import { getConfiguredRedisUrl, getRedisConnectionDefaults } from '@/lib/core/config/redis'
+import {
+  coldConnectionBudgetMs,
+  getConfiguredRedisUrl,
+  getRedisConnectionDefaults,
+} from '@/lib/core/config/redis'
 
 const logger = createLogger('ExecutionSignalHub')
 const EXECUTION_SIGNAL_PREFIX = 'execution:signal:'
-const SUBSCRIBER_TIMEOUT_MS = 5000
+/**
+ * Tight, because this client only ever issues `SUBSCRIBE`/`UNSUBSCRIBE`, and
+ * only once the connection is ready — sub-millisecond commands that never sit
+ * in the offline queue behind a handshake. Its main job is bounding how long a
+ * dead handshake takes to be diagnosed and torn down.
+ */
+const SUBSCRIBER_COMMAND_TIMEOUT_MS = 2_000
+const subscriberRetryDelayMs = (attempt: number): number => Math.min(attempt * 500, 5000)
+/**
+ * Room for two dead handshakes and then a healthy one, so ioredis's own
+ * reconnect can be what rescues a stalled connection instead of the wait
+ * expiring while the first attempt is still being diagnosed.
+ */
+export const SUBSCRIBER_READY_TIMEOUT_MS = coldConnectionBudgetMs({
+  commandTimeoutMs: SUBSCRIBER_COMMAND_TIMEOUT_MS,
+  retryDelaysMs: [1, 2].map(subscriberRetryDelayMs),
+})
 export const LEGACY_EXECUTION_CANCEL_CHANNEL = 'execution:cancel'
 
 export type ExecutionSignalReason = 'event' | 'cancelled' | 'reconnected' | 'unavailable'
@@ -19,6 +39,8 @@ interface ChannelSubscription {
 
 export interface ExecutionSignalHub {
   subscribe(executionId: string, handler: ExecutionSignalHandler): Promise<() => void>
+  /** Resolves `true` once the hub can deliver signals, `false` if that could not be established in time. Never rejects. */
+  warm(): Promise<boolean>
 }
 
 export function getExecutionSignalChannel(executionId: string): string {
@@ -35,10 +57,10 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
   constructor(redisUrl: string) {
     const options = {
       ...getRedisConnectionDefaults(redisUrl),
-      commandTimeout: SUBSCRIBER_TIMEOUT_MS,
+      commandTimeout: SUBSCRIBER_COMMAND_TIMEOUT_MS,
       connectionName: 'execution-signal-hub',
       maxRetriesPerRequest: null,
-      retryStrategy: (attempt: number) => Math.min(attempt * 500, 5000),
+      retryStrategy: subscriberRetryDelayMs,
     } satisfies RedisOptions
     this.subscriber = new Redis(redisUrl, options)
     this.subscriber.on('message', (channel: string, message: string) => {
@@ -109,6 +131,24 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     }
   }
 
+  async warm(): Promise<boolean> {
+    try {
+      while (this.subscriber.status !== 'ready') {
+        await this.waitForConnectionReady()
+      }
+      return true
+    } catch (error) {
+      logger.warn(
+        'Execution signal subscriber warm-up gave up; first subscribe will pay the handshake',
+        {
+          error: toError(error).message,
+          status: this.subscriber.status,
+        }
+      )
+      return false
+    }
+  }
+
   private createSubscription(channels: string[]): ChannelSubscription {
     const subscription: ChannelSubscription = {
       acknowledged: false,
@@ -154,7 +194,7 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
       const onEnd = () => fail(new Error('Redis subscriber connection ended'))
       const timeout = setTimeout(
         () => fail(new Error('Timed out waiting for Redis subscriber readiness')),
-        SUBSCRIBER_TIMEOUT_MS
+        SUBSCRIBER_READY_TIMEOUT_MS
       )
       this.subscriber.once('ready', onReady)
       this.subscriber.once('end', onEnd)
@@ -212,6 +252,10 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
 class LocalExecutionSignalHub implements ExecutionSignalHub {
   private readonly handlers = new Map<string, Set<ExecutionSignalHandler>>()
 
+  warm(): Promise<boolean> {
+    return Promise.resolve(true)
+  }
+
   async subscribe(executionId: string, handler: ExecutionSignalHandler): Promise<() => void> {
     const channel = getExecutionSignalChannel(executionId)
     let channelHandlers = this.handlers.get(channel)
@@ -261,6 +305,24 @@ export function getExecutionSignalHub(): ExecutionSignalHub {
   return executionSignalGlobal._executionSignalHub
 }
 
+/**
+ * Establishes the hub's subscriber connection ahead of the first execution, so
+ * a run's cancellation subscription does not pay the handshake inside its own
+ * readiness budget. Never throws: this runs from process start-up hooks where
+ * a throw would fail the run, and a cold hub is only slower, not wrong.
+ */
+export async function warmExecutionSignalHub(): Promise<boolean> {
+  let hub: ExecutionSignalHub
+  try {
+    hub = getExecutionSignalHub()
+  } catch {
+    // A misconfigured URL belongs to the first real subscriber, which can report
+    // it against the execution that needed signals.
+    return false
+  }
+  return hub.warm()
+}
+
 export function publishLocalExecutionSignal(
   executionId: string,
   reason: Extract<ExecutionSignalReason, 'event' | 'cancelled'>
diff --git a/apps/sim/trigger.config.ts b/apps/sim/trigger.config.ts
index 6a218121ee5..9e9c421bbd5 100644
--- a/apps/sim/trigger.config.ts
+++ b/apps/sim/trigger.config.ts
@@ -102,18 +102,24 @@ export default defineConfig({
    * environment variables whether Trigger.dev is available: a process that
    * Trigger.dev is executing has Trigger.dev available by definition.
    *
-   * Also warms the shared Redis connection, because a run's first Redis call is
-   * typically a lock acquire and would otherwise pay the handshake inside its
-   * own command deadline. Awaited so the connection is up before `run()` issues
-   * anything; imported dynamically so deploy-time evaluation of this config does
-   * not pull the client, and never throwing because a throw here fails the run.
+   * Also warms both Redis connections a run opens — the shared client, whose
+   * first call is typically a lock acquire, and the execution-signal subscriber,
+   * whose first call is the cancellation subscription — because each would
+   * otherwise pay its handshake inside its own deadline. Warmed in parallel so
+   * the run waits for the slower of the two, not the sum; awaited so both are up
+   * before `run()` issues anything; imported dynamically so deploy-time
+   * evaluation of this config does not pull the clients; and neither ever
+   * throws, because a throw here fails the run.
    *
    * @see https://trigger.dev/docs/config/config-file#lifecycle-functions
    */
   init: async () => {
     markInsideTriggerRun()
-    const { warmRedisConnection } = await import('./lib/core/config/redis')
-    await warmRedisConnection()
+    const [{ warmRedisConnection }, { warmExecutionSignalHub }] = await Promise.all([
+      import('./lib/core/config/redis'),
+      import('./lib/execution/execution-signal'),
+    ])
+    await Promise.all([warmRedisConnection(), warmExecutionSignalHub()])
   },
   ...(grafanaTelemetry ? { telemetry: grafanaTelemetry } : {}),
   build: {
diff --git a/packages/testing/src/mocks/redis-config.mock.test.ts b/packages/testing/src/mocks/redis-config.mock.test.ts
index 0db6bbdd54a..91b4f4b09fc 100644
--- a/packages/testing/src/mocks/redis-config.mock.test.ts
+++ b/packages/testing/src/mocks/redis-config.mock.test.ts
@@ -23,6 +23,18 @@ describe('redis-config mock', () => {
     })
   })
 
+  it('mirrors the real cold-connection budget arithmetic', () => {
+    // Two command deadlines per dead handshake, its reconnect delay, then a
+    // healthy-handshake allowance — the same formula the real module uses, so
+    // a budget derived under this mock is the number production would derive.
+    expect(
+      redisConfigMock.coldConnectionBudgetMs({
+        commandTimeoutMs: 2_000,
+        retryDelaysMs: [500, 1_000],
+      })
+    ).toBe(2 * 2_000 + 500 + 2 * 2_000 + 1_000 + 1_000)
+  })
+
   it('resetRedisConfigMock restores defaults after overrides', async () => {
     const fakeClient = { ping: () => 'PONG' }
     redisConfigMockFns.mockGetConfiguredRedisUrl.mockReturnValue('redis://localhost:6379')
diff --git a/packages/testing/src/mocks/redis-config.mock.ts b/packages/testing/src/mocks/redis-config.mock.ts
index 9e5e548f2c8..d3c255bce0e 100644
--- a/packages/testing/src/mocks/redis-config.mock.ts
+++ b/packages/testing/src/mocks/redis-config.mock.ts
@@ -43,6 +43,22 @@ function getRedisConnectionDefaultsImpl(url?: string): {
   }
 }
 
+/**
+ * Mirrors the real `coldConnectionBudgetMs`: pure arithmetic with no I/O, and
+ * evaluated at module load by consumers deriving their readiness budgets, so
+ * the mock has to answer it for those modules to import at all.
+ */
+function coldConnectionBudgetMsImpl(options: {
+  commandTimeoutMs: number
+  retryDelaysMs: readonly number[]
+}): number {
+  const recovery = options.retryDelaysMs.reduce(
+    (total, retryDelayMs) => total + 2 * options.commandTimeoutMs + retryDelayMs,
+    0
+  )
+  return recovery + 1_000
+}
+
 /**
  * Mirrors the real `describeRedisConnection` under its Redis-unavailable
  * default: no client, no lifecycle history, and nothing derivable from an
@@ -91,6 +107,7 @@ export const redisConfigMockFns = {
   mockCloseRedisConnection: vi.fn().mockResolvedValue(undefined),
   mockResetForTesting: vi.fn(),
   mockDescribeRedisConnection: vi.fn(describeRedisConnectionImpl),
+  mockColdConnectionBudgetMs: vi.fn(coldConnectionBudgetMsImpl),
 }
 
 /**
@@ -108,6 +125,9 @@ export function resetRedisConfigMock(): void {
   redisConfigMockFns.mockExtendLock.mockReset().mockResolvedValue(true)
   redisConfigMockFns.mockCloseRedisConnection.mockReset().mockResolvedValue(undefined)
   redisConfigMockFns.mockResetForTesting.mockReset()
+  redisConfigMockFns.mockColdConnectionBudgetMs
+    .mockReset()
+    .mockImplementation(coldConnectionBudgetMsImpl)
   redisConfigMockFns.mockDescribeRedisConnection
     .mockReset()
     .mockImplementation(describeRedisConnectionImpl)
@@ -133,4 +153,5 @@ export const redisConfigMock = {
   closeRedisConnection: redisConfigMockFns.mockCloseRedisConnection,
   resetForTesting: redisConfigMockFns.mockResetForTesting,
   describeRedisConnection: redisConfigMockFns.mockDescribeRedisConnection,
+  coldConnectionBudgetMs: redisConfigMockFns.mockColdConnectionBudgetMs,
 }

From f18d24589e7991899c504b70939109ac7d2b407d Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Thu, 10 Sep 2026 23:08:59 -0700
Subject: [PATCH 02/10] fix(redis): keep the subscriber's live command deadline
 at 5s
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ioredis has one `commandTimeout` for the handshake commands and every live
command on the socket, so tightening it to diagnose dead handshakes faster also
cut the initial SUBSCRIBE from 5s to 2s — on the one path where a rejection
fails the whole run. A ready-but-slow server delays SUBSCRIBE too.

Restore the 5s tolerance that subscribe has always had and derive the
readiness budget from it: one dead handshake recovers in ~10.5s, so an 11.5s
budget still lets ioredis's reconnect rescue the case that failed. The local
test mock now delegates to the shared, drift-guarded mirror of the budget
arithmetic so the derived number under test is the one production derives.
---
 .../lib/execution/execution-signal.test.ts    | 37 +++++++++++--------
 apps/sim/lib/execution/execution-signal.ts    | 14 +++----
 2 files changed, 28 insertions(+), 23 deletions(-)

diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index b8e0ef34ad7..5eed000c9f8 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/apps/sim/lib/execution/execution-signal.test.ts
@@ -40,19 +40,24 @@ vi.mock('ioredis', () => ({
   },
 }))
 
-vi.mock('@/lib/core/config/redis', () => ({
-  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
-  },
-}))
+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)
+    },
+  }
+})
 
 import {
   getExecutionSignalHub,
@@ -385,11 +390,11 @@ describe('ExecutionSignalHub', () => {
       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.
+    // 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), options.retryStrategy(2)],
+      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(
diff --git a/apps/sim/lib/execution/execution-signal.ts b/apps/sim/lib/execution/execution-signal.ts
index c6cdec80c91..4159ac76602 100644
--- a/apps/sim/lib/execution/execution-signal.ts
+++ b/apps/sim/lib/execution/execution-signal.ts
@@ -11,21 +11,21 @@ import {
 const logger = createLogger('ExecutionSignalHub')
 const EXECUTION_SIGNAL_PREFIX = 'execution:signal:'
 /**
- * Tight, because this client only ever issues `SUBSCRIBE`/`UNSUBSCRIBE`, and
- * only once the connection is ready — sub-millisecond commands that never sit
- * in the offline queue behind a handshake. Its main job is bounding how long a
- * dead handshake takes to be diagnosed and torn down.
+ * Bounds the live `SUBSCRIBE` as well as the handshake commands — ioredis has
+ * one deadline for both — and an initial subscribe that rejects fails the run,
+ * so this stays at the tolerance a ready-but-slow server has always been
+ * given rather than being tightened to diagnose dead handshakes faster.
  */
-const SUBSCRIBER_COMMAND_TIMEOUT_MS = 2_000
+const SUBSCRIBER_COMMAND_TIMEOUT_MS = 5_000
 const subscriberRetryDelayMs = (attempt: number): number => Math.min(attempt * 500, 5000)
 /**
- * Room for two dead handshakes and then a healthy one, so ioredis's own
+ * Room for one dead handshake and then a healthy one, so ioredis's own
  * reconnect can be what rescues a stalled connection instead of the wait
  * expiring while the first attempt is still being diagnosed.
  */
 export const SUBSCRIBER_READY_TIMEOUT_MS = coldConnectionBudgetMs({
   commandTimeoutMs: SUBSCRIBER_COMMAND_TIMEOUT_MS,
-  retryDelaysMs: [1, 2].map(subscriberRetryDelayMs),
+  retryDelaysMs: [subscriberRetryDelayMs(1)],
 })
 export const LEGACY_EXECUTION_CANCEL_CHANNEL = 'execution:cancel'
 

From 3e60fcb4ba83be783151813b3e56b5e018c0a49f Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Thu, 10 Sep 2026 23:53:31 -0700
Subject: [PATCH 03/10] improvement(redis): warm the signal subscriber on
 intent, not at worker start
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Only the tasks that execute a workflow ever subscribe to cancellation signals,
and they are a minority of runs: over a week, 71% of Trigger.dev runs never
subscribe, and a single document-processing task is 61% on its own. Warming
the subscriber in the global init hook opened a TLS connection on every one
of those runs that nothing would use.

Warm it instead at `executeWorkflowCore` — the one path every execution
shares — fire-and-forget as the first statement, so the handshake overlaps the
custom-block read and preprocessing ahead of the cancellation subscribe rather
than being paid inside that subscribe's readiness budget. No task-id list to
keep in sync: coverage follows from the funnel. The subscribe and the warm-up
share the hub's memoized readiness wait, so their budgets never compound
within a run.

The global init keeps warming only the shared client, which nearly every task
uses; the Next server keeps warming both at boot, one connection per
long-lived process.
---
 apps/sim/lib/execution/execution-signal.ts    | 11 +++++---
 .../workflows/executor/execution-core.test.ts | 25 +++++++++++++++++++
 .../lib/workflows/executor/execution-core.ts  |  9 +++++++
 apps/sim/trigger.config.ts                    | 22 ++++++++--------
 4 files changed, 51 insertions(+), 16 deletions(-)

diff --git a/apps/sim/lib/execution/execution-signal.ts b/apps/sim/lib/execution/execution-signal.ts
index 4159ac76602..bf633621c36 100644
--- a/apps/sim/lib/execution/execution-signal.ts
+++ b/apps/sim/lib/execution/execution-signal.ts
@@ -306,10 +306,13 @@ export function getExecutionSignalHub(): ExecutionSignalHub {
 }
 
 /**
- * Establishes the hub's subscriber connection ahead of the first execution, so
- * a run's cancellation subscription does not pay the handshake inside its own
- * readiness budget. Never throws: this runs from process start-up hooks where
- * a throw would fail the run, and a cold hub is only slower, not wrong.
+ * Establishes the hub's subscriber connection ahead of a cancellation
+ * subscription, so that subscribe does not pay the handshake inside its own
+ * readiness budget. Called fire-and-forget at the execution entry point — the
+ * one path every execution shares, early enough to overlap the work ahead of
+ * the subscribe — and at boot in long-lived servers. Never throws or rejects:
+ * a throw from a start-up hook would fail the run, and a cold hub is only
+ * slower, not wrong.
  */
 export async function warmExecutionSignalHub(): Promise<boolean> {
   let hub: ExecutionSignalHub
diff --git a/apps/sim/lib/workflows/executor/execution-core.test.ts b/apps/sim/lib/workflows/executor/execution-core.test.ts
index b5bd98d9446..dafdcd77905 100644
--- a/apps/sim/lib/workflows/executor/execution-core.test.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.test.ts
@@ -97,6 +97,14 @@ vi.mock('@/lib/execution/cancellation', () => ({
   clearExecutionCancellation: clearExecutionCancellationMock,
 }))
 
+const { warmExecutionSignalHubMock } = vi.hoisted(() => ({
+  warmExecutionSignalHubMock: vi.fn(),
+}))
+
+vi.mock('@/lib/execution/execution-signal', () => ({
+  warmExecutionSignalHub: warmExecutionSignalHubMock,
+}))
+
 vi.mock('@/lib/core/security/encryption', () => ({
   decryptSecret: decryptSecretMock,
 }))
@@ -376,6 +384,23 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     expect(executorConstructorMock).toHaveBeenCalledTimes(1)
   })
 
+  it('begins warming the signal subscriber synchronously, before the first await', async () => {
+    warmExecutionSignalHubMock.mockResolvedValue(true)
+
+    const executionPromise = executeWorkflowCore({
+      snapshot: createSnapshot() as any,
+      callbacks: {},
+      loggingSession: loggingSession as any,
+    })
+
+    // Asserted with no await in between: the handshake has to start ahead of
+    // the custom-block read, or it stops overlapping the work that precedes the
+    // cancellation subscribe and is paid inside that subscribe's budget instead.
+    expect(warmExecutionSignalHubMock).toHaveBeenCalledOnce()
+
+    await executionPromise
+  })
+
   it('routes onBlockStart through logging session persistence path', async () => {
     executorExecuteMock.mockResolvedValue({
       success: true,
diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts
index 82ba64adbc1..1e121108fd8 100644
--- a/apps/sim/lib/workflows/executor/execution-core.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.ts
@@ -22,6 +22,7 @@ import {
 import { withDatabaseReadRetry } from '@/lib/db/read-retry'
 import { getExecutionEnvironment } from '@/lib/environment/utils'
 import { clearExecutionCancellation } from '@/lib/execution/cancellation'
+import { warmExecutionSignalHub } from '@/lib/execution/execution-signal'
 import { warmLargeValueRefs } from '@/lib/execution/payloads/hydration'
 import { parseLargeExecutionValue } from '@/lib/execution/payloads/large-execution-value'
 import type { LoggingSession } from '@/lib/logs/execution/logging-session'
@@ -381,6 +382,14 @@ async function finalizeExecutionError(params: {
 export async function executeWorkflowCore(
   options: ExecuteWorkflowCoreOptions
 ): Promise<ExecutionResult> {
+  // First, and not awaited: every execution subscribes to cancellation signals
+  // once its engine starts, so the subscriber's handshake is begun here — the
+  // one path all of them share — and overlaps the reads and preprocessing
+  // ahead of that subscribe instead of being paid inside its readiness budget.
+  // Warming on intent rather than at worker start keeps the tasks that never
+  // execute a workflow, most of the fleet by volume, from opening a connection
+  // they would never use.
+  void warmExecutionSignalHub()
   const workspaceId = options.snapshot.metadata.workspaceId
   const rows = workspaceId
     ? await withDatabaseReadRetry(() => getCustomBlockRowsForWorkspace(workspaceId), {
diff --git a/apps/sim/trigger.config.ts b/apps/sim/trigger.config.ts
index 9e9c421bbd5..7966fdbf94a 100644
--- a/apps/sim/trigger.config.ts
+++ b/apps/sim/trigger.config.ts
@@ -102,24 +102,22 @@ export default defineConfig({
    * environment variables whether Trigger.dev is available: a process that
    * Trigger.dev is executing has Trigger.dev available by definition.
    *
-   * Also warms both Redis connections a run opens — the shared client, whose
-   * first call is typically a lock acquire, and the execution-signal subscriber,
-   * whose first call is the cancellation subscription — because each would
-   * otherwise pay its handshake inside its own deadline. Warmed in parallel so
-   * the run waits for the slower of the two, not the sum; awaited so both are up
+   * Also warms the shared Redis connection, because nearly every task's first
+   * Redis call — a lock acquire, a usage reservation — would otherwise pay the
+   * handshake inside its own command deadline. Awaited so the connection is up
    * before `run()` issues anything; imported dynamically so deploy-time
-   * evaluation of this config does not pull the clients; and neither ever
-   * throws, because a throw here fails the run.
+   * evaluation of this config does not pull the client; and never throwing,
+   * because a throw here fails the run. The execution-signal subscriber is
+   * deliberately not warmed here: only the tasks that execute a workflow ever
+   * subscribe, and they are a minority of runs, so that connection is warmed
+   * on intent at the execution entry point instead.
    *
    * @see https://trigger.dev/docs/config/config-file#lifecycle-functions
    */
   init: async () => {
     markInsideTriggerRun()
-    const [{ warmRedisConnection }, { warmExecutionSignalHub }] = await Promise.all([
-      import('./lib/core/config/redis'),
-      import('./lib/execution/execution-signal'),
-    ])
-    await Promise.all([warmRedisConnection(), warmExecutionSignalHub()])
+    const { warmRedisConnection } = await import('./lib/core/config/redis')
+    await warmRedisConnection()
   },
   ...(grafanaTelemetry ? { telemetry: grafanaTelemetry } : {}),
   build: {

From c64d1c278f07509f1dc2d0ba72a23e3e6cdce8f5 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Thu, 10 Sep 2026 23:58:41 -0700
Subject: [PATCH 04/10] chore(instrumentation): import the warm-ups through the
 path alias

The dynamic imports in the Redis warm-up block used relative paths; the file
already resolves `@/` and the repository rule is absolute imports.
---
 apps/sim/instrumentation-node.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/sim/instrumentation-node.ts b/apps/sim/instrumentation-node.ts
index fe48ce4cc50..9d20839cde9 100644
--- a/apps/sim/instrumentation-node.ts
+++ b/apps/sim/instrumentation-node.ts
@@ -408,8 +408,8 @@ export async function register() {
   // 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'),
+    import('@/lib/core/config/redis'),
+    import('@/lib/execution/execution-signal'),
   ])
   void warmRedisConnection()
   void warmExecutionSignalHub()

From acf01a92c0034007b0cf4f5385af938833c817b5 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Fri, 11 Sep 2026 00:13:41 -0700
Subject: [PATCH 05/10] fix(redis): give every readiness waiter its own
 deadline and correct the budget model
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 apps/sim/instrumentation-node.ts              |  12 +-
 apps/sim/lib/core/config/redis.test.ts        |  36 ++---
 apps/sim/lib/core/config/redis.ts             |  62 ++++----
 .../lib/execution/execution-signal.test.ts    | 133 +++++++++++-------
 apps/sim/lib/execution/execution-signal.ts    | 109 +++++++++-----
 .../lib/workflows/executor/execution-core.ts  |  15 +-
 .../src/mocks/redis-config.mock.test.ts       |  12 --
 .../testing/src/mocks/redis-config.mock.ts    |  27 +---
 packages/utils/src/retry.test.ts              |  33 +++++
 packages/utils/src/retry.ts                   |  39 +++++
 10 files changed, 288 insertions(+), 190 deletions(-)
 create mode 100644 packages/utils/src/retry.test.ts

diff --git a/apps/sim/instrumentation-node.ts b/apps/sim/instrumentation-node.ts
index 9d20839cde9..d9a3668d304 100644
--- a/apps/sim/instrumentation-node.ts
+++ b/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()
 }
diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts
index 9911a54f2e3..ce945386a92 100644
--- a/apps/sim/lib/core/config/redis.test.ts
+++ b/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)
     })
diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts
index f95115b1fdf..63f1c6c25d4 100644
--- a/apps/sim/lib/core/config/redis.ts
+++ b/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
diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index 5eed000c9f8..af093bcf6f7 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/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)
diff --git a/apps/sim/lib/execution/execution-signal.ts b/apps/sim/lib/execution/execution-signal.ts
index bf633621c36..dcdc6ca3518 100644
--- a/apps/sim/lib/execution/execution-signal.ts
+++ b/apps/sim/lib/execution/execution-signal.ts
@@ -1,12 +1,9 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
+import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import Redis, { type RedisOptions } from 'ioredis'
-import {
-  coldConnectionBudgetMs,
-  getConfiguredRedisUrl,
-  getRedisConnectionDefaults,
-} from '@/lib/core/config/redis'
+import { getConfiguredRedisUrl, getRedisConnectionDefaults } from '@/lib/core/config/redis'
 
 const logger = createLogger('ExecutionSignalHub')
 const EXECUTION_SIGNAL_PREFIX = 'execution:signal:'
@@ -18,15 +15,6 @@ const EXECUTION_SIGNAL_PREFIX = 'execution:signal:'
  */
 const SUBSCRIBER_COMMAND_TIMEOUT_MS = 5_000
 const subscriberRetryDelayMs = (attempt: number): number => Math.min(attempt * 500, 5000)
-/**
- * Room for one dead handshake and then a healthy one, so ioredis's own
- * reconnect can be what rescues a stalled connection instead of the wait
- * expiring while the first attempt is still being diagnosed.
- */
-export const SUBSCRIBER_READY_TIMEOUT_MS = coldConnectionBudgetMs({
-  commandTimeoutMs: SUBSCRIBER_COMMAND_TIMEOUT_MS,
-  retryDelaysMs: [subscriberRetryDelayMs(1)],
-})
 export const LEGACY_EXECUTION_CANCEL_CHANNEL = 'execution:cancel'
 
 export type ExecutionSignalReason = 'event' | 'cancelled' | 'reconnected' | 'unavailable'
@@ -37,6 +25,12 @@ interface ChannelSubscription {
   acknowledged: boolean
 }
 
+/** Settles on the subscriber's next `ready` or `end`; `detach` stops listening for either. */
+interface ReadySignal {
+  promise: Promise<void>
+  detach: () => void
+}
+
 export interface ExecutionSignalHub {
   subscribe(executionId: string, handler: ExecutionSignalHandler): Promise<() => void>
   /** Resolves `true` once the hub can deliver signals, `false` if that could not be established in time. Never rejects. */
@@ -49,9 +43,21 @@ export function getExecutionSignalChannel(executionId: string): string {
 
 class RedisExecutionSignalHub implements ExecutionSignalHub {
   private readonly subscriber: Redis
+  /**
+   * How long any one caller waits for readiness: room for one dead attempt and
+   * then a healthy one, so ioredis's own reconnect can be what rescues a stalled
+   * connection instead of the wait expiring while the first attempt is still
+   * being diagnosed. Derived from the exact options the client is built with.
+   *
+   * Every `subscribe` pays this too, not only warm-up — including against a
+   * Redis that is simply unreachable, where the connect deadline is what runs
+   * out and nothing is being diagnosed. That is the cost of the recovery.
+   */
+  private readonly readyTimeoutMs: number
   private readonly handlers = new Map<string, Set<ExecutionSignalHandler>>()
   private readonly subscriptions = new Map<string, ChannelSubscription>()
-  private connectionReady: Promise<void> | undefined
+  private readySignal: ReadySignal | undefined
+  private readyWaiters = 0
   private connectedOnce = false
 
   constructor(redisUrl: string) {
@@ -62,6 +68,11 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
       maxRetriesPerRequest: null,
       retryStrategy: subscriberRetryDelayMs,
     } satisfies RedisOptions
+    this.readyTimeoutMs = coldConnectionBudgetMs({
+      connectTimeoutMs: options.connectTimeout,
+      commandTimeoutMs: options.commandTimeout,
+      reconnectDelayMs: subscriberRetryDelayMs(1),
+    })
     this.subscriber = new Redis(redisUrl, options)
     this.subscriber.on('message', (channel: string, message: string) => {
       if (channel === LEGACY_EXECUTION_CANCEL_CHANNEL) {
@@ -171,37 +182,67 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     await this.subscriber.subscribe(...channels)
   }
 
+  /**
+   * One readiness signal is shared by every waiter so the client carries a
+   * single `ready`/`end` listener pair, but each waiter runs its own deadline
+   * over it. A shared timer would hand a subscribe that joins an in-flight
+   * warm-up only the remainder of that warm-up's budget — down to nothing —
+   * where it used to be guaranteed a full wait of its own. The signal is torn
+   * down when its last waiter gives up, so a timeout leaves nothing attached.
+   */
   private waitForConnectionReady(): Promise<void> {
-    if (this.connectionReady) return this.connectionReady
     if (this.subscriber.status === 'end') {
       return Promise.reject(new Error('Redis subscriber connection ended'))
     }
-
-    this.connectionReady = new Promise<void>((resolve, reject) => {
-      const cleanup = () => {
+    const signal = (this.readySignal ??= this.createReadySignal())
+    this.readyWaiters++
+    return new Promise<void>((resolve, reject) => {
+      const leave = () => {
         clearTimeout(timeout)
-        this.subscriber.removeListener('ready', onReady)
-        this.subscriber.removeListener('end', onEnd)
+        if (--this.readyWaiters === 0 && this.readySignal === signal) {
+          signal.detach()
+          this.readySignal = undefined
+        }
       }
+      const timeout = setTimeout(() => {
+        leave()
+        reject(new Error('Timed out waiting for Redis subscriber readiness'))
+      }, this.readyTimeoutMs)
+      signal.promise.then(
+        () => {
+          leave()
+          resolve()
+        },
+        (error: unknown) => {
+          leave()
+          reject(error)
+        }
+      )
+    })
+  }
+
+  private createReadySignal(): ReadySignal {
+    let detach = () => {}
+    const promise = new Promise<void>((resolve, reject) => {
       const onReady = () => {
-        cleanup()
+        detach()
         resolve()
       }
-      const fail = (error: Error) => {
-        cleanup()
-        reject(error)
+      const onEnd = () => {
+        detach()
+        reject(new Error('Redis subscriber connection ended'))
+      }
+      detach = () => {
+        this.subscriber.removeListener('ready', onReady)
+        this.subscriber.removeListener('end', onEnd)
       }
-      const onEnd = () => fail(new Error('Redis subscriber connection ended'))
-      const timeout = setTimeout(
-        () => fail(new Error('Timed out waiting for Redis subscriber readiness')),
-        SUBSCRIBER_READY_TIMEOUT_MS
-      )
       this.subscriber.once('ready', onReady)
       this.subscriber.once('end', onEnd)
-    }).finally(() => {
-      this.connectionReady = undefined
     })
-    return this.connectionReady
+    // Detached before settling, this promise is simply dropped; an `end` that
+    // arrives after every waiter has left must not surface as unhandled.
+    promise.catch(() => undefined)
+    return { promise, detach }
   }
 
   private async handleReady(): Promise<void> {
@@ -319,8 +360,6 @@ export async function warmExecutionSignalHub(): Promise<boolean> {
   try {
     hub = getExecutionSignalHub()
   } catch {
-    // A misconfigured URL belongs to the first real subscriber, which can report
-    // it against the execution that needed signals.
     return false
   }
   return hub.warm()
diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts
index 1e121108fd8..0ce8af40b62 100644
--- a/apps/sim/lib/workflows/executor/execution-core.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.ts
@@ -378,17 +378,18 @@ async function finalizeExecutionError(params: {
  * the background job — puts `custom_block_*` types in scope for serialization,
  * execution, and any nested child-workflow serialization (ALS propagates to the
  * whole async subtree).
+ *
+ * Also begins warming the execution-signal subscriber first, without awaiting
+ * it: every execution subscribes to cancellation signals once its engine
+ * starts, so starting that handshake here — the one path all of them share —
+ * lets it overlap the reads and preprocessing ahead of the subscribe instead
+ * of being paid inside its readiness budget. Warming on intent rather than at
+ * worker start keeps the tasks that never execute a workflow, most of the
+ * fleet by volume, from opening a connection they would never use.
  */
 export async function executeWorkflowCore(
   options: ExecuteWorkflowCoreOptions
 ): Promise<ExecutionResult> {
-  // First, and not awaited: every execution subscribes to cancellation signals
-  // once its engine starts, so the subscriber's handshake is begun here — the
-  // one path all of them share — and overlaps the reads and preprocessing
-  // ahead of that subscribe instead of being paid inside its readiness budget.
-  // Warming on intent rather than at worker start keeps the tasks that never
-  // execute a workflow, most of the fleet by volume, from opening a connection
-  // they would never use.
   void warmExecutionSignalHub()
   const workspaceId = options.snapshot.metadata.workspaceId
   const rows = workspaceId
diff --git a/packages/testing/src/mocks/redis-config.mock.test.ts b/packages/testing/src/mocks/redis-config.mock.test.ts
index 91b4f4b09fc..0db6bbdd54a 100644
--- a/packages/testing/src/mocks/redis-config.mock.test.ts
+++ b/packages/testing/src/mocks/redis-config.mock.test.ts
@@ -23,18 +23,6 @@ describe('redis-config mock', () => {
     })
   })
 
-  it('mirrors the real cold-connection budget arithmetic', () => {
-    // Two command deadlines per dead handshake, its reconnect delay, then a
-    // healthy-handshake allowance — the same formula the real module uses, so
-    // a budget derived under this mock is the number production would derive.
-    expect(
-      redisConfigMock.coldConnectionBudgetMs({
-        commandTimeoutMs: 2_000,
-        retryDelaysMs: [500, 1_000],
-      })
-    ).toBe(2 * 2_000 + 500 + 2 * 2_000 + 1_000 + 1_000)
-  })
-
   it('resetRedisConfigMock restores defaults after overrides', async () => {
     const fakeClient = { ping: () => 'PONG' }
     redisConfigMockFns.mockGetConfiguredRedisUrl.mockReturnValue('redis://localhost:6379')
diff --git a/packages/testing/src/mocks/redis-config.mock.ts b/packages/testing/src/mocks/redis-config.mock.ts
index d3c255bce0e..4a8154e83a3 100644
--- a/packages/testing/src/mocks/redis-config.mock.ts
+++ b/packages/testing/src/mocks/redis-config.mock.ts
@@ -43,22 +43,6 @@ function getRedisConnectionDefaultsImpl(url?: string): {
   }
 }
 
-/**
- * Mirrors the real `coldConnectionBudgetMs`: pure arithmetic with no I/O, and
- * evaluated at module load by consumers deriving their readiness budgets, so
- * the mock has to answer it for those modules to import at all.
- */
-function coldConnectionBudgetMsImpl(options: {
-  commandTimeoutMs: number
-  retryDelaysMs: readonly number[]
-}): number {
-  const recovery = options.retryDelaysMs.reduce(
-    (total, retryDelayMs) => total + 2 * options.commandTimeoutMs + retryDelayMs,
-    0
-  )
-  return recovery + 1_000
-}
-
 /**
  * Mirrors the real `describeRedisConnection` under its Redis-unavailable
  * default: no client, no lifecycle history, and nothing derivable from an
@@ -107,7 +91,8 @@ export const redisConfigMockFns = {
   mockCloseRedisConnection: vi.fn().mockResolvedValue(undefined),
   mockResetForTesting: vi.fn(),
   mockDescribeRedisConnection: vi.fn(describeRedisConnectionImpl),
-  mockColdConnectionBudgetMs: vi.fn(coldConnectionBudgetMsImpl),
+  mockWarmRedisConnection: vi.fn().mockResolvedValue(false),
+  mockSharedReconnectDelayMs: vi.fn().mockReturnValue(1_000),
 }
 
 /**
@@ -125,9 +110,8 @@ export function resetRedisConfigMock(): void {
   redisConfigMockFns.mockExtendLock.mockReset().mockResolvedValue(true)
   redisConfigMockFns.mockCloseRedisConnection.mockReset().mockResolvedValue(undefined)
   redisConfigMockFns.mockResetForTesting.mockReset()
-  redisConfigMockFns.mockColdConnectionBudgetMs
-    .mockReset()
-    .mockImplementation(coldConnectionBudgetMsImpl)
+  redisConfigMockFns.mockWarmRedisConnection.mockReset().mockResolvedValue(false)
+  redisConfigMockFns.mockSharedReconnectDelayMs.mockReset().mockReturnValue(1_000)
   redisConfigMockFns.mockDescribeRedisConnection
     .mockReset()
     .mockImplementation(describeRedisConnectionImpl)
@@ -153,5 +137,6 @@ export const redisConfigMock = {
   closeRedisConnection: redisConfigMockFns.mockCloseRedisConnection,
   resetForTesting: redisConfigMockFns.mockResetForTesting,
   describeRedisConnection: redisConfigMockFns.mockDescribeRedisConnection,
-  coldConnectionBudgetMs: redisConfigMockFns.mockColdConnectionBudgetMs,
+  warmRedisConnection: redisConfigMockFns.mockWarmRedisConnection,
+  sharedReconnectDelayMs: redisConfigMockFns.mockSharedReconnectDelayMs,
 }
diff --git a/packages/utils/src/retry.test.ts b/packages/utils/src/retry.test.ts
new file mode 100644
index 00000000000..aa27f284756
--- /dev/null
+++ b/packages/utils/src/retry.test.ts
@@ -0,0 +1,33 @@
+import { describe, expect, it } from 'vitest'
+import { coldConnectionBudgetMs } from './retry'
+
+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.
+    expect(
+      coldConnectionBudgetMs({
+        connectTimeoutMs: 1_000,
+        commandTimeoutMs: 2_000,
+        reconnectDelayMs: 500,
+      })
+    ).toBe(2 * 2_000 + 500 + 1_000)
+  })
+
+  it('charges the connect deadline when a connection that never completes is the longer case', () => {
+    // Lowering the command deadline must not shrink the budget below what a
+    // connect that never completes costs before retryStrategy can fire.
+    expect(
+      coldConnectionBudgetMs({
+        connectTimeoutMs: 10_000,
+        commandTimeoutMs: 3_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 })
+    ).toBe(1_000)
+  })
+})
diff --git a/packages/utils/src/retry.ts b/packages/utils/src/retry.ts
index c3fe72461d7..dfe5cced729 100644
--- a/packages/utils/src/retry.ts
+++ b/packages/utils/src/retry.ts
@@ -60,3 +60,42 @@ export function parseRetryAfter(header: string | null, maxMs = RETRY_AFTER_MAX_M
   }
   return null
 }
+
+/** 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
+  /** What the client's `retryStrategy` returns for its first reconnect. */
+  reconnectDelayMs: number
+}
+
+/**
+ * How long a wait for a cold 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`. A
+ * connection that opens but whose handshake is never answered costs command
+ * deadlines: one when the URL carries a password, because ioredis treats a
+ * timed-out `AUTH` as fatal and tears the socket down; two when it does not,
+ * because `CLIENT SETNAME`/`SETINFO` must settle — by timing out — before the
+ * `INFO` ready check even starts its own. The budget takes the largest of those
+ * so it holds for either URL shape without parsing it. Only then does the
+ * client's `retryStrategy` delay 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.
+ */
+export function coldConnectionBudgetMs(options: ColdConnectionBudgetOptions): number {
+  const deadAttemptMs = Math.max(options.connectTimeoutMs, 2 * options.commandTimeoutMs)
+  return deadAttemptMs + options.reconnectDelayMs + HEALTHY_HANDSHAKE_ALLOWANCE_MS
+}

From 4221bf6610f745c0fe34e116036d8dca8c5cb27f Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Fri, 11 Sep 2026 00:18:48 -0700
Subject: [PATCH 06/10] fix(redis): leave the readiness wait exactly once per
 waiter
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A waiter that timed out was still subscribed to the shared readiness signal,
so when that signal later settled for another waiter its cleanup ran a second
time. The waiter count drifted negative, the last-waiter teardown could never
match again, and the settled signal stayed memoized — a later subscribe during
a reconnect would have observed a readiness that had already passed.

Each waiter now leaves exactly once, and a settling signal clears itself from
the memo so the next waiter observes the connection afresh regardless of the
count.
---
 .../lib/execution/execution-signal.test.ts    | 44 +++++++++++++++++++
 apps/sim/lib/execution/execution-signal.ts    | 22 +++++++---
 2 files changed, 61 insertions(+), 5 deletions(-)

diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index af093bcf6f7..4d2c68636b4 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/apps/sim/lib/execution/execution-signal.test.ts
@@ -427,6 +427,50 @@ describe('ExecutionSignalHub', () => {
     }
   })
 
+  it('keeps its waiter accounting exact when one waiter times out before the signal settles', async () => {
+    vi.useFakeTimers()
+    try {
+      connection.status = 'connect'
+      const hub = getExecutionSignalHub()
+      // Waiter A will time out; waiter B, started later, is still waiting when it does.
+      const early = hub.subscribe('execution-early', vi.fn())
+      const earlySettled = vi.fn()
+      void early.then(earlySettled, earlySettled)
+      await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000)
+      const late = hub.subscribe('execution-late', vi.fn())
+      await vi.advanceTimersByTimeAsync(1000)
+      await expect(early).rejects.toThrow('Timed out waiting for Redis subscriber readiness')
+
+      // The signal settles for B — and must not run A's cleanup a second time.
+      connection.status = 'ready'
+      connection.client?.emit('ready')
+      await late
+      expect(connection.client?.listenerCount('ready')).toBe(1)
+      expect(connection.client?.listenerCount('end')).toBe(0)
+
+      // Connection drops again: a new subscribe must wait for a fresh ready,
+      // not reuse a readiness that has already passed.
+      connection.status = 'connect'
+      connection.client?.emit('close')
+      mockSubscribe.mockClear()
+      const again = hub.subscribe('execution-again', vi.fn())
+      const againSettled = vi.fn()
+      void again.then(againSettled, againSettled)
+      await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1)
+      expect(againSettled).not.toHaveBeenCalled()
+      expect(mockSubscribe).not.toHaveBeenCalled()
+
+      connection.status = 'ready'
+      connection.client?.emit('ready')
+      await again
+      expect(mockSubscribe).toHaveBeenCalledOnce()
+      expect(connection.client?.listenerCount('ready')).toBe(1)
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('warms to true once the subscriber becomes ready', async () => {
     connection.status = 'connecting'
     const warm = warmExecutionSignalHub()
diff --git a/apps/sim/lib/execution/execution-signal.ts b/apps/sim/lib/execution/execution-signal.ts
index dcdc6ca3518..033387f56ba 100644
--- a/apps/sim/lib/execution/execution-signal.ts
+++ b/apps/sim/lib/execution/execution-signal.ts
@@ -197,7 +197,12 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     const signal = (this.readySignal ??= this.createReadySignal())
     this.readyWaiters++
     return new Promise<void>((resolve, reject) => {
+      // A waiter that timed out is still subscribed to the signal, so it must
+      // leave exactly once whichever of its deadline or the signal fires first.
+      let left = false
       const leave = () => {
+        if (left) return
+        left = true
         clearTimeout(timeout)
         if (--this.readyWaiters === 0 && this.readySignal === signal) {
           signal.detach()
@@ -223,13 +228,20 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
 
   private createReadySignal(): ReadySignal {
     let detach = () => {}
-    const promise = new Promise<void>((resolve, reject) => {
+    const signal: ReadySignal = { promise: Promise.resolve(), detach: () => detach() }
+    // A settled signal describes a moment that has passed; the next waiter must
+    // observe the connection afresh rather than a readiness that may be gone.
+    const settle = () => {
+      detach()
+      if (this.readySignal === signal) this.readySignal = undefined
+    }
+    signal.promise = new Promise<void>((resolve, reject) => {
       const onReady = () => {
-        detach()
+        settle()
         resolve()
       }
       const onEnd = () => {
-        detach()
+        settle()
         reject(new Error('Redis subscriber connection ended'))
       }
       detach = () => {
@@ -241,8 +253,8 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     })
     // Detached before settling, this promise is simply dropped; an `end` that
     // arrives after every waiter has left must not surface as unhandled.
-    promise.catch(() => undefined)
-    return { promise, detach }
+    signal.promise.catch(() => undefined)
+    return signal
   }
 
   private async handleReady(): Promise<void> {

From 84116a8d5727af9ccb80c5de1da9013078225082 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Fri, 11 Sep 2026 00:21:01 -0700
Subject: [PATCH 07/10] test(redis): assert the fresh subscribe by channel, not
 by call count

A reconnect also re-subscribes surviving channels, so the new subscription is
identified by its channel rather than by SUBSCRIBE having been issued once.
---
 apps/sim/lib/execution/execution-signal.test.ts | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index 4d2c68636b4..15ecb5e6528 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/apps/sim/lib/execution/execution-signal.test.ts
@@ -463,7 +463,12 @@ describe('ExecutionSignalHub', () => {
       connection.status = 'ready'
       connection.client?.emit('ready')
       await again
-      expect(mockSubscribe).toHaveBeenCalledOnce()
+      // The reconnect also re-subscribes the surviving channel; what matters is
+      // that the new one went out only once readiness was genuinely observed.
+      expect(mockSubscribe).toHaveBeenCalledWith(
+        'execution:signal:execution-again',
+        'execution:cancel'
+      )
       expect(connection.client?.listenerCount('ready')).toBe(1)
       expect(vi.getTimerCount()).toBe(0)
     } finally {

From 7730723c241535787dd9f7d39c88c1becab1fc28 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Fri, 11 Sep 2026 00:23:28 -0700
Subject: [PATCH 08/10] test(redis): prove waiter accounting by the teardown a
 timed-out waiter must perform

The settle path clears the memo on its own, so a drifted waiter count only
shows when a later lone waiter times out and fails to tear the signal down.
Exercise that path directly; the test now fails without the per-waiter guard.
---
 apps/sim/lib/execution/execution-signal.test.ts | 14 +++++---------
 1 file changed, 5 insertions(+), 9 deletions(-)

diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index 15ecb5e6528..f16dc0e584b 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/apps/sim/lib/execution/execution-signal.test.ts
@@ -460,16 +460,12 @@ describe('ExecutionSignalHub', () => {
       expect(againSettled).not.toHaveBeenCalled()
       expect(mockSubscribe).not.toHaveBeenCalled()
 
-      connection.status = 'ready'
-      connection.client?.emit('ready')
-      await again
-      // The reconnect also re-subscribes the surviving channel; what matters is
-      // that the new one went out only once readiness was genuinely observed.
-      expect(mockSubscribe).toHaveBeenCalledWith(
-        'execution:signal:execution-again',
-        'execution:cancel'
-      )
+      // And when that lone waiter gives up, it must be the one that tears the
+      // signal down — which only holds if every earlier waiter left exactly once.
+      await vi.advanceTimersByTimeAsync(1)
+      await expect(again).rejects.toThrow('Timed out waiting for Redis subscriber readiness')
       expect(connection.client?.listenerCount('ready')).toBe(1)
+      expect(connection.client?.listenerCount('end')).toBe(0)
       expect(vi.getTimerCount()).toBe(0)
     } finally {
       vi.useRealTimers()

From 39e039a2c17dd596b37676392854e2abd7be1b4f Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Fri, 11 Sep 2026 00:28:38 -0700
Subject: [PATCH 09/10] test(execution-core): type the warm-up test's fixtures
 instead of casting to any

The partial fixtures are widened through unknown to the real ExecutionSnapshot
and LoggingSession types, as the repository's TypeScript rule prescribes.
---
 apps/sim/lib/workflows/executor/execution-core.test.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/sim/lib/workflows/executor/execution-core.test.ts b/apps/sim/lib/workflows/executor/execution-core.test.ts
index dafdcd77905..917b452dde8 100644
--- a/apps/sim/lib/workflows/executor/execution-core.test.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.test.ts
@@ -388,9 +388,9 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     warmExecutionSignalHubMock.mockResolvedValue(true)
 
     const executionPromise = executeWorkflowCore({
-      snapshot: createSnapshot() as any,
+      snapshot: createSnapshot() as unknown as ExecutionSnapshot,
       callbacks: {},
-      loggingSession: loggingSession as any,
+      loggingSession: loggingSession as unknown as LoggingSession,
     })
 
     // Asserted with no await in between: the handshake has to start ahead of

From b91b90a898c9c11d3a6eb3c4e6cfd9ac76bff9ae Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Fri, 11 Sep 2026 11:19:21 -0700
Subject: [PATCH 10/10] fix(redis): connect the signal subscriber on intent and
 complete the budget model
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 .../sim/lib/core/config/redis-budget.test.ts  |  23 ++-
 apps/sim/lib/core/config/redis-budget.ts      |  45 ++++++
 apps/sim/lib/core/config/redis.test.ts        |  19 ++-
 apps/sim/lib/core/config/redis.ts             |  11 +-
 .../lib/execution/execution-signal.test.ts    |  76 ++++-----
 apps/sim/lib/execution/execution-signal.ts    | 148 +++++++-----------
 .../workflows/executor/execution-core.test.ts |  12 +-
 .../lib/workflows/executor/execution-core.ts  |  18 +--
 .../src/mocks/redis-config.mock.test.ts       |   1 +
 .../testing/src/mocks/redis-config.mock.ts    |   5 +
 packages/utils/src/retry.ts                   |  39 -----
 11 files changed, 192 insertions(+), 205 deletions(-)
 rename packages/utils/src/retry.test.ts => apps/sim/lib/core/config/redis-budget.test.ts (53%)
 create mode 100644 apps/sim/lib/core/config/redis-budget.ts

diff --git a/packages/utils/src/retry.test.ts b/apps/sim/lib/core/config/redis-budget.test.ts
similarity index 53%
rename from packages/utils/src/retry.test.ts
rename to apps/sim/lib/core/config/redis-budget.test.ts
index aa27f284756..55c786d3514 100644
--- a/packages/utils/src/retry.test.ts
+++ b/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,6 +25,7 @@ describe('coldConnectionBudgetMs', () => {
       coldConnectionBudgetMs({
         connectTimeoutMs: 10_000,
         commandTimeoutMs: 3_000,
+        disconnectTimeoutMs: 2_000,
         reconnectDelayMs: 500,
       })
     ).toBe(10_000 + 500 + 1_000)
@@ -27,7 +33,12 @@ describe('coldConnectionBudgetMs', () => {
 
   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)
   })
 })
diff --git a/apps/sim/lib/core/config/redis-budget.ts b/apps/sim/lib/core/config/redis-budget.ts
new file mode 100644
index 00000000000..f9743251a4b
--- /dev/null
+++ b/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
+}
diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts
index ce945386a92..933fceae3ec 100644
--- a/apps/sim/lib/core/config/redis.test.ts
+++ b/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),
         })
       )
diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts
index 63f1c6c25d4..7ca7889952d 100644
--- a/apps/sim/lib/core/config/redis.ts
+++ b/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),
 })
 
diff --git a/apps/sim/lib/execution/execution-signal.test.ts b/apps/sim/lib/execution/execution-signal.test.ts
index f16dc0e584b..e29f92c1a04 100644
--- a/apps/sim/lib/execution/execution-signal.test.ts
+++ b/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,14 +40,16 @@ 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. */
@@ -56,11 +57,13 @@ 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 () => {
diff --git a/apps/sim/lib/execution/execution-signal.ts b/apps/sim/lib/execution/execution-signal.ts
index 033387f56ba..65554f0f16c 100644
--- a/apps/sim/lib/execution/execution-signal.ts
+++ b/apps/sim/lib/execution/execution-signal.ts
@@ -1,9 +1,9 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import Redis, { type RedisOptions } from 'ioredis'
 import { getConfiguredRedisUrl, getRedisConnectionDefaults } from '@/lib/core/config/redis'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 const logger = createLogger('ExecutionSignalHub')
 const EXECUTION_SIGNAL_PREFIX = 'execution:signal:'
@@ -25,16 +25,19 @@ interface ChannelSubscription {
   acknowledged: boolean
 }
 
-/** Settles on the subscriber's next `ready` or `end`; `detach` stops listening for either. */
+/**
+ * Settles on the subscriber's next `ready` or `end`. Shared by every waiter so
+ * the client carries a single listener pair; `waiters` counts them so the last
+ * one to leave can `detach` the listeners.
+ */
 interface ReadySignal {
   promise: Promise<void>
   detach: () => void
+  waiters: number
 }
 
 export interface ExecutionSignalHub {
   subscribe(executionId: string, handler: ExecutionSignalHandler): Promise<() => void>
-  /** Resolves `true` once the hub can deliver signals, `false` if that could not be established in time. Never rejects. */
-  warm(): Promise<boolean>
 }
 
 export function getExecutionSignalChannel(executionId: string): string {
@@ -44,20 +47,19 @@ export function getExecutionSignalChannel(executionId: string): string {
 class RedisExecutionSignalHub implements ExecutionSignalHub {
   private readonly subscriber: Redis
   /**
-   * How long any one caller waits for readiness: room for one dead attempt and
-   * then a healthy one, so ioredis's own reconnect can be what rescues a stalled
-   * connection instead of the wait expiring while the first attempt is still
-   * being diagnosed. Derived from the exact options the client is built with.
-   *
-   * Every `subscribe` pays this too, not only warm-up — including against a
-   * Redis that is simply unreachable, where the connect deadline is what runs
-   * out and nothing is being diagnosed. That is the cost of the recovery.
+   * How long any one subscribe waits for readiness: room for one dead attempt
+   * and then a healthy one, so ioredis's own reconnect can be what rescues a
+   * stalled connection instead of the wait expiring while the first attempt is
+   * still being diagnosed. Derived from the exact options the client is built
+   * with. It is paid against a Redis that is simply unreachable as well, where
+   * the connect deadline is what runs out and nothing is being diagnosed; that
+   * is the cost of the recovery, and it widens the window in which a short
+   * execution timeout can pre-empt the wait and report itself instead.
    */
   private readonly readyTimeoutMs: number
   private readonly handlers = new Map<string, Set<ExecutionSignalHandler>>()
   private readonly subscriptions = new Map<string, ChannelSubscription>()
   private readySignal: ReadySignal | undefined
-  private readyWaiters = 0
   private connectedOnce = false
 
   constructor(redisUrl: string) {
@@ -71,6 +73,7 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     this.readyTimeoutMs = coldConnectionBudgetMs({
       connectTimeoutMs: options.connectTimeout,
       commandTimeoutMs: options.commandTimeout,
+      disconnectTimeoutMs: options.disconnectTimeout,
       reconnectDelayMs: subscriberRetryDelayMs(1),
     })
     this.subscriber = new Redis(redisUrl, options)
@@ -142,24 +145,6 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     }
   }
 
-  async warm(): Promise<boolean> {
-    try {
-      while (this.subscriber.status !== 'ready') {
-        await this.waitForConnectionReady()
-      }
-      return true
-    } catch (error) {
-      logger.warn(
-        'Execution signal subscriber warm-up gave up; first subscribe will pay the handshake',
-        {
-          error: toError(error).message,
-          status: this.subscriber.status,
-        }
-      )
-      return false
-    }
-  }
-
   private createSubscription(channels: string[]): ChannelSubscription {
     const subscription: ChannelSubscription = {
       acknowledged: false,
@@ -183,59 +168,47 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
   }
 
   /**
-   * One readiness signal is shared by every waiter so the client carries a
-   * single `ready`/`end` listener pair, but each waiter runs its own deadline
-   * over it. A shared timer would hand a subscribe that joins an in-flight
-   * warm-up only the remainder of that warm-up's budget — down to nothing —
-   * where it used to be guaranteed a full wait of its own. The signal is torn
-   * down when its last waiter gives up, so a timeout leaves nothing attached.
+   * Each waiter runs its own deadline over the shared readiness signal. One
+   * shared timer — which is what the memoized promise had — hands a waiter that
+   * joins late only the remainder of the first waiter's budget, down to
+   * nothing. The last waiter to leave detaches the signal, so a timeout leaves
+   * nothing attached; a signal that settles clears itself, so a later waiter
+   * observes the connection afresh rather than a readiness that has passed.
    */
   private waitForConnectionReady(): Promise<void> {
     if (this.subscriber.status === 'end') {
       return Promise.reject(new Error('Redis subscriber connection ended'))
     }
     const signal = (this.readySignal ??= this.createReadySignal())
-    this.readyWaiters++
-    return new Promise<void>((resolve, reject) => {
-      // A waiter that timed out is still subscribed to the signal, so it must
-      // leave exactly once whichever of its deadline or the signal fires first.
-      let left = false
-      const leave = () => {
-        if (left) return
-        left = true
-        clearTimeout(timeout)
-        if (--this.readyWaiters === 0 && this.readySignal === signal) {
-          signal.detach()
-          this.readySignal = undefined
-        }
-      }
-      const timeout = setTimeout(() => {
-        leave()
-        reject(new Error('Timed out waiting for Redis subscriber readiness'))
-      }, this.readyTimeoutMs)
-      signal.promise.then(
-        () => {
-          leave()
-          resolve()
-        },
-        (error: unknown) => {
-          leave()
-          reject(error)
-        }
+    signal.waiters++
+    let timer: NodeJS.Timeout | undefined
+    const deadline = new Promise<never>((_resolve, reject) => {
+      timer = setTimeout(
+        () => reject(new Error('Timed out waiting for Redis subscriber readiness')),
+        this.readyTimeoutMs
       )
     })
+    return Promise.race([signal.promise, deadline]).finally(() => {
+      clearTimeout(timer)
+      if (--signal.waiters === 0) {
+        signal.detach()
+        if (this.readySignal === signal) this.readySignal = undefined
+      }
+    })
   }
 
+  /**
+   * The promise is deliberately marked handled: a waiter that gives up stops
+   * observing it, and an `end` that arrives after the last one has left must
+   * not surface as an unhandled rejection.
+   */
   private createReadySignal(): ReadySignal {
-    let detach = () => {}
-    const signal: ReadySignal = { promise: Promise.resolve(), detach: () => detach() }
-    // A settled signal describes a moment that has passed; the next waiter must
-    // observe the connection afresh rather than a readiness that may be gone.
-    const settle = () => {
-      detach()
-      if (this.readySignal === signal) this.readySignal = undefined
-    }
+    const signal: ReadySignal = { promise: Promise.resolve(), detach: () => undefined, waiters: 0 }
     signal.promise = new Promise<void>((resolve, reject) => {
+      const settle = () => {
+        signal.detach()
+        if (this.readySignal === signal) this.readySignal = undefined
+      }
       const onReady = () => {
         settle()
         resolve()
@@ -244,15 +217,13 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
         settle()
         reject(new Error('Redis subscriber connection ended'))
       }
-      detach = () => {
+      signal.detach = () => {
         this.subscriber.removeListener('ready', onReady)
         this.subscriber.removeListener('end', onEnd)
       }
       this.subscriber.once('ready', onReady)
       this.subscriber.once('end', onEnd)
     })
-    // Detached before settling, this promise is simply dropped; an `end` that
-    // arrives after every waiter has left must not surface as unhandled.
     signal.promise.catch(() => undefined)
     return signal
   }
@@ -305,10 +276,6 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
 class LocalExecutionSignalHub implements ExecutionSignalHub {
   private readonly handlers = new Map<string, Set<ExecutionSignalHandler>>()
 
-  warm(): Promise<boolean> {
-    return Promise.resolve(true)
-  }
-
   async subscribe(executionId: string, handler: ExecutionSignalHandler): Promise<() => void> {
     const channel = getExecutionSignalChannel(executionId)
     let channelHandlers = this.handlers.get(channel)
@@ -359,22 +326,21 @@ export function getExecutionSignalHub(): ExecutionSignalHub {
 }
 
 /**
- * Establishes the hub's subscriber connection ahead of a cancellation
- * subscription, so that subscribe does not pay the handshake inside its own
- * readiness budget. Called fire-and-forget at the execution entry point — the
- * one path every execution shares, early enough to overlap the work ahead of
- * the subscribe — and at boot in long-lived servers. Never throws or rejects:
- * a throw from a start-up hook would fail the run, and a cold hub is only
- * slower, not wrong.
+ * Begins the hub's subscriber connection ahead of a cancellation subscription,
+ * so that subscribe does not pay the handshake inside its own readiness
+ * budget. Constructing the hub is what connects — ioredis dials in its
+ * constructor — so there is nothing to await. Called at the execution entry
+ * point, the one path every execution shares, early enough to overlap the work
+ * ahead of the subscribe. Never throws: a misconfigured URL belongs to the
+ * first real subscriber, which reports it against the execution that needed
+ * signals, and a cold hub is only slower, not wrong.
  */
-export async function warmExecutionSignalHub(): Promise<boolean> {
-  let hub: ExecutionSignalHub
+export function connectExecutionSignalHub(): void {
   try {
-    hub = getExecutionSignalHub()
+    getExecutionSignalHub()
   } catch {
-    return false
+    return
   }
-  return hub.warm()
 }
 
 export function publishLocalExecutionSignal(
diff --git a/apps/sim/lib/workflows/executor/execution-core.test.ts b/apps/sim/lib/workflows/executor/execution-core.test.ts
index 917b452dde8..78e03f78577 100644
--- a/apps/sim/lib/workflows/executor/execution-core.test.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.test.ts
@@ -97,12 +97,12 @@ vi.mock('@/lib/execution/cancellation', () => ({
   clearExecutionCancellation: clearExecutionCancellationMock,
 }))
 
-const { warmExecutionSignalHubMock } = vi.hoisted(() => ({
-  warmExecutionSignalHubMock: vi.fn(),
+const { connectExecutionSignalHubMock } = vi.hoisted(() => ({
+  connectExecutionSignalHubMock: vi.fn(),
 }))
 
 vi.mock('@/lib/execution/execution-signal', () => ({
-  warmExecutionSignalHub: warmExecutionSignalHubMock,
+  connectExecutionSignalHub: connectExecutionSignalHubMock,
 }))
 
 vi.mock('@/lib/core/security/encryption', () => ({
@@ -384,9 +384,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     expect(executorConstructorMock).toHaveBeenCalledTimes(1)
   })
 
-  it('begins warming the signal subscriber synchronously, before the first await', async () => {
-    warmExecutionSignalHubMock.mockResolvedValue(true)
-
+  it('begins connecting the signal subscriber synchronously, before the first await', async () => {
     const executionPromise = executeWorkflowCore({
       snapshot: createSnapshot() as unknown as ExecutionSnapshot,
       callbacks: {},
@@ -396,7 +394,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     // Asserted with no await in between: the handshake has to start ahead of
     // the custom-block read, or it stops overlapping the work that precedes the
     // cancellation subscribe and is paid inside that subscribe's budget instead.
-    expect(warmExecutionSignalHubMock).toHaveBeenCalledOnce()
+    expect(connectExecutionSignalHubMock).toHaveBeenCalledOnce()
 
     await executionPromise
   })
diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts
index 0ce8af40b62..a465c6b48e4 100644
--- a/apps/sim/lib/workflows/executor/execution-core.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.ts
@@ -22,7 +22,7 @@ import {
 import { withDatabaseReadRetry } from '@/lib/db/read-retry'
 import { getExecutionEnvironment } from '@/lib/environment/utils'
 import { clearExecutionCancellation } from '@/lib/execution/cancellation'
-import { warmExecutionSignalHub } from '@/lib/execution/execution-signal'
+import { connectExecutionSignalHub } from '@/lib/execution/execution-signal'
 import { warmLargeValueRefs } from '@/lib/execution/payloads/hydration'
 import { parseLargeExecutionValue } from '@/lib/execution/payloads/large-execution-value'
 import type { LoggingSession } from '@/lib/logs/execution/logging-session'
@@ -379,18 +379,18 @@ async function finalizeExecutionError(params: {
  * execution, and any nested child-workflow serialization (ALS propagates to the
  * whole async subtree).
  *
- * Also begins warming the execution-signal subscriber first, without awaiting
- * it: every execution subscribes to cancellation signals once its engine
- * starts, so starting that handshake here — the one path all of them share —
- * lets it overlap the reads and preprocessing ahead of the subscribe instead
- * of being paid inside its readiness budget. Warming on intent rather than at
- * worker start keeps the tasks that never execute a workflow, most of the
- * fleet by volume, from opening a connection they would never use.
+ * Also begins the execution-signal subscriber's connection first: every
+ * execution subscribes to cancellation signals once its engine starts, so
+ * starting that handshake here — the one path all of them share — lets it
+ * overlap the reads and preprocessing ahead of the subscribe instead of being
+ * paid inside its readiness budget. Connecting on intent rather than at worker
+ * start keeps the tasks that never execute a workflow, most of the fleet by
+ * volume, from opening a connection they would never use.
  */
 export async function executeWorkflowCore(
   options: ExecuteWorkflowCoreOptions
 ): Promise<ExecutionResult> {
-  void warmExecutionSignalHub()
+  connectExecutionSignalHub()
   const workspaceId = options.snapshot.metadata.workspaceId
   const rows = workspaceId
     ? await withDatabaseReadRetry(() => getCustomBlockRowsForWorkspace(workspaceId), {
diff --git a/packages/testing/src/mocks/redis-config.mock.test.ts b/packages/testing/src/mocks/redis-config.mock.test.ts
index 0db6bbdd54a..a226b955be2 100644
--- a/packages/testing/src/mocks/redis-config.mock.test.ts
+++ b/packages/testing/src/mocks/redis-config.mock.test.ts
@@ -19,6 +19,7 @@ describe('redis-config mock', () => {
     expect(redisConfigMock.getRedisConnectionDefaults('redis://localhost:6379')).toEqual({
       keepAlive: 1000,
       connectTimeout: 10000,
+      disconnectTimeout: 2000,
       enableOfflineQueue: true,
     })
   })
diff --git a/packages/testing/src/mocks/redis-config.mock.ts b/packages/testing/src/mocks/redis-config.mock.ts
index 4a8154e83a3..d45128747f0 100644
--- a/packages/testing/src/mocks/redis-config.mock.ts
+++ b/packages/testing/src/mocks/redis-config.mock.ts
@@ -31,6 +31,7 @@ function resolveTlsOptionsImpl(url: string | undefined): { servername: string }
 function getRedisConnectionDefaultsImpl(url?: string): {
   keepAlive: number
   connectTimeout: number
+  disconnectTimeout: number
   enableOfflineQueue: boolean
   tls?: { servername: string }
 } {
@@ -38,6 +39,7 @@ function getRedisConnectionDefaultsImpl(url?: string): {
   return {
     keepAlive: 1000,
     connectTimeout: 10000,
+    disconnectTimeout: 2000,
     enableOfflineQueue: true,
     ...(tls ? { tls } : {}),
   }
@@ -127,6 +129,9 @@ export function resetRedisConfigMock(): void {
  * ```
  */
 export const redisConfigMock = {
+  CONNECT_TIMEOUT_MS: 10_000,
+  DISCONNECT_TIMEOUT_MS: 2_000,
+  SHARED_COMMAND_TIMEOUT_MS: 5_000,
   getConfiguredRedisUrl: redisConfigMockFns.mockGetConfiguredRedisUrl,
   getRedisClient: redisConfigMockFns.mockGetRedisClient,
   getRedisConnectionDefaults: redisConfigMockFns.mockGetRedisConnectionDefaults,
diff --git a/packages/utils/src/retry.ts b/packages/utils/src/retry.ts
index dfe5cced729..c3fe72461d7 100644
--- a/packages/utils/src/retry.ts
+++ b/packages/utils/src/retry.ts
@@ -60,42 +60,3 @@ export function parseRetryAfter(header: string | null, maxMs = RETRY_AFTER_MAX_M
   }
   return null
 }
-
-/** 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
-  /** What the client's `retryStrategy` returns for its first reconnect. */
-  reconnectDelayMs: number
-}
-
-/**
- * How long a wait for a cold 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`. A
- * connection that opens but whose handshake is never answered costs command
- * deadlines: one when the URL carries a password, because ioredis treats a
- * timed-out `AUTH` as fatal and tears the socket down; two when it does not,
- * because `CLIENT SETNAME`/`SETINFO` must settle — by timing out — before the
- * `INFO` ready check even starts its own. The budget takes the largest of those
- * so it holds for either URL shape without parsing it. Only then does the
- * client's `retryStrategy` delay 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.
- */
-export function coldConnectionBudgetMs(options: ColdConnectionBudgetOptions): number {
-  const deadAttemptMs = Math.max(options.connectTimeoutMs, 2 * options.commandTimeoutMs)
-  return deadAttemptMs + options.reconnectDelayMs + HEALTHY_HANDSHAKE_ALLOWANCE_MS
-}