fix(redis): stop a slow handshake from failing as a command timeout

apps/sim/background/async-preprocessing-correlation.test.ts

@@ -771,4 +771,49 @@ describe('async preprocessing correlation threading', () => {
       })
     )
   })
+
+  it('keeps the enqueuer admission when a reservation refresh throws', async () => {
+    // A throw is ambiguous: the refresh mutates the local reservation and the
+    // pointer separately, so it can fail after the slot's TTL was extended.
+    mockRefreshExecutionSlotExpiry.mockRejectedValueOnce(new Error('Command timed out'))
+    mockPreprocessExecution.mockResolvedValueOnce({
+      success: true,
+      actorUserId: 'actor-1',
+      workflowRecord: {
+        id: 'workflow-1',
+        userId: 'owner-1',
+        workspaceId: 'workspace-1',
+        variables: {},
+      },
+      billingAttribution,
+      executionTimeout: {},
+    })
+    mockExecuteWorkflowCore.mockResolvedValueOnce({
+      success: true,
+      status: 'success',
+      output: { ok: true },
+      metadata: { duration: 10, userId: 'actor-1' },
+    })
+
+    await expect(
+      executeWorkflowJob({
+        principal,
+        workflowId: 'workflow-1',
+        userId: 'actor-1',
+        workspaceId: 'workspace-1',
+        billingAttribution,
+        triggerType: 'api',
+        executionId: 'execution-refresh-throw',
+        requestId: 'request-refresh-throw',
+        admissionCompleted: true,
+        executionTimeoutMs: 60_000,
+      })
+    ).resolves.toEqual(expect.objectContaining({ success: true }))
+
+    // Re-admitting would spend another rate-limit token and could reject a run
+    // that still holds a valid slot.
+    expect(mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({ checkRateLimit: false, skipUsageLimits: true })
+    )
+  })
 })

apps/sim/background/workflow-execution.ts

