refactor(webapp): rename the realtime backend to native and tune it via env vars

The new realtime runs backend is now named native instead of notifier:
the feature flag selects electric | native | shadow, and all of its env
vars share the REALTIME_BACKEND_NATIVE_ prefix, including the dedicated
pub/sub Redis and ClickHouse pool configs.

Remaining hardcoded tunables (live-poll jitter, working-set TTL,
single-run cache, backend flag cache, default concurrency limit) moved
to env vars with unchanged defaults. Backend selection reuses the org
feature flags already loaded by request auth instead of an extra
organization read, and its caches survive dev-mode reloads. Oversized
comment blocks trimmed throughout.

Author: ericallam

apps/webapp/app/entry.server.tsx

@@ -270,8 +270,8 @@ process.on("uncaughtException", (error, origin) => {
 
 singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers);
 singleton("SetupBatchQueueCallbacks", setupBatchQueueCallbacks);
-// Attach the run-changed notifier delegations to the engine event bus.
-// No-ops (registers nothing) unless REALTIME_NOTIFIER_ENABLED=1.
+// Attach the realtime run-changed publish delegations to the engine event bus.
+// No-ops (registers nothing) unless REALTIME_BACKEND_NATIVE_ENABLED=1.
 singleton("RunChangeNotifierHandlers", registerRunChangeNotifierHandlers);
 
 // Wrapped in singleton() so Remix's dev-mode CJS reloads don't append

apps/webapp/app/env.server.ts

@@ -300,46 +300,36 @@ const EnvironmentSchema = z
       .int()
       .default(24 * 60 * 60 * 1000), // 1 day in milliseconds
 
-    // Master switch for the notifier-backed realtime feed.
-    // "0" (default) = the existing realtime path serves everything, publishes are
-    // no-ops, and no notifier Redis connections are opened (zero-overhead off).
-    // "1" = run-changed signals are published and the per-org `realtimeBackend`
-    // feature flag selects the backend per request.
-    REALTIME_NOTIFIER_ENABLED: z.string().default("0"),
-    // Backstop wait before a live notifier request refetches the run (ms). Matches
-    // Electric's ~20s live long-poll hold so the client polling cadence is unchanged
-    // across backends (a ±15% jitter is applied per request to avoid refetch herds).
-    REALTIME_NOTIFIER_LIVE_POLL_TIMEOUT_MS: z.coerce.number().int().default(20_000),
-    // Hard cap on the tag-list snapshot size served by the notifier feed.
-    REALTIME_NOTIFIER_MAX_LIST_RESULTS: z.coerce.number().int().default(1_000),
-    // Short-TTL coalescing cache for the multi-run (tag-list/batch) resolve+hydrate.
-    // Concurrent same-filter feeds share one ClickHouse resolve + Postgres hydrate
-    // within this window, so an env-wide wake doesn't fan out into per-feed queries.
-    // Staleness budget: a newly-matching run is visible within ~ttl + poll interval.
-    REALTIME_NOTIFIER_RUNSET_CACHE_TTL_MS: z.coerce.number().int().default(1_000),
-    REALTIME_NOTIFIER_RUNSET_CACHE_MAX_ENTRIES: z.coerce.number().int().default(5_000),
-    // Cap on the per-handle working-set cache (runId -> updatedAt) the notifier keeps
-    // for diffing multi-run live polls.
-    REALTIME_NOTIFIER_WORKING_SET_MAX_ENTRIES: z.coerce.number().int().default(10_000),
-    // Quantize the tag-list createdAt lower bound to this epoch-aligned bucket (ms) so
-    // same-tag feeds that pin their window within the same bucket share one resolve+
-    // 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"),
-    // Max concurrent fresh ClickHouse resolves (cache misses) per instance. Caps the
-    // distinct-filter reconnect stampede: a mass reconnect of N feeds on N different filters
-    // queues to this many concurrent CH queries instead of firing all N at once. Same-filter
-    // bursts collapse via the single-flight cache before taking a permit. 0 disables the gate.
-    REALTIME_NOTIFIER_RESOLVE_ADMISSION_LIMIT: z.coerce.number().int().default(16),
+    // Master switch for the native realtime backend; off = Electric serves everything, publishes no-op.
+    REALTIME_BACKEND_NATIVE_ENABLED: z.string().default("0"),
+    // Live long-poll backstop hold (ms); matches Electric's ~20s cadence.
+    REALTIME_BACKEND_NATIVE_LIVE_POLL_TIMEOUT_MS: z.coerce.number().int().default(20_000),
+    // Jitter ratio on the live-poll hold (0.15 = ±15%) to avoid synchronized refetch herds.
+    REALTIME_BACKEND_NATIVE_LIVE_POLL_JITTER_RATIO: z.coerce.number().default(0.15),
+    // Hard cap on the tag-list snapshot size.
+    REALTIME_BACKEND_NATIVE_MAX_LIST_RESULTS: z.coerce.number().int().default(1_000),
+    // TTL/size of the coalescing cache for the multi-run resolve+hydrate (same-filter feeds share one query).
+    REALTIME_BACKEND_NATIVE_RUNSET_CACHE_TTL_MS: z.coerce.number().int().default(1_000),
+    REALTIME_BACKEND_NATIVE_RUNSET_CACHE_MAX_ENTRIES: z.coerce.number().int().default(5_000),
+    // Size/TTL of the per-handle working-set cache used to diff multi-run live polls.
+    REALTIME_BACKEND_NATIVE_WORKING_SET_MAX_ENTRIES: z.coerce.number().int().default(10_000),
+    REALTIME_BACKEND_NATIVE_WORKING_SET_TTL_MS: z.coerce.number().int().default(300_000),
+    // Bucket (ms) the tag-list createdAt floor is quantized to so same-tag feeds share a cache entry; 0 disables.
+    REALTIME_BACKEND_NATIVE_RUNSET_CREATED_AT_BUCKET_MS: z.coerce.number().int().default(60_000),
+    // Leading-edge throttle (ms) on per-env wake delivery; 0 wakes on every change.
+    REALTIME_BACKEND_NATIVE_ENV_WAKE_COALESCE_WINDOW_MS: z.coerce.number().int().default(100),
+    // "1" holds a multi-run live poll open on a non-matching wake instead of replying up-to-date.
+    REALTIME_BACKEND_NATIVE_HOLD_ON_EMPTY: z.string().default("1"),
+    // Max concurrent fresh ClickHouse resolves per instance (reconnect-stampede gate); 0 disables.
+    REALTIME_BACKEND_NATIVE_RESOLVE_ADMISSION_LIMIT: z.coerce.number().int().default(16),
+    // Fallback per-env concurrent-connection limit when the org has none configured.
+    REALTIME_BACKEND_NATIVE_DEFAULT_CONCURRENCY_LIMIT: z.coerce.number().int().default(100_000),
+    // TTL/size of the single-run read-through cache that collapses duplicate refetch bursts.
+    REALTIME_BACKEND_NATIVE_RUN_CACHE_TTL_MS: z.coerce.number().int().default(250),
+    REALTIME_BACKEND_NATIVE_RUN_CACHE_MAX_ENTRIES: z.coerce.number().int().default(5_000),
+    // TTL/size of the per-org realtimeBackend flag cache used to pick the serving backend.
+    REALTIME_BACKEND_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
+    REALTIME_BACKEND_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(50_000),
 
     PUBSUB_REDIS_HOST: z
       .string()
@@ -373,40 +363,35 @@ const EnvironmentSchema = z
     PUBSUB_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
     PUBSUB_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
-    // Dedicated pub/sub Redis for the realtime runs feed's run-changed notifier, so
-    // its publish/subscribe traffic can run on its own instance. Each value falls
-    // back to the shared PUBSUB_REDIS_* (then REDIS_*) when unset, so the default is
-    // unchanged until explicitly pointed at a dedicated instance.
-    REALTIME_RUNS_PUBSUB_REDIS_HOST: z
+    // Dedicated pub/sub Redis for the native realtime backend; falls back to PUBSUB_REDIS_* then REDIS_*.
+    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_HOST: z
       .string()
       .optional()
       .transform((v) => v ?? process.env.PUBSUB_REDIS_HOST ?? process.env.REDIS_HOST),
-    REALTIME_RUNS_PUBSUB_REDIS_PORT: z.coerce
+    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_PORT: z.coerce
       .number()
       .optional()
       .transform((v) => {
         if (v !== undefined) return v;
         const raw = process.env.PUBSUB_REDIS_PORT ?? process.env.REDIS_PORT;
         return raw ? parseInt(raw) : undefined;
       }),
-    REALTIME_RUNS_PUBSUB_REDIS_USERNAME: z
+    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_USERNAME: z
       .string()
       .optional()
       .transform((v) => v ?? process.env.PUBSUB_REDIS_USERNAME ?? process.env.REDIS_USERNAME),
-    REALTIME_RUNS_PUBSUB_REDIS_PASSWORD: z
+    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_PASSWORD: z
       .string()
       .optional()
       .transform((v) => v ?? process.env.PUBSUB_REDIS_PASSWORD ?? process.env.REDIS_PASSWORD),
-    REALTIME_RUNS_PUBSUB_REDIS_TLS_DISABLED: z
+    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_TLS_DISABLED: z
       .string()
       .default(process.env.PUBSUB_REDIS_TLS_DISABLED ?? process.env.REDIS_TLS_DISABLED ?? "false"),
-    REALTIME_RUNS_PUBSUB_REDIS_CLUSTER_MODE_ENABLED: z
+    REALTIME_BACKEND_NATIVE_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"),
+    // Use sharded pub/sub (SSUBSCRIBE/SPUBLISH) in cluster mode; "0" forces classic pub/sub.
+    REALTIME_BACKEND_NATIVE_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),
@@ -1684,20 +1669,18 @@ const EnvironmentSchema = z
       .enum(["log", "error", "warn", "info", "debug"])
       .default("info"),
     RUN_ENGINE_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
-    // ClickHouse client used by the realtime runs feed for tag/batch id resolution.
-    // Kept on its own URL + pool so the feed's reads can't contend with the main
-    // analytics client (CLICKHOUSE_URL). Falls back to the main URL when unset.
-    REALTIME_RUNS_CLICKHOUSE_URL: z
+    // Dedicated ClickHouse pool for the native backend's tag/batch id resolution; falls back to CLICKHOUSE_URL.
+    REALTIME_BACKEND_NATIVE_CLICKHOUSE_URL: z
       .string()
       .optional()
       .transform((v) => v ?? process.env.CLICKHOUSE_URL),
-    REALTIME_RUNS_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
-    REALTIME_RUNS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
-    REALTIME_RUNS_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
-    REALTIME_RUNS_CLICKHOUSE_LOG_LEVEL: z
+    REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
+    REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
+    REALTIME_BACKEND_NATIVE_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
+    REALTIME_BACKEND_NATIVE_CLICKHOUSE_LOG_LEVEL: z
       .enum(["log", "error", "warn", "info", "debug"])
       .default("info"),
-    REALTIME_RUNS_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
+    REALTIME_BACKEND_NATIVE_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
     EVENTS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().default(1000),
     EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     METRICS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().default(10000),

apps/webapp/app/routes/realtime.v1.batches.$batchId.ts

@@ -33,8 +33,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ authentication, request, resource: batchRun, apiVersion }) => {
-    // Pick the Electric proxy or the notifier-backed batch feed
-    // per org (defaults to Electric). Both implement streamBatch.
+    // Pick the Electric proxy or the native backend per org (defaults to Electric); both implement streamBatch.
     const client = await resolveRealtimeStreamClient(authentication.environment);
 
     return client.streamBatch(

apps/webapp/app/routes/realtime.v1.runs.$runId.ts

@@ -48,9 +48,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ authentication, request, resource: run, apiVersion }) => {
-    // Pick the Electric proxy or the notifier-backed shim per org (defaults to
-    // Electric; controlled by REALTIME_NOTIFIER_ENABLED + the realtimeBackend
-    // feature flag). Both implement the same streamRun contract.
+    // Pick the Electric proxy or the native backend per org (defaults to Electric); both implement streamRun.
     const client = await resolveRealtimeStreamClient(authentication.environment);
 
     return client.streamRun(
@@ -60,10 +58,7 @@ export const loader = createLoaderApiRoute(
       apiVersion,
       authentication.realtime,
       request.headers.get("x-trigger-electric-version") ?? undefined,
-      // Propagate abort on client disconnect so the upstream Electric long-poll
-      // fetch is cancelled too. Without this, undici buffers from the unconsumed
-      // upstream response body accumulate until Electric's poll timeout, causing
-      // steady RSS growth on api (see docs/runbooks for the H1 isolation test).
+      // Propagate abort on client disconnect so the upstream Electric long-poll is cancelled too, else undici buffers grow RSS until the poll timeout.
       getRequestAbortSignal()
     );
   }

apps/webapp/app/routes/realtime.v1.runs.ts

@@ -25,12 +25,7 @@ export const loader = createLoaderApiRoute(
     authorization: {
       action: "read",
       resource: (_, __, searchParams) =>
-        // Pre-RBAC, the resource was the searchParams object itself and
-        // the legacy `checkAuthorization` iterated `Object.keys`, so a
-        // JWT with type-level `read:tags` (no id) granted access to the
-        // unfiltered runs stream. Including `{ type: "tags" }` here
-        // preserves that — per-id `read:tags:<tag>` still grants only
-        // when the filter includes that tag.
+        // `{ type: "tags" }` preserves pre-RBAC type-level `read:tags` access to the unfiltered stream; per-id `read:tags:<tag>` still grants only when the filter includes that tag.
         anyResource([
           { type: "runs" },
           { type: "tags" },
@@ -39,8 +34,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ searchParams, authentication, request, apiVersion }) => {
-    // Pick the Electric proxy or the notifier-backed tag-list feed per org
-    // (defaults to Electric). Both implement streamRuns.
+    // Pick the Electric proxy or the native backend per org (defaults to Electric); both implement streamRuns.
     const client = await resolveRealtimeStreamClient(authentication.environment);
 
     return client.streamRuns(

apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts

@@ -211,33 +211,33 @@ function initializeRunEngineClickhouseClient(): ClickHouse {
   });
 }
 
-/** Realtime runs feed tag/batch id resolution (`REALTIME_RUNS_CLICKHOUSE_URL`);
+/** Realtime runs feed tag/batch id resolution (`REALTIME_BACKEND_NATIVE_CLICKHOUSE_URL`);
  *  falls back to the default client if unset. */
 const defaultRealtimeClickhouseClient = singleton(
   "realtimeClickhouseClient",
   initializeRealtimeClickhouseClient
 );
 
 function initializeRealtimeClickhouseClient(): ClickHouse {
-  if (!env.REALTIME_RUNS_CLICKHOUSE_URL) {
+  if (!env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_URL) {
     return defaultClickhouseClient;
   }
 
-  const url = new URL(env.REALTIME_RUNS_CLICKHOUSE_URL);
+  const url = new URL(env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_URL);
   url.searchParams.delete("secure");
 
   return new ClickHouse({
     url: url.toString(),
     name: "realtime-runs-clickhouse",
     keepAlive: {
-      enabled: env.REALTIME_RUNS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
-      idleSocketTtl: env.REALTIME_RUNS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+      enabled: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
     },
-    logLevel: env.REALTIME_RUNS_CLICKHOUSE_LOG_LEVEL,
+    logLevel: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_LOG_LEVEL,
     compression: {
-      request: env.REALTIME_RUNS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
+      request: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_COMPRESSION_REQUEST === "1",
     },
-    maxOpenConnections: env.REALTIME_RUNS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
+    maxOpenConnections: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
   });
 }
 
@@ -366,14 +366,14 @@ function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHou
         url: parsed.toString(),
         name,
         keepAlive: {
-          enabled: env.REALTIME_RUNS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
-          idleSocketTtl: env.REALTIME_RUNS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+          enabled: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+          idleSocketTtl: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
         },
-        logLevel: env.REALTIME_RUNS_CLICKHOUSE_LOG_LEVEL,
+        logLevel: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_LOG_LEVEL,
         compression: {
-          request: env.REALTIME_RUNS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
+          request: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_COMPRESSION_REQUEST === "1",
         },
-        maxOpenConnections: env.REALTIME_RUNS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
+        maxOpenConnections: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
       });
     case "standard":
     case "query":

apps/webapp/app/services/realtime/boundedTtlCache.ts

@@ -1,15 +1,6 @@
 /**
- * Tiny in-process bounded TTL cache shared by the realtime feeds.
- *
- * Entries expire after `ttlMs`. An expired entry is evicted when read (`get`); on
- * write, if the cache is at `maxEntries`, expired entries are swept and, if it's
- * still full (pathologically all live), the oldest insertion is dropped. Node is
- * single-threaded so no locking is needed. Used where a miss is cheap and
- * correctness-safe (read-through hydration, per-handle working sets, per-org flag
- * resolution).
- *
- * A stored value of `undefined` cannot be distinguished from a miss; callers that
- * need to cache "absence" should store an explicit sentinel (e.g. `null`).
+ * Tiny in-process bounded TTL cache shared by the realtime feeds: entries expire after `ttlMs` (evicted on read),
+ * and at-capacity writes sweep expired entries then drop the oldest. A stored `undefined` is indistinguishable from a miss (use `null` for absence).
  */
 export class BoundedTtlCache<V> {
   readonly #entries = new Map<string, { value: V; expiresAt: number }>();

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

@@ -10,15 +10,9 @@ export type ClickHouseRunListResolverOptions = {
 };
 
 /**
- * Resolves the realtime tag/list filter into matching run ids via ClickHouse
- * `listRunIds`. Tag matching is contains-ANY (OR), the same
- * semantics the dashboard runs list uses. Filter-only: ids only, hydrated from
- * Postgres by id afterward. This keeps the realtime tag feed off the Postgres
- * `runTags` GIN index entirely.
- *
- * (Multi-tag subscribeToRunsWithTag is therefore OR, not the AND that Electric's
- * `runTags @> ARRAY[...]` shape used. Restoring AND is a follow-up: add a
- * `hasAll` mode to the ClickHouse runs filter and use it here.)
+ * Resolves the realtime tag/list filter into matching run ids via ClickHouse `listRunIds` (filter-only;
+ * rows hydrated from Postgres by id afterward). Tag matching is contains-ANY (OR) — note this differs from
+ * Electric's `runTags @> ARRAY[...]` AND shape; restoring AND needs a `hasAll` mode on the ClickHouse filter.
  */
 export class ClickHouseRunListResolver implements RunListResolver {
   constructor(private readonly options: ClickHouseRunListResolverOptions) {}