Skip to content

Commit fd8a45f

Browse files
committed
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.
1 parent 568a539 commit fd8a45f

5 files changed

Lines changed: 165 additions & 2 deletions

File tree

apps/sim/instrumentation-node.ts

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -403,4 +403,10 @@ export async function register() {
403403

404404
const { startMemoryTelemetry } = await import('./lib/monitoring/memory-telemetry')
405405
startMemoryTelemetry()
406+
407+
// Not awaited: the connection is warmed in the background so the first request
408+
// that needs Redis does not pay the handshake inside its own command deadline,
409+
// but boot never waits on Redis to serve requests that do not touch it.
410+
const { warmRedisConnection } = await import('./lib/core/config/redis')
411+
void warmRedisConnection()
406412
}

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

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ import {
4949
getRedisClient,
5050
onRedisReconnect,
5151
resetForTesting,
52+
warmRedisConnection,
5253
} from '@/lib/core/config/redis'
5354

5455
describe('redis config', () => {
@@ -472,6 +473,64 @@ describe('redis config', () => {
472473
})
473474
})
474475

476+
describe('warmRedisConnection', () => {
477+
it('resolves immediately when the connection is already usable', async () => {
478+
mockRedisInstance.status = 'ready'
479+
480+
await expect(warmRedisConnection()).resolves.toBe(true)
481+
})
482+
483+
it('resolves once the connection becomes ready', async () => {
484+
mockRedisInstance.status = 'connecting'
485+
const warm = warmRedisConnection()
486+
const client = getRedisClient()
487+
488+
Object.assign(client ?? {}, { status: 'ready' })
489+
client?.emit('ready')
490+
491+
await expect(warm).resolves.toBe(true)
492+
})
493+
494+
it('gives up at the deadline rather than waiting on a connection that never lands', async () => {
495+
mockRedisInstance.status = 'connecting'
496+
const warm = warmRedisConnection(10_000)
497+
498+
await vi.advanceTimersByTimeAsync(10_000)
499+
500+
// False, not a rejection: a cold connection is the caller's normal case.
501+
await expect(warm).resolves.toBe(false)
502+
})
503+
504+
it('shares one warm-up across concurrent callers', async () => {
505+
mockRedisInstance.status = 'connecting'
506+
507+
expect(warmRedisConnection()).toBe(warmRedisConnection())
508+
})
509+
510+
it('warms again after the health check replaces the client', async () => {
511+
mockRedisInstance.status = 'connecting'
512+
const first = warmRedisConnection()
513+
resetForTesting()
514+
515+
// Keyed on the client, so a replacement is warmed on its own terms rather
516+
// than inheriting a settled promise describing a connection that is gone.
517+
expect(warmRedisConnection()).not.toBe(first)
518+
})
519+
520+
it('reports not-warm instead of throwing when Redis is unconfigured', async () => {
521+
mockEnv.REDIS_URL = undefined
522+
523+
await expect(warmRedisConnection()).resolves.toBe(false)
524+
})
525+
526+
it('reports not-warm instead of throwing when the URL is invalid', async () => {
527+
// A start-up hook that throws here would take the whole run attempt with it.
528+
mockEnv.REDIS_URL = 'https://cache.example.com'
529+
530+
await expect(warmRedisConnection()).resolves.toBe(false)
531+
})
532+
})
533+
475534
describe('capability validation', () => {
476535
it('rejects a non-Redis URL before constructing a client', () => {
477536
mockEnv.REDIS_URL = 'https://cache.example.com'

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

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,14 @@ interface RedisState {
6767
reconnects: number
6868
errors: number
6969
lastErrorMessage: string | null
70+
/**
71+
* In-flight warm-up, tied to the client it is warming. Keying on the client
72+
* is what makes a replacement re-warm, so this is never cleared by hand —
73+
* every site that drops the client would otherwise have to remember to, and
74+
* one that forgot would hand back a promise describing a connection that is
75+
* already gone.
76+
*/
77+
warmup: { client: Redis; promise: Promise<boolean> } | null
7078
}
7179

7280
const g = globalThis as typeof globalThis & { _redisState?: RedisState }
@@ -84,6 +92,7 @@ if (!g._redisState) {
8492
reconnects: 0,
8593
errors: 0,
8694
lastErrorMessage: null,
95+
warmup: null,
8796
}
8897
}
8998
const state = g._redisState
@@ -194,6 +203,12 @@ export function describeRedisConnection(
194203

195204
const PING_INTERVAL_MS = 15_000
196205
const MAX_PING_FAILURES = 2
206+
/**
207+
* Warm-up budget. Sized to outlast a slow handshake rather than a fast one,
208+
* because giving up early just returns the handshake to the first command's
209+
* deadline, which is the thing this exists to avoid.
210+
*/
211+
const REDIS_WARMUP_TIMEOUT_MS = 10_000
197212

198213
export function getConfiguredRedisUrl(): string | null {
199214
if (getConfiguredCacheProvider() === 'database') return null
@@ -343,6 +358,61 @@ export function getRedisClient(): Redis | null {
343358
}
344359
}
345360

361+
/**
362+
* Establishing a connection is far more expensive than the commands that run
363+
* over it, so it should cost once per process rather than once per unit of
364+
* work. Its own budget, too: `commandTimeout` is armed before ioredis checks
365+
* whether the socket is writable, so a first command issued against a client
366+
* still shaking hands spends that budget waiting to connect and fails as a
367+
* command timeout from a server that never received it.
368+
*
369+
* Resolves `true` once the shared connection is usable, `false` when Redis is
370+
* not configured or the wait ran out. It never throws and never rejects:
371+
* callers run at process start — a Trigger.dev `init` hook fails the whole run
372+
* attempt if it throws — and a warm-up is an optimization, so failing to warm
373+
* must cost nothing beyond the connection staying cold.
374+
*/
375+
export function warmRedisConnection(timeoutMs = REDIS_WARMUP_TIMEOUT_MS): Promise<boolean> {
376+
let client: Redis | null = null
377+
try {
378+
client = getRedisClient()
379+
} catch {
380+
// A misconfigured URL belongs to the first real caller, which can report it
381+
// against the operation that needed Redis. Warming must not turn it into a
382+
// start-up failure.
383+
return Promise.resolve(false)
384+
}
385+
if (!client) return Promise.resolve(false)
386+
if (client.status === 'ready') return Promise.resolve(true)
387+
if (state.warmup?.client === client) return state.warmup.promise
388+
389+
const warming = client
390+
const promise = new Promise<boolean>((resolve) => {
391+
let settled = false
392+
const finish = (warm: boolean) => {
393+
if (settled) return
394+
settled = true
395+
clearTimeout(timer)
396+
warming.removeListener('ready', onReady)
397+
resolve(warm)
398+
}
399+
const onReady = () => finish(true)
400+
const timer = setTimeout(() => {
401+
logger.warn('Redis warm-up timed out; first command will pay the handshake', {
402+
timeoutMs,
403+
redis: describeRedisConnection(warming),
404+
})
405+
finish(false)
406+
}, timeoutMs)
407+
// A pending warm-up must never be the reason a process stays alive.
408+
timer.unref?.()
409+
warming.on('ready', onReady)
410+
})
411+
412+
state.warmup = { client: warming, promise }
413+
return promise
414+
}
415+
346416
/**
347417
* Lua script for safe lock release.
348418
* Only deletes the key if the value matches (ownership verification).

apps/sim/trigger.config.ts

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,10 +102,18 @@ export default defineConfig({
102102
* environment variables whether Trigger.dev is available: a process that
103103
* Trigger.dev is executing has Trigger.dev available by definition.
104104
*
105+
* Also warms the shared Redis connection, because a run's first Redis call is
106+
* typically a lock acquire and would otherwise pay the handshake inside its
107+
* own command deadline. Awaited so the connection is up before `run()` issues
108+
* anything; imported dynamically so deploy-time evaluation of this config does
109+
* not pull the client, and never throwing because a throw here fails the run.
110+
*
105111
* @see https://trigger.dev/docs/config/config-file#lifecycle-functions
106112
*/
107-
init: () => {
113+
init: async () => {
108114
markInsideTriggerRun()
115+
const { warmRedisConnection } = await import('./lib/core/config/redis')
116+
await warmRedisConnection()
109117
},
110118
...(grafanaTelemetry ? { telemetry: grafanaTelemetry } : {}),
111119
build: {

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

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,11 @@ import { vi } from 'vitest'
1414
* ```
1515
*/
1616
export function createMockRedis() {
17+
/** Per-instance listener registry, so `emit` can drive the lifecycle events
18+
* a real client emits. `on` stays a spy: tests read `on.mock.calls` to reach
19+
* the handlers the client registered. */
20+
const listeners = new Map<string, Set<(...args: unknown[]) => void>>()
21+
1722
return {
1823
// Hash operations
1924
hset: vi.fn().mockResolvedValue(1),
@@ -49,7 +54,22 @@ export function createMockRedis() {
4954
publish: vi.fn().mockResolvedValue(0),
5055
subscribe: vi.fn().mockResolvedValue(undefined),
5156
unsubscribe: vi.fn().mockResolvedValue(undefined),
52-
on: vi.fn(),
57+
on: vi.fn((event: string, listener: (...args: unknown[]) => void) => {
58+
const existing = listeners.get(event)
59+
if (existing) existing.add(listener)
60+
else listeners.set(event, new Set([listener]))
61+
}),
62+
removeListener: vi.fn((event: string, listener: (...args: unknown[]) => void) => {
63+
listeners.get(event)?.delete(listener)
64+
}),
65+
/** Drives the lifecycle events a real client emits (`connect`, `ready`, `error`). */
66+
emit: vi.fn((event: string, ...args: unknown[]) => {
67+
const registered = listeners.get(event)
68+
if (!registered?.size) return false
69+
// Copy first: a listener may remove itself while the event is dispatching.
70+
for (const listener of [...registered]) listener(...args)
71+
return true
72+
}),
5373

5474
// Transaction
5575
multi: vi.fn(() => ({

0 commit comments

Comments
 (0)