improvement(redis): warm the shared connection at process start (#7683)

* improvement(redis): warm the shared connection at process start

Establishing a connection is far more expensive than the commands that run over
it, so it should cost once per process rather than once per unit of work. It
also needs its own budget: `commandTimeout` is armed before ioredis checks
whether the socket is writable, so a first command issued against a client still
shaking hands spends that budget waiting to connect and fails as a command
timeout from a server that never received it. A run's first Redis call is
typically a lock acquire, which is exactly where that surfaces.

`warmRedisConnection` resolves once the connection is usable, or `false` when
Redis is unconfigured or the wait ran out. It never throws and never rejects —
a Trigger.dev `init` hook that throws fails the whole run attempt, and a
warm-up is an optimization, so failing to warm must cost nothing beyond the
connection staying cold. The deadline is its own, and its timer is unref'd so a
pending warm-up can never hold a process open.

The in-flight warm-up is keyed on the client it is warming, which is what makes
a replacement re-warm. That keying is the only mechanism: clearing by hand at
every site that drops the client is an invariant that rots the first time one
forgets.

Trigger.dev awaits it in the global `init` hook so the connection is up before
`run()` issues anything; Next starts it without awaiting so boot never waits on
Redis to serve requests that do not touch it.

Gives the shared Redis mock a real listener registry so lifecycle events can be
driven in tests. `on` stays a spy — tests read `on.mock.calls` to reach the
handlers the client registered.

* fix(testing): scope mock Redis listeners to the client that registered them

The listener registry outlived the spies: `vi.clearAllMocks()` and
`clearRedisMocks` reset call history but left handlers registered, so they
accumulated across tests and a later `emit` could reach handlers belonging to a
client the test under way never created.

Adds `removeAllListeners`, which real clients have, and drops listeners in
`clearRedisMocks` alongside spy history. Where one mock instance stands in for
every client a module constructs, the registry is now emptied per construction —
a real client starts with none, so binding listener lifetime to construction
makes the isolation automatic rather than something each test has to remember.

Covers the mock's event behavior in the testing package, where it lives.

Author: waleedlatif1

apps/sim/instrumentation-node.ts

@@ -403,4 +403,10 @@ 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')
+  void warmRedisConnection()
 }

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

@@ -18,13 +18,16 @@ const { mockEnv, MockRedisConstructor, mockLogger } = vi.hoisted(() => ({
 }))
 
 const mockRedisInstance = createMockRedis()
-MockRedisConstructor.mockImplementation(
-  class {
-    constructor() {
-      Object.assign(this, mockRedisInstance)
-    }
-  }
-)
+/** One mock instance stands in for every client the module constructs, so its
+ *  listener registry has to be emptied per construction — a real client starts
+ *  with none, and keeping them would let an `emit` reach handlers registered by
+ *  a client that no longer exists. */
+function newMockClient(this: object) {
+  mockRedisInstance.removeAllListeners()
+  Object.assign(this, mockRedisInstance)
+}
+
+MockRedisConstructor.mockImplementation(newMockClient)
 
 vi.unmock('@/lib/core/config/redis')
 vi.mock('@/lib/core/config/env', () => ({ env: mockEnv }))
