diff --git a/apps/sim/instrumentation-node.ts b/apps/sim/instrumentation-node.ts index b35aa21c931..d9a3668d304 100644 --- a/apps/sim/instrumentation-node.ts +++ b/apps/sim/instrumentation-node.ts @@ -407,6 +407,6 @@ export async function register() { // 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') + const { warmRedisConnection } = await import('@/lib/core/config/redis') void warmRedisConnection() } diff --git a/apps/sim/lib/core/config/redis-budget.test.ts b/apps/sim/lib/core/config/redis-budget.test.ts new file mode 100644 index 00000000000..55c786d3514 --- /dev/null +++ b/apps/sim/lib/core/config/redis-budget.test.ts @@ -0,0 +1,44 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget' + +describe('coldConnectionBudgetMs', () => { + 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: 5_000, + disconnectTimeoutMs: 2_000, + reconnectDelayMs: 500, + }) + ).toBe(2 * 5_000 + 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, + disconnectTimeoutMs: 2_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, + 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 bf02a57e48d..933fceae3ec 100644 --- a/apps/sim/lib/core/config/redis.test.ts +++ b/apps/sim/lib/core/config/redis.test.ts @@ -47,14 +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(() => { @@ -471,7 +476,50 @@ describe('redis config', () => { }) }) + 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) + }) + }) + describe('warmRedisConnection', () => { + it('outlasts one dead handshake so the reconnect can be what warms it', async () => { + mockRedisInstance.status = 'connecting' + const warm = warmRedisConnection() + + // 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') + + 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( + coldConnectionBudgetMs({ + connectTimeoutMs: CONNECT_TIMEOUT_MS, + commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS, + disconnectTimeoutMs: DISCONNECT_TIMEOUT_MS, + reconnectDelayMs: sharedReconnectDelayMs(1, 1), + }) + ) + + 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..7ca7889952d 100644 --- a/apps/sim/lib/core/config/redis.ts +++ b/apps/sim/lib/core/config/redis.ts @@ -2,9 +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 Redis, { type RedisOptions } from 'ioredis' +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') @@ -42,13 +43,24 @@ 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 { +export interface RedisConnectionDefaults { + keepAlive: number + connectTimeout: number + disconnectTimeout: number + enableOfflineQueue: boolean + tls?: { servername: string } +} + +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: 10000, + connectTimeout: CONNECT_TIMEOUT_MS, + disconnectTimeout: DISCONNECT_TIMEOUT_MS, enableOfflineQueue: true, ...(tls ? { tls } : {}), } @@ -203,12 +215,34 @@ export function describeRedisConnection( const PING_INTERVAL_MS = 15_000 const MAX_PING_FAILURES = 2 +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 + +/** + * 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 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. 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 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. */ -const REDIS_WARMUP_TIMEOUT_MS = 10_000 +const REDIS_WARMUP_TIMEOUT_MS = coldConnectionBudgetMs({ + connectTimeoutMs: CONNECT_TIMEOUT_MS, + commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS, + disconnectTimeoutMs: DISCONNECT_TIMEOUT_MS, + reconnectDelayMs: sharedReconnectDelayMs(1, 1), +}) export function getConfiguredRedisUrl(): string | null { if (getConfiguredCacheProvider() === 'database') return null @@ -296,7 +330,7 @@ export function getRedisClient(): Redis | null { state.client = new Redis(redisUrl, { ...defaults, - commandTimeout: 5000, + commandTimeout: SHARED_COMMAND_TIMEOUT_MS, maxRetriesPerRequest: 5, retryStrategy: (times) => { @@ -304,9 +338,7 @@ 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 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 4b73887d294..e29f92c1a04 100644 --- a/apps/sim/lib/execution/execution-signal.test.ts +++ b/apps/sim/lib/execution/execution-signal.test.ts @@ -8,17 +8,22 @@ const { connection, mockRedisUrl, mockSubscribe, mockUnsubscribe } = vi.hoisted( connection: { status: 'ready', client: undefined as EventEmitter | undefined, + options: undefined as Record | undefined, + }, + mockRedisUrl: { + value: 'redis://localhost:6379' as string | undefined, + error: undefined as Error | undefined, }, - mockRedisUrl: { value: 'redis://localhost:6379' as string | undefined }, mockSubscribe: vi.fn(), mockUnsubscribe: vi.fn(), })) vi.mock('ioredis', () => ({ default: class extends EventEmitter { - constructor() { + constructor(_url: string, options: Record) { super() connection.client = this + connection.options = options } get status() { @@ -31,15 +36,38 @@ vi.mock('ioredis', () => ({ })) vi.mock('@/lib/core/config/redis', () => ({ - getConfiguredRedisUrl: () => mockRedisUrl.value, - getRedisConnectionDefaults: () => ({}), + getConfiguredRedisUrl: () => { + if (mockRedisUrl.error) throw mockRedisUrl.error + return mockRedisUrl.value + }, + // 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, } 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 + disconnectTimeout: number + retryStrategy: (attempt: number) => number + } + return coldConnectionBudgetMs({ + connectTimeoutMs: options.connectTimeout, + commandTimeoutMs: options.commandTimeout, + disconnectTimeoutMs: options.disconnectTimeout, + reconnectDelayMs: options.retryStrategy(1), + }) +} + describe('ExecutionSignalHub', () => { beforeEach(() => { vi.clearAllMocks() @@ -48,6 +76,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 +270,7 @@ describe('ExecutionSignalHub', () => { 'Timed out waiting for Redis subscriber readiness' ) - const timeout = vi.advanceTimersByTimeAsync(4000).then(() => { + const timeout = vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000).then(() => { connection.client?.emit('error', new Error('ECONNREFUSED')) return vi.advanceTimersByTimeAsync(1000) }) @@ -356,6 +385,119 @@ describe('ExecutionSignalHub', () => { expect(replacement).not.toHaveBeenCalledWith('unavailable') }) + 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() + } + }) + + it('gives a subscribe that joins another in-flight wait its own full budget', async () => { + vi.useFakeTimers() + try { + connection.status = 'connect' + 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 second = hub.subscribe('execution-second', vi.fn()) + const secondSettled = vi.fn() + void second.then(secondSettled, secondSettled) + + await vi.advanceTimersByTimeAsync(1000) + 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 second + expect(mockSubscribe).toHaveBeenCalledWith( + 'execution:signal:execution-second', + 'execution:cancel' + ) + } finally { + vi.useRealTimers() + } + }) + + 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() + + // 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() + } + }) + + it('begins connecting when asked to connect ahead of a subscription', () => { + connection.status = 'connecting' + + connectExecutionSignalHub() + + // Constructing the hub is what dials; the client exists before any subscribe. + expect(connection.client).toBeDefined() + expect(mockSubscribe).not.toHaveBeenCalled() + }) + + 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') + + expect(() => connectExecutionSignalHub()).not.toThrow() + expect(connection.client).toBeUndefined() + }) + 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..65554f0f16c 100644 --- a/apps/sim/lib/execution/execution-signal.ts +++ b/apps/sim/lib/execution/execution-signal.ts @@ -3,10 +3,18 @@ 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 } from '@/lib/core/config/redis-budget' const logger = createLogger('ExecutionSignalHub') const EXECUTION_SIGNAL_PREFIX = 'execution:signal:' -const SUBSCRIBER_TIMEOUT_MS = 5000 +/** + * 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 = 5_000 +const subscriberRetryDelayMs = (attempt: number): number => Math.min(attempt * 500, 5000) export const LEGACY_EXECUTION_CANCEL_CHANNEL = 'execution:cancel' export type ExecutionSignalReason = 'event' | 'cancelled' | 'reconnected' | 'unavailable' @@ -17,6 +25,17 @@ interface ChannelSubscription { acknowledged: boolean } +/** + * 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 + detach: () => void + waiters: number +} + export interface ExecutionSignalHub { subscribe(executionId: string, handler: ExecutionSignalHandler): Promise<() => void> } @@ -27,19 +46,36 @@ export function getExecutionSignalChannel(executionId: string): string { class RedisExecutionSignalHub implements ExecutionSignalHub { private readonly subscriber: Redis + /** + * 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>() private readonly subscriptions = new Map() - private connectionReady: Promise | undefined + private readySignal: ReadySignal | undefined private connectedOnce = false 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.readyTimeoutMs = coldConnectionBudgetMs({ + connectTimeoutMs: options.connectTimeout, + commandTimeoutMs: options.commandTimeout, + disconnectTimeoutMs: options.disconnectTimeout, + reconnectDelayMs: subscriberRetryDelayMs(1), + }) this.subscriber = new Redis(redisUrl, options) this.subscriber.on('message', (channel: string, message: string) => { if (channel === LEGACY_EXECUTION_CANCEL_CHANNEL) { @@ -131,37 +167,65 @@ class RedisExecutionSignalHub implements ExecutionSignalHub { await this.subscriber.subscribe(...channels) } + /** + * 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 { - if (this.connectionReady) return this.connectionReady if (this.subscriber.status === 'end') { return Promise.reject(new Error('Redis subscriber connection ended')) } + const signal = (this.readySignal ??= this.createReadySignal()) + signal.waiters++ + let timer: NodeJS.Timeout | undefined + const deadline = new Promise((_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 + } + }) + } - this.connectionReady = new Promise((resolve, reject) => { - const cleanup = () => { - clearTimeout(timeout) - this.subscriber.removeListener('ready', onReady) - this.subscriber.removeListener('end', onEnd) + /** + * 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 { + const signal: ReadySignal = { promise: Promise.resolve(), detach: () => undefined, waiters: 0 } + signal.promise = new Promise((resolve, reject) => { + const settle = () => { + signal.detach() + if (this.readySignal === signal) this.readySignal = undefined } const onReady = () => { - cleanup() + settle() resolve() } - const fail = (error: Error) => { - cleanup() - reject(error) + const onEnd = () => { + settle() + reject(new Error('Redis subscriber connection ended')) + } + signal.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_TIMEOUT_MS - ) this.subscriber.once('ready', onReady) this.subscriber.once('end', onEnd) - }).finally(() => { - this.connectionReady = undefined }) - return this.connectionReady + signal.promise.catch(() => undefined) + return signal } private async handleReady(): Promise { @@ -261,6 +325,24 @@ export function getExecutionSignalHub(): ExecutionSignalHub { return executionSignalGlobal._executionSignalHub } +/** + * 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 function connectExecutionSignalHub(): void { + try { + getExecutionSignalHub() + } catch { + return + } +} + export function publishLocalExecutionSignal( executionId: string, reason: Extract diff --git a/apps/sim/lib/workflows/executor/execution-core.test.ts b/apps/sim/lib/workflows/executor/execution-core.test.ts index b5bd98d9446..78e03f78577 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 { connectExecutionSignalHubMock } = vi.hoisted(() => ({ + connectExecutionSignalHubMock: vi.fn(), +})) + +vi.mock('@/lib/execution/execution-signal', () => ({ + connectExecutionSignalHub: connectExecutionSignalHubMock, +})) + vi.mock('@/lib/core/security/encryption', () => ({ decryptSecret: decryptSecretMock, })) @@ -376,6 +384,21 @@ describe('executeWorkflowCore terminal finalization sequencing', () => { expect(executorConstructorMock).toHaveBeenCalledTimes(1) }) + it('begins connecting the signal subscriber synchronously, before the first await', async () => { + const executionPromise = executeWorkflowCore({ + snapshot: createSnapshot() as unknown as ExecutionSnapshot, + callbacks: {}, + loggingSession: loggingSession as unknown as LoggingSession, + }) + + // 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(connectExecutionSignalHubMock).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..a465c6b48e4 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 { 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' @@ -377,10 +378,19 @@ 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 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 { + connectExecutionSignalHub() 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 6a218121ee5..7966fdbf94a 100644 --- a/apps/sim/trigger.config.ts +++ b/apps/sim/trigger.config.ts @@ -102,11 +102,15 @@ 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 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 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 */ 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 9e5e548f2c8..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 } : {}), } @@ -91,6 +93,8 @@ export const redisConfigMockFns = { mockCloseRedisConnection: vi.fn().mockResolvedValue(undefined), mockResetForTesting: vi.fn(), mockDescribeRedisConnection: vi.fn(describeRedisConnectionImpl), + mockWarmRedisConnection: vi.fn().mockResolvedValue(false), + mockSharedReconnectDelayMs: vi.fn().mockReturnValue(1_000), } /** @@ -108,6 +112,8 @@ export function resetRedisConfigMock(): void { redisConfigMockFns.mockExtendLock.mockReset().mockResolvedValue(true) redisConfigMockFns.mockCloseRedisConnection.mockReset().mockResolvedValue(undefined) redisConfigMockFns.mockResetForTesting.mockReset() + redisConfigMockFns.mockWarmRedisConnection.mockReset().mockResolvedValue(false) + redisConfigMockFns.mockSharedReconnectDelayMs.mockReset().mockReturnValue(1_000) redisConfigMockFns.mockDescribeRedisConnection .mockReset() .mockImplementation(describeRedisConnectionImpl) @@ -123,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, @@ -133,4 +142,6 @@ export const redisConfigMock = { closeRedisConnection: redisConfigMockFns.mockCloseRedisConnection, resetForTesting: redisConfigMockFns.mockResetForTesting, describeRedisConnection: redisConfigMockFns.mockDescribeRedisConnection, + warmRedisConnection: redisConfigMockFns.mockWarmRedisConnection, + sharedReconnectDelayMs: redisConfigMockFns.mockSharedReconnectDelayMs, }