From fd8a45f64654f320f1a3e15c70f6b46eadb3bf25 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 9 Sep 2026 13:48:54 -0700 Subject: [PATCH 1/2] improvement(redis): warm the shared connection at process start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/instrumentation-node.ts | 6 ++ apps/sim/lib/core/config/redis.test.ts | 59 ++++++++++++++++++++ apps/sim/lib/core/config/redis.ts | 70 ++++++++++++++++++++++++ apps/sim/trigger.config.ts | 10 +++- packages/testing/src/mocks/redis.mock.ts | 22 +++++++- 5 files changed, 165 insertions(+), 2 deletions(-) diff --git a/apps/sim/instrumentation-node.ts b/apps/sim/instrumentation-node.ts index eb59f1bfe1c..b35aa21c931 100644 --- a/apps/sim/instrumentation-node.ts +++ b/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() } diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts index 6a03ef905cd..ad987e88bf9 100644 --- a/apps/sim/lib/core/config/redis.test.ts +++ b/apps/sim/lib/core/config/redis.test.ts @@ -49,6 +49,7 @@ import { getRedisClient, onRedisReconnect, resetForTesting, + warmRedisConnection, } from '@/lib/core/config/redis' describe('redis config', () => { @@ -472,6 +473,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' diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts index b604664b24d..2c11595df32 100644 --- a/apps/sim/lib/core/config/redis.ts +++ b/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 } | 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 { + 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((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). diff --git a/apps/sim/trigger.config.ts b/apps/sim/trigger.config.ts index 19fd5bd8290..6a218121ee5 100644 --- a/apps/sim/trigger.config.ts +++ b/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: { diff --git a/packages/testing/src/mocks/redis.mock.ts b/packages/testing/src/mocks/redis.mock.ts index 6771714cb2e..9b1aee39483 100644 --- a/packages/testing/src/mocks/redis.mock.ts +++ b/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 void>>() + return { // Hash operations hset: vi.fn().mockResolvedValue(1), @@ -49,7 +54,22 @@ 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) + }), + /** 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(() => ({ From 904ec3d9db7e0f4c7f8d7f4df82070d4dd4e62fa Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 9 Sep 2026 14:10:19 -0700 Subject: [PATCH 2/2] fix(testing): scope mock Redis listeners to the client that registered them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/lib/core/config/redis.test.ts | 25 +++--- packages/testing/src/mocks/redis.mock.test.ts | 90 +++++++++++++++++++ packages/testing/src/mocks/redis.mock.ts | 11 +++ 3 files changed, 112 insertions(+), 14 deletions(-) create mode 100644 packages/testing/src/mocks/redis.mock.test.ts diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts index ad987e88bf9..5314c2095fb 100644 --- a/apps/sim/lib/core/config/redis.test.ts +++ b/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 })) @@ -60,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(() => { diff --git a/packages/testing/src/mocks/redis.mock.test.ts b/packages/testing/src/mocks/redis.mock.test.ts new file mode 100644 index 00000000000..be2bd207ca0 --- /dev/null +++ b/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() + }) +}) diff --git a/packages/testing/src/mocks/redis.mock.ts b/packages/testing/src/mocks/redis.mock.ts index 9b1aee39483..40d6a91c928 100644 --- a/packages/testing/src/mocks/redis.mock.ts +++ b/packages/testing/src/mocks/redis.mock.ts @@ -62,6 +62,12 @@ export function createMockRedis() { 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) @@ -93,8 +99,13 @@ export type MockRedis = ReturnType /** * 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()