@@ -49,6 +52,7 @@ import {
   getRedisClient,
   onRedisReconnect,
   resetForTesting,
+  warmRedisConnection,
 } from '@/lib/core/config/redis'
 
 describe('redis config', () => {
@@ -59,13 +63,7 @@ describe('redis config', () => {
     mockRedisInstance.status = 'ready'
     mockEnv.REDIS_URL = 'redis://localhost:6379'
     mockEnv.REDIS_TLS_SERVERNAME = undefined
-    MockRedisConstructor.mockImplementation(
-      class {
-        constructor() {
-          Object.assign(this, mockRedisInstance)
-        }
-      }
-    )
+    MockRedisConstructor.mockImplementation(newMockClient)
   })
 
   afterEach(() => {
@@ -472,6 +470,64 @@ describe('redis config', () => {
     })
   })
 
+  describe('warmRedisConnection', () => {
+    it('resolves immediately when the connection is already usable', async () => {
+      mockRedisInstance.status = 'ready'
+
+      await expect(warmRedisConnection()).resolves.toBe(true)
+    })
+
+    it('resolves once the connection becomes ready', async () => {
+      mockRedisInstance.status = 'connecting'
+      const warm = warmRedisConnection()
+      const client = getRedisClient()
+
+      Object.assign(client ?? {}, { status: 'ready' })
+      client?.emit('ready')
+
+      await expect(warm).resolves.toBe(true)
+    })
+
+    it('gives up at the deadline rather than waiting on a connection that never lands', async () => {
+      mockRedisInstance.status = 'connecting'
+      const warm = warmRedisConnection(10_000)
+
+      await vi.advanceTimersByTimeAsync(10_000)
+
+      // False, not a rejection: a cold connection is the caller's normal case.
+      await expect(warm).resolves.toBe(false)
+    })
+
+    it('shares one warm-up across concurrent callers', async () => {
+      mockRedisInstance.status = 'connecting'
+
+      expect(warmRedisConnection()).toBe(warmRedisConnection())
+    })
+
+    it('warms again after the health check replaces the client', async () => {
+      mockRedisInstance.status = 'connecting'
+      const first = warmRedisConnection()
+      resetForTesting()
+
+      // Keyed on the client, so a replacement is warmed on its own terms rather
+      // than inheriting a settled promise describing a connection that is gone.
+      expect(warmRedisConnection()).not.toBe(first)
+    })
+
+    it('reports not-warm instead of throwing when Redis is unconfigured', async () => {
+      mockEnv.REDIS_URL = undefined
+
+      await expect(warmRedisConnection()).resolves.toBe(false)
+    })
+
+    it('reports not-warm instead of throwing when the URL is invalid', async () => {
+      // A start-up hook that throws here would take the whole run attempt with it.
+      mockEnv.REDIS_URL = 'https://cache.example.com'
+
+      await expect(warmRedisConnection()).resolves.toBe(false)
+    })
+  })
+
   describe('capability validation', () => {
     it('rejects a non-Redis URL before constructing a client', () => {
       mockEnv.REDIS_URL = 'https://cache.example.com'

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

@@ -67,6 +67,14 @@ interface RedisState {
   reconnects: number
   errors: number
   lastErrorMessage: string | null
+  /**
+   * In-flight warm-up, tied to the client it is warming. Keying on the client
+   * is what makes a replacement re-warm, so this is never cleared by hand —
+   * every site that drops the client would otherwise have to remember to, and
+   * one that forgot would hand back a promise describing a connection that is
+   * already gone.
+   */
+  warmup: { client: Redis; promise: Promise<boolean> } | null
 }
 
 const g = globalThis as typeof globalThis & { _redisState?: RedisState }
@@ -84,6 +92,7 @@ if (!g._redisState) {
     reconnects: 0,
     errors: 0,
     lastErrorMessage: null,
+    warmup: null,
   }
 }
 const state = g._redisState
@@ -194,6 +203,12 @@ export function describeRedisConnection(
 
 const PING_INTERVAL_MS = 15_000
 const MAX_PING_FAILURES = 2
+/**
+ * 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.
+ */
+const REDIS_WARMUP_TIMEOUT_MS = 10_000
 
 export function getConfiguredRedisUrl(): string | null {
   if (getConfiguredCacheProvider() === 'database') return null
@@ -343,6 +358,61 @@ export function getRedisClient(): Redis | null {
   }
 }
 
+/**
+ * Establishing a connection is far more expensive than the commands that run
+ * over it, so it should cost once per process rather than once per unit of
+ * work. Its own budget, too: `commandTimeout` is armed before ioredis checks
+ * whether the socket is writable, so a first command issued against a client
+ * still shaking hands spends that budget waiting to connect and fails as a
+ * command timeout from a server that never received it.
+ *
+ * Resolves `true` once the shared connection is usable, `false` when Redis is
+ * not configured or the wait ran out. It never throws and never rejects:
+ * callers run at process start — a Trigger.dev `init` hook fails the whole run
+ * attempt if it throws — and a warm-up is an optimization, so failing to warm
+ * must cost nothing beyond the connection staying cold.
+ */
+export function warmRedisConnection(timeoutMs = REDIS_WARMUP_TIMEOUT_MS): Promise<boolean> {
+  let client: Redis | null = null
+  try {
+    client = getRedisClient()
+  } catch {
+    // A misconfigured URL belongs to the first real caller, which can report it
+    // against the operation that needed Redis. Warming must not turn it into a
+    // start-up failure.
+    return Promise.resolve(false)
+  }
+  if (!client) return Promise.resolve(false)
+  if (client.status === 'ready') return Promise.resolve(true)
+  if (state.warmup?.client === client) return state.warmup.promise
+
+  const warming = client
+  const promise = new Promise<boolean>((resolve) => {
+    let settled = false
+    const finish = (warm: boolean) => {
+      if (settled) return
+      settled = true
+      clearTimeout(timer)
+      warming.removeListener('ready', onReady)
+      resolve(warm)
+    }
+    const onReady = () => finish(true)
+    const timer = setTimeout(() => {
+      logger.warn('Redis warm-up timed out; first command will pay the handshake', {
+        timeoutMs,
+        redis: describeRedisConnection(warming),
+      })
+      finish(false)
+    }, timeoutMs)
+    // A pending warm-up must never be the reason a process stays alive.
+    timer.unref?.()
+    warming.on('ready', onReady)
+  })
+
+  state.warmup = { client: warming, promise }
+  return promise
+}
+
 /**
  * Lua script for safe lock release.
  * Only deletes the key if the value matches (ownership verification).

apps/sim/trigger.config.ts

@@ -102,10 +102,18 @@ 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.
+   *
    * @see https://trigger.dev/docs/config/config-file#lifecycle-functions
    */
-  init: () => {
+  init: async () => {
     markInsideTriggerRun()
+    const { warmRedisConnection } = await import('./lib/core/config/redis')
+    await warmRedisConnection()
   },
   ...(grafanaTelemetry ? { telemetry: grafanaTelemetry } : {}),
   build: {

packages/testing/src/mocks/redis.mock.test.ts

@@ -0,0 +1,90 @@
+import { describe, expect, it, vi } from 'vitest'
+import { clearRedisMocks, createMockRedis } from './redis.mock'
+
+describe('createMockRedis events', () => {
+  it('dispatches an emitted event to its registered listeners', () => {
+    const redis = createMockRedis()
+    const onReady = vi.fn()
+    redis.on('ready', onReady)
+
+    expect(redis.emit('ready')).toBe(true)
+    expect(onReady).toHaveBeenCalledOnce()
+  })
+
+  it('reports no delivery when nothing is listening', () => {
+    expect(createMockRedis().emit('ready')).toBe(false)
+  })
+
+  it('stops delivering to a removed listener', () => {
+    const redis = createMockRedis()
+    const onReady = vi.fn()
+    redis.on('ready', onReady)
+    redis.removeListener('ready', onReady)
+
+    redis.emit('ready')
+    expect(onReady).not.toHaveBeenCalled()
+  })
+
+  it('lets a listener remove itself while the event is dispatching', () => {
+    const redis = createMockRedis()
+    const onReady = vi.fn(() => redis.removeListener('ready', onReady))
+    redis.on('ready', onReady)
+
+    expect(() => redis.emit('ready')).not.toThrow()
+    redis.emit('ready')
+    expect(onReady).toHaveBeenCalledOnce()
+  })
+
+  it('drops every listener on removeAllListeners', () => {
+    const redis = createMockRedis()
+    const onReady = vi.fn()
+    const onError = vi.fn()
+    redis.on('ready', onReady)
+    redis.on('error', onError)
+
+    redis.removeAllListeners()
+
+    redis.emit('ready')
+    redis.emit('error')
+    expect(onReady).not.toHaveBeenCalled()
+    expect(onError).not.toHaveBeenCalled()
+  })
+
+  it('drops only the named event when one is given', () => {
+    const redis = createMockRedis()
+    const onReady = vi.fn()
+    const onError = vi.fn()
+    redis.on('ready', onReady)
+    redis.on('error', onError)
+
+    redis.removeAllListeners('ready')
+
+    redis.emit('ready')
+    redis.emit('error')
+    expect(onReady).not.toHaveBeenCalled()
+    expect(onError).toHaveBeenCalledOnce()
+  })
+
+  it('clears listeners alongside spy history, not just spy history', () => {
+    // Handlers left behind would be invoked by a later emit on behalf of a
+    // client the test under way never created.
+    const redis = createMockRedis()
+    const onReady = vi.fn()
+    redis.on('ready', onReady)
+
+    clearRedisMocks(redis)
+
+    expect(redis.emit('ready')).toBe(false)
+    expect(onReady).not.toHaveBeenCalled()
+  })
+
+  it('keeps listeners scoped to the instance that registered them', () => {
+    const a = createMockRedis()
+    const b = createMockRedis()
+    const onA = vi.fn()
+    a.on('ready', onA)
+
+    b.emit('ready')
+    expect(onA).not.toHaveBeenCalled()
+  })
+})

packages/testing/src/mocks/redis.mock.ts

@@ -14,6 +14,11 @@ import { vi } from 'vitest'
  * ```
  */
 export function createMockRedis() {
+  /** Per-instance listener registry, so `emit` can drive the lifecycle events
+   *  a real client emits. `on` stays a spy: tests read `on.mock.calls` to reach
+   *  the handlers the client registered. */
+  const listeners = new Map<string, Set<(...args: unknown[]) => void>>()
+
   return {
     // Hash operations
     hset: vi.fn().mockResolvedValue(1),
@@ -49,7 +54,28 @@ export function createMockRedis() {
     publish: vi.fn().mockResolvedValue(0),
     subscribe: vi.fn().mockResolvedValue(undefined),
     unsubscribe: vi.fn().mockResolvedValue(undefined),
-    on: vi.fn(),
+    on: vi.fn((event: string, listener: (...args: unknown[]) => void) => {
+      const existing = listeners.get(event)
+      if (existing) existing.add(listener)
+      else listeners.set(event, new Set([listener]))
+    }),
+    removeListener: vi.fn((event: string, listener: (...args: unknown[]) => void) => {
+      listeners.get(event)?.delete(listener)
+    }),
+    /** Listeners belong to a client, so a caller reusing this instance as a new
+     *  client clears them the way a real one starts empty. */
+    removeAllListeners: vi.fn((event?: string) => {
+      if (event === undefined) listeners.clear()
+      else listeners.delete(event)
+    }),
+    /** Drives the lifecycle events a real client emits (`connect`, `ready`, `error`). */
+    emit: vi.fn((event: string, ...args: unknown[]) => {
+      const registered = listeners.get(event)
+      if (!registered?.size) return false
+      // Copy first: a listener may remove itself while the event is dispatching.
+      for (const listener of [...registered]) listener(...args)
+      return true
+    }),
 
     // Transaction
     multi: vi.fn(() => ({
@@ -73,8 +99,13 @@ export type MockRedis = ReturnType<typeof createMockRedis>
 
 /**
  * Clears all Redis mock calls.
+ *
+ * Also drops registered listeners: spy history and the listener registry are
+ * separate state, and handlers left behind would be invoked by a later `emit`
+ * on behalf of a client the test under way never created.
  */
 export function clearRedisMocks(redis: MockRedis) {
+  redis.removeAllListeners()
   Object.values(redis).forEach((value) => {
     if (typeof value === 'function' && 'mockClear' in value) {
       value.mockClear()