perf(webapp): scale the realtime runs feed under high concurrency

Keep a single busy environment from overwhelming the realtime runs feed:

- Coalesce per-environment wake notifications to a bounded rate, so a
  high-throughput environment wakes its subscribers at a steady cap
  instead of once per run change.
- Hold a multi-run live poll open on an empty result instead of returning
  an immediate up-to-date, cutting wasted round-trips when a wake does not
  match a subscriber's filter.
- Support Redis Cluster sharded pub/sub (SSUBSCRIBE/SPUBLISH) so the feed's
  pub/sub scales horizontally across shards by environment/run.

All behind the existing feature flag and tunable env vars. Bumps ioredis to
5.6.x across the workspace (required for cluster sharded pub/sub).

Author: ericallam

apps/supervisor/package.json

@@ -18,7 +18,7 @@
     "@kubernetes/client-node": "^1.0.0",
     "@trigger.dev/core": "workspace:*",
     "dockerode": "^4.0.6",
-    "ioredis": "^5.3.2",
+    "ioredis": "~5.6.0",
     "p-limit": "^6.2.0",
     "prom-client": "^15.1.0",
     "socket.io": "4.7.4",

apps/webapp/app/env.server.ts

@@ -324,6 +324,15 @@ const EnvironmentSchema = z
     // hydrate cache entry. Floored, so the window only ever widens by < bucket. 0
     // disables bucketing (each feed keeps its exact lower bound).
     REALTIME_NOTIFIER_RUNSET_CREATED_AT_BUCKET_MS: z.coerce.number().int().default(60_000),
+    // Leading-edge throttle (ms) on the per-env wake channel: a busy env's run-change
+    // firehose is collapsed to at most one feed-wake per window, decoupling wake load
+    // from run throughput. Lossless because consumers refetch current state on a wake.
+    // 0 disables coalescing (every change wakes immediately).
+    REALTIME_NOTIFIER_ENV_WAKE_COALESCE_WINDOW_MS: z.coerce.number().int().default(100),
+    // When "1", a multi-run live poll woken by a change irrelevant to its filter keeps
+    // holding the long-poll (re-resolving cheaply) instead of returning an empty
+    // up-to-date the client would immediately re-issue. "0" reverts to per-wake replies.
+    REALTIME_NOTIFIER_HOLD_ON_EMPTY: z.string().default("1"),
 
     PUBSUB_REDIS_HOST: z
       .string()
@@ -387,6 +396,10 @@ const EnvironmentSchema = z
     REALTIME_RUNS_PUBSUB_REDIS_CLUSTER_MODE_ENABLED: z
       .string()
       .default(process.env.PUBSUB_REDIS_CLUSTER_MODE_ENABLED ?? "0"),
+    // Use sharded pub/sub (SSUBSCRIBE/SPUBLISH) when in cluster mode, so a busy env's
+    // traffic stays on one shard instead of broadcasting to every node. Only takes
+    // effect alongside CLUSTER_MODE_ENABLED. "0" forces classic pub/sub on the cluster.
+    REALTIME_RUNS_PUBSUB_REDIS_SHARDED_ENABLED: z.string().default("1"),
 
     DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
     DEFAULT_ENV_EXECUTION_CONCURRENCY_BURST_FACTOR: z.coerce.number().default(1.0),

apps/webapp/app/services/realtime/notifierRealtimeClient.server.ts

@@ -91,6 +91,11 @@ export type NotifierRealtimeClientOptions = {
    * same-tag feeds pinned within the same bucket share a cache entry. Defaults to
    * 60000. 0 disables bucketing. */
   runSetCreatedAtBucketMs?: number;
+  /** When true (default), a multi-run live poll woken by a change irrelevant to its
+   * filter keeps holding the long-poll (re-resolving cheaply on each wake) instead of
+   * returning an empty up-to-date the client would immediately re-issue. The empty
+   * response is the dominant cost under a busy per-env wake channel. */
+  holdOnEmpty?: boolean;
   /** Observability hook: why a live request woke (notify vs timeout vs abort). */
   onWakeup?: (reason: WakeupReason) => void;
   /** Observability hook: whether a multi-run resolve hit the cache, coalesced onto
@@ -417,55 +422,93 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
     signal: AbortSignal | undefined
   ): Promise<Response> {
     return this.#withConcurrencySlot(environment, async () => {
-      // One env-scoped subscription per feed (not one per run): any run change in
-      // the env wakes us, then we re-resolve the filter.
-      const reason = await this.#waitForEnvChange(environment.id, signal);
-      this.options.onWakeup?.(reason);
-
-      const cached = this.#workingSetCache.get(handle);
       const offsetFloorMs = parseOffsetUpdatedAtMs(offset);
-      const seq = this.#nextSeq();
+      // Total time to hold this long-poll, jittered to avoid synchronized refetch herds.
+      const deadline = Date.now() + this.#jitteredTimeout();
+      const holdOnEmpty = this.options.holdOnEmpty ?? true;
+
+      // Working set we diff against: seeded from the cache (or the offset floor on a
+      // miss) and advanced on each refetch within this held request.
+      let prevSeen = this.#workingSetCache.get(handle);
+
+      // The per-env channel wakes this feed on ANY run change in the environment, but
+      // most changes don't match this feed's filter. Rather than return an empty
+      // up-to-date the client would immediately re-issue (the dominant cost under a
+      // busy env), we hold the connection and only respond when THIS feed has a real
+      // delta or the backstop elapses. Each wake re-resolves via the coalesced +
+      // short-TTL cache, so an env-wide wake never fans out into per-feed CH+PG queries.
+      while (true) {
+        const remaining = deadline - Date.now();
+        // One env-scoped subscription per wait (not one per run); re-subscribed each
+        // loop until a relevant delta or the budget runs out.
+        const reason =
+          remaining > 0 ? await this.#waitForEnvChange(environment.id, signal, remaining) : "timeout";
+        this.options.onWakeup?.(reason);
+
+        if (reason === "abort") {
+          // Client disconnected; the response is discarded. Skip the refetch.
+          return this.#buildResponse(buildUpToDateBody(), apiVersion, clientVersion, {
+            offset,
+            handle,
+            cursor: String(this.#nextSeq()),
+          });
+        }
 
-      // ClickHouse resolves the (possibly stale) membership; Postgres hydrates the
-      // authoritative current rows, so status is always fresh even if CH lags. The
-      // resolve+hydrate is coalesced + short-TTL cached so a single env-wide wake
-      // doesn't fan out into one CH+PG query per concurrent same-filter feed.
-      const rows = await this.#resolveAndHydrate(environment, filter, skipColumns);
-
-      // Diff against what the client already has, using the hydrated updatedAt:
-      // cache hit => per-row (new = insert, advanced = update); miss => anything
-      // newer than the offset floor as a merge-safe update.
-      const changes: RowChange[] = [];
-      const seen: WorkingSet = new Map();
-      let maxUpdatedAt = offsetFloorMs;
-      for (const row of rows) {
-        const updatedAtMs = row.updatedAt.getTime();
-        seen.set(row.id, updatedAtMs);
-        maxUpdatedAt = Math.max(maxUpdatedAt, updatedAtMs);
-
-        if (cached) {
-          const prior = cached.get(row.id);
-          if (prior === undefined) {
-            changes.push({ row, operation: "insert" });
-          } else if (updatedAtMs > prior) {
+        // ClickHouse resolves the (possibly stale) membership; Postgres hydrates the
+        // authoritative current rows, so status is always fresh even if CH lags. We
+        // refetch on every wake AND on the final timeout, so a wake missed during the
+        // brief re-subscribe gap is still caught by the backstop.
+        const rows = await this.#resolveAndHydrate(environment, filter, skipColumns);
+
+        // Diff against what the client already has, using the hydrated updatedAt:
+        // prior working set => per-row (new = insert, advanced = update); miss =>
+        // anything newer than the offset floor as a merge-safe update.
+        const changes: RowChange[] = [];
+        const seen: WorkingSet = new Map();
+        let maxUpdatedAt = offsetFloorMs;
+        for (const row of rows) {
+          const updatedAtMs = row.updatedAt.getTime();
+          seen.set(row.id, updatedAtMs);
+          maxUpdatedAt = Math.max(maxUpdatedAt, updatedAtMs);
+
+          if (prevSeen) {
+            const prior = prevSeen.get(row.id);
+            if (prior === undefined) {
+              changes.push({ row, operation: "insert" });
+            } else if (updatedAtMs > prior) {
+              changes.push({ row, operation: "update" });
+            }
+          } else if (updatedAtMs > offsetFloorMs) {
             changes.push({ row, operation: "update" });
           }
-        } else if (updatedAtMs > offsetFloorMs) {
-          changes.push({ row, operation: "update" });
         }
-      }
-
-      // Refresh the working set so runs that left the filter stop being tracked
-      // (the client keeps showing them; the SDK never applies deletes).
-      this.#workingSetCache.set(handle, seen);
 
-      const body = changes.length === 0 ? buildUpToDateBody() : buildRowsBody(changes, skipColumns);
+        // Refresh the working set so runs that left the filter stop being tracked
+        // (the client keeps showing them; the SDK never applies deletes).
+        this.#workingSetCache.set(handle, seen);
+        prevSeen = seen;
+
+        if (changes.length > 0) {
+          const seq = this.#nextSeq();
+          return this.#buildResponse(buildRowsBody(changes, skipColumns), apiVersion, clientVersion, {
+            offset: encodeOffset(maxUpdatedAt, seq),
+            handle,
+            cursor: String(seq),
+          });
+        }
 
-      return this.#buildResponse(body, apiVersion, clientVersion, {
-        offset: encodeOffset(maxUpdatedAt, seq),
-        handle,
-        cursor: String(seq),
-      });
+        // Empty diff. With hold-on-empty (default) keep waiting until a real delta or
+        // the budget elapses; otherwise fall back to the legacy per-wake up-to-date.
+        if (reason === "timeout" || !holdOnEmpty) {
+          const seq = this.#nextSeq();
+          return this.#buildResponse(buildUpToDateBody(), apiVersion, clientVersion, {
+            offset: encodeOffset(maxUpdatedAt, seq),
+            handle,
+            cursor: String(seq),
+          });
+        }
+        // reason === "notify" with an empty diff: keep holding (loop, re-subscribe).
+      }
     });
   }
 
@@ -654,21 +697,33 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
     }
   }
 
-  #waitForChange(runId: string, signal?: AbortSignal): Promise<WakeupReason> {
-    return this.#waitForSubscription(this.options.notifier.subscribeToRunChanges(runId), signal);
+  #waitForChange(runId: string, signal?: AbortSignal, timeoutMs?: number): Promise<WakeupReason> {
+    return this.#waitForSubscription(
+      this.options.notifier.subscribeToRunChanges(runId),
+      signal,
+      timeoutMs
+    );
   }
 
-  #waitForEnvChange(environmentId: string, signal?: AbortSignal): Promise<WakeupReason> {
+  #waitForEnvChange(
+    environmentId: string,
+    signal?: AbortSignal,
+    timeoutMs?: number
+  ): Promise<WakeupReason> {
     return this.#waitForSubscription(
       this.options.notifier.subscribeToEnvChanges(environmentId),
-      signal
+      signal,
+      timeoutMs
     );
   }
 
-  /** Race a notifier subscription against the backstop timeout and the abort signal. */
+  /** Race a notifier subscription against a timeout (the jittered backstop by default,
+   * or an explicit remaining budget when a live request holds across wakes) and the
+   * abort signal. */
   async #waitForSubscription(
     subscription: RunChangeSubscription,
-    signal?: AbortSignal
+    signal?: AbortSignal,
+    timeoutMs?: number
   ): Promise<WakeupReason> {
     if (signal?.aborted) {
       subscription.unsubscribe();
@@ -682,7 +737,7 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
       return await new Promise<WakeupReason>((resolve) => {
         subscription.changed.then(() => resolve("notify")).catch(() => resolve("timeout"));
 
-        timer = setTimeout(() => resolve("timeout"), this.#jitteredTimeout());
+        timer = setTimeout(() => resolve("timeout"), timeoutMs ?? this.#jitteredTimeout());
 
         if (signal) {
           onAbort = () => resolve("abort");

apps/webapp/app/services/realtime/notifierRealtimeClientInstance.server.ts

@@ -77,6 +77,7 @@ function initializeNotifierRealtimeClient(): NotifierRealtimeClient {
     runSetResolveCacheMaxEntries: env.REALTIME_NOTIFIER_RUNSET_CACHE_MAX_ENTRIES,
     listCacheMaxEntries: env.REALTIME_NOTIFIER_WORKING_SET_MAX_ENTRIES,
     runSetCreatedAtBucketMs: env.REALTIME_NOTIFIER_RUNSET_CREATED_AT_BUCKET_MS,
+    holdOnEmpty: env.REALTIME_NOTIFIER_HOLD_ON_EMPTY === "1",
     onWakeup: (reason) => wakeups.inc({ reason }),
     onRunSetResolve: (result) => runSetResolves.inc({ result }),
     onRunSetQuery: (stage, ms) => runSetQueryMs.observe({ stage }, ms),