@@ -165,14 +165,38 @@ export async function executeWorkflowJob(
     const executionDeadlineAt = getExecutionDeadlineAt(timeoutController.signal)?.getTime()
     let admissionCompleted = payload.admissionCompleted === true
     if (admissionCompleted && executionDeadlineAt !== undefined) {
-      admissionCompleted = await refreshExecutionSlotExpiry(
-        executionId,
-        executionDeadlineAt + RESERVATION_TTL_BUFFER_MS
-      )
-      if (!admissionCompleted) {
-        logger.warn('Queued workflow reservation expired; repeating usage admission', {
+      try {
+        admissionCompleted = await refreshExecutionSlotExpiry(
+          executionId,
+          executionDeadlineAt + RESERVATION_TTL_BUFFER_MS
+        )
+        if (!admissionCompleted) {
+          logger.warn('Queued workflow reservation expired; repeating usage admission', {
+            workflowId,
+            executionId,
+          })
+        }
+      } catch (error) {
+        /**
+         * Only a `false` return proves the reservation is gone, and that is the
+         * one outcome re-admission is right for. A throw does not: the refresh
+         * is several separate Redis mutations, so it can fail after the slot's
+         * TTL was already extended, and a client-side command timeout abandons
+         * a call the server may still have applied. Re-admitting on that
+         * ambiguity would spend another rate-limit token and could reject a run
+         * that still holds a perfectly valid slot, so keep the admission the
+         * enqueuing surface already completed.
+         *
+         * Worst case the slot lapses before the run ends, which under-counts
+         * concurrency for this one execution; the later release is already a
+         * no-op when the reservation is gone. Previously this threw before the
+         * logging session existed, so the run left no log row at all and simply
+         * vanished from the workspace's logs.
+         */
+        logger.warn('Reservation refresh failed; continuing on the existing admission', {
           workflowId,
           executionId,
+          error: toError(error).message,
         })
       }
     }

apps/sim/instrumentation-node.ts

@@ -397,4 +397,14 @@ export async function register() {
 
   const { startMemoryTelemetry } = await import('./lib/monitoring/memory-telemetry')
   startMemoryTelemetry()
+
+  /**
+   * Open the shared Redis connection during boot so the first request does not
+   * pay the TLS handshake inside its own command deadline. Deliberately not
+   * awaited: a slow or unreachable Redis must never hold up serving, and
+   * `warmRedisConnection` already bounds and swallows its own failures.
+   */
+  void import('./lib/core/config/redis')
+    .then(({ warmRedisConnection }) => warmRedisConnection())
+    .catch((error) => logger.warn('Redis warm-up could not start', { error }))
 }

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

@@ -29,8 +29,11 @@ import {
   closeRedisConnection,
   extendLock,
   getRedisClient,
+  getRedisConnectionDefaults,
   onRedisReconnect,
+  REDIS_COMMAND_TIMEOUT_MS,
   resetForTesting,
+  warmRedisConnection,
 } from '@/lib/core/config/redis'
 
 describe('redis config', () => {
@@ -303,6 +306,110 @@ describe('redis config', () => {
     })
   })
 
+  describe('command timeout', () => {
+    it('stays above the connect timeout so a slow handshake is not reported as a command timeout', () => {
+      const { connectTimeout } = getRedisConnectionDefaults('redis://localhost:6379')
+
+      expect(connectTimeout).toBeDefined()
+      expect(REDIS_COMMAND_TIMEOUT_MS).toBeGreaterThan(connectTimeout as number)
+    })
+
+    it('applies that timeout to the shared client', () => {
+      getRedisClient()
+
+      expect(MockRedisConstructor).toHaveBeenCalledWith(
+        expect.any(String),
+        expect.objectContaining({ commandTimeout: REDIS_COMMAND_TIMEOUT_MS })
+      )
+    })
+
+    it('does not let the command deadline govern how fast a dead connection is detected', async () => {
+      const listener = vi.fn()
+      onRedisReconnect(listener)
+      getRedisClient()
+
+      // A PING that never settles — the failure mode the health check exists for.
+      mockRedisInstance.ping.mockReturnValue(new Promise(() => {}))
+
+      // Two intervals plus two probe deadlines is well under two command
+      // deadlines, so this only passes while the probe has its own budget.
+      await vi.advanceTimersByTimeAsync(15_000)
+      await vi.advanceTimersByTimeAsync(15_000)
+      await vi.advanceTimersByTimeAsync(15_000)
+
+      expect(listener).toHaveBeenCalledTimes(1)
+      expect(3 * 15_000).toBeLessThan(2 * REDIS_COMMAND_TIMEOUT_MS + 2 * 15_000)
+    })
+  })
+
+  describe('warmRedisConnection', () => {
+    it('resolves without waiting when the client is already connected', async () => {
+      mockRedisInstance.status = 'ready'
+
+      await expect(warmRedisConnection()).resolves.toBeUndefined()
+      expect(mockRedisInstance.once).not.toHaveBeenCalled()
+    })
+
+    it('resolves once the connection reports ready', async () => {
+      mockRedisInstance.status = 'connecting'
+      const readyHandlers: Array<() => void> = []
+      mockRedisInstance.once.mockImplementation((event: string, cb: () => void) => {
+        if (event === 'ready') readyHandlers.push(cb)
+      })
+
+      let settled = false
+      const warm = warmRedisConnection().then(() => {
+        settled = true
+      })
+
+      await vi.advanceTimersByTimeAsync(2_000)
+      expect(settled).toBe(false)
+
+      for (const handler of readyHandlers) handler()
+      await warm
+
+      expect(settled).toBe(true)
+    })
+
+    it('gives up at the connect deadline rather than blocking startup forever', async () => {
+      mockRedisInstance.status = 'connecting'
+      mockRedisInstance.once.mockImplementation(() => {})
+
+      let settled = false
+      const warm = warmRedisConnection().then(() => {
+        settled = true
+      })
+
+      await vi.advanceTimersByTimeAsync(9_000)
+      expect(settled).toBe(false)
+
+      await vi.advanceTimersByTimeAsync(2_000)
+      await warm
+
+      expect(settled).toBe(true)
+    })
+
+    it('warms a given client once so a warm process pays nothing per unit of work', async () => {
+      mockRedisInstance.status = 'connecting'
+      mockRedisInstance.once.mockImplementation((event: string, cb: () => void) => {
+        if (event === 'ready') cb()
+      })
+
+      await warmRedisConnection()
+      const callsAfterFirst = mockRedisInstance.once.mock.calls.length
+
+      await warmRedisConnection()
+
+      expect(mockRedisInstance.once.mock.calls.length).toBe(callsAfterFirst)
+    })
+
+    it('resolves instead of throwing when Redis is not configured', async () => {
+      mockEnv.REDIS_URL = undefined
+
+      await expect(warmRedisConnection()).resolves.toBeUndefined()
+    })
+  })
+
   describe('retryStrategy', () => {
     function captureRetryStrategy(): (times: number) => number {
       let capturedConfig: Record<string, unknown> = {}