improvement(redis): warm the signal subscriber on intent, not at worker start

Only the tasks that execute a workflow ever subscribe to cancellation signals,
and they are a minority of runs: over a week, 71% of Trigger.dev runs never
subscribe, and a single document-processing task is 61% on its own. Warming
the subscriber in the global init hook opened a TLS connection on every one
of those runs that nothing would use.

Warm it instead at `executeWorkflowCore` — the one path every execution
shares — fire-and-forget as the first statement, so the handshake overlaps the
custom-block read and preprocessing ahead of the cancellation subscribe rather
than being paid inside that subscribe's readiness budget. No task-id list to
keep in sync: coverage follows from the funnel. The subscribe and the warm-up
share the hub's memoized readiness wait, so their budgets never compound
within a run.

The global init keeps warming only the shared client, which nearly every task
uses; the Next server keeps warming both at boot, one connection per
long-lived process.

Author: waleedlatif1

apps/sim/lib/execution/execution-signal.ts

@@ -306,10 +306,13 @@ export function getExecutionSignalHub(): ExecutionSignalHub {
 }
 
 /**
- * Establishes the hub's subscriber connection ahead of the first execution, so
- * a run's cancellation subscription does not pay the handshake inside its own
- * readiness budget. Never throws: this runs from process start-up hooks where
- * a throw would fail the run, and a cold hub is only slower, not wrong.
+ * Establishes the hub's subscriber connection ahead of a cancellation
+ * subscription, so that subscribe does not pay the handshake inside its own
+ * readiness budget. Called fire-and-forget at the execution entry point — the
+ * one path every execution shares, early enough to overlap the work ahead of
+ * the subscribe — and at boot in long-lived servers. Never throws or rejects:
+ * a throw from a start-up hook would fail the run, and a cold hub is only
+ * slower, not wrong.
  */
 export async function warmExecutionSignalHub(): Promise<boolean> {
   let hub: ExecutionSignalHub

apps/sim/lib/workflows/executor/execution-core.test.ts

@@ -97,6 +97,14 @@ vi.mock('@/lib/execution/cancellation', () => ({
   clearExecutionCancellation: clearExecutionCancellationMock,
 }))
 
+const { warmExecutionSignalHubMock } = vi.hoisted(() => ({
+  warmExecutionSignalHubMock: vi.fn(),
+}))
+
+vi.mock('@/lib/execution/execution-signal', () => ({
+  warmExecutionSignalHub: warmExecutionSignalHubMock,
+}))
+
 vi.mock('@/lib/core/security/encryption', () => ({
   decryptSecret: decryptSecretMock,
 }))
@@ -376,6 +384,23 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     expect(executorConstructorMock).toHaveBeenCalledTimes(1)
   })
 
+  it('begins warming the signal subscriber synchronously, before the first await', async () => {
+    warmExecutionSignalHubMock.mockResolvedValue(true)
+
+    const executionPromise = executeWorkflowCore({
+      snapshot: createSnapshot() as any,
+      callbacks: {},
+      loggingSession: loggingSession as any,
+    })
+
+    // 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(warmExecutionSignalHubMock).toHaveBeenCalledOnce()
+
+    await executionPromise
+  })
+
   it('routes onBlockStart through logging session persistence path', async () => {
     executorExecuteMock.mockResolvedValue({
       success: true,

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 { warmExecutionSignalHub } 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'
@@ -381,6 +382,14 @@ async function finalizeExecutionError(params: {
 export async function executeWorkflowCore(
   options: ExecuteWorkflowCoreOptions
 ): Promise<ExecutionResult> {
+  // First, and not awaited: every execution subscribes to cancellation signals
+  // once its engine starts, so the subscriber's handshake is begun here — the
+  // one path all of them share — and overlaps the reads and preprocessing
+  // ahead of that subscribe instead of being paid inside its readiness budget.
+  // Warming 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.
+  void warmExecutionSignalHub()
   const workspaceId = options.snapshot.metadata.workspaceId
   const rows = workspaceId
     ? await withDatabaseReadRetry(() => getCustomBlockRowsForWorkspace(workspaceId), {

apps/sim/trigger.config.ts

@@ -102,24 +102,22 @@ 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 both Redis connections a run opens — the shared client, whose
-   * first call is typically a lock acquire, and the execution-signal subscriber,
-   * whose first call is the cancellation subscription — because each would
-   * otherwise pay its handshake inside its own deadline. Warmed in parallel so
-   * the run waits for the slower of the two, not the sum; awaited so both are up
+   * 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 clients; and neither ever
-   * throws, because a throw here fails the run.
+   * 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
    */
   init: async () => {
     markInsideTriggerRun()
-    const [{ warmRedisConnection }, { warmExecutionSignalHub }] = await Promise.all([
-      import('./lib/core/config/redis'),
-      import('./lib/execution/execution-signal'),
-    ])
-    await Promise.all([warmRedisConnection(), warmExecutionSignalHub()])
+    const { warmRedisConnection } = await import('./lib/core/config/redis')
+    await warmRedisConnection()
   },
   ...(grafanaTelemetry ? { telemetry: grafanaTelemetry } : {}),
   build: {