fix(webapp): harden the realtime runs backend

Addresses review feedback on the new backend:

- skip cache eviction when updating an existing key at capacity
- treat a concurrency limit of 0 as valid (enforce it, not a 500)
- gate subscribeToRunChanges behind the enable switch
- keep protocol-reserved columns in the hydration projection
- re-clamp a handle-recovered createdAt to the max-age floor
- bulk-hydrate the shadow comparator instead of per-run reads
- log only run id and column on divergence, never raw cell values

Author: ericallam

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

@@ -34,7 +34,9 @@ export class BoundedTtlCache<V> {
   }
 
   set(key: string, value: V): void {
-    if (this.#entries.size >= this.maxEntries) {
+    // Only run capacity eviction when inserting a NEW key — updating an existing key
+    // doesn't grow the map, so it must never drop an unrelated live entry.
+    if (!this.#entries.has(key) && this.#entries.size >= this.maxEntries) {
       const now = Date.now();
       for (const [key, entry] of this.#entries) {
         if (entry.expiresAt <= now) {

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

@@ -238,10 +238,15 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
     }
 
     // Recover the pinned window from the handle so the lower bound never drifts.
+    // Re-clamp the recovered value to the max-age floor so a stale or crafted handle
+    // can't widen the lookback past the configured ceiling.
+    const recoveredMs = this.#filterMsFromHandle(handle);
     const filter: RunSetFilter = {
       tags,
       createdAtAfter: new Date(
-        this.#filterMsFromHandle(handle) ?? this.#computeCreatedAtFilter(params.createdAt).getTime()
+        recoveredMs !== undefined
+          ? this.#clampCreatedAtFloor(recoveredMs)
+          : this.#computeCreatedAtFilter(params.createdAt).getTime()
       ),
     };
 
@@ -573,6 +578,13 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
     return bucket > 0 ? Math.floor(ms / bucket) * bucket : ms;
   }
 
+  /** Clamp a handle-recovered createdAt lower bound up to the max-age floor (so a
+   * stale or crafted handle can't widen the window past the ceiling), then re-bucket. */
+  #clampCreatedAtFloor(ms: number): number {
+    const floorMs = Date.now() - this.options.maximumCreatedAtFilterAgeMs;
+    return this.#bucketCreatedAtMs(Math.max(ms, floorMs));
+  }
+
   #mintListHandle(createdAtFilterMs: number): string {
     // Pins the createdAt threshold in the opaque handle so live polls reuse the
     // same lower bound even on a working-set cache miss.
@@ -615,7 +627,7 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
       DEFAULT_CONCURRENCY_LIMIT
     );
 
-    if (!concurrencyLimit) {
+    if (concurrencyLimit == null) {
       logger.error("[notifierRealtimeClient] Failed to get concurrency limit", {
         organizationId: environment.organizationId,
       });

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

@@ -69,5 +69,8 @@ export function publishManyRunChanged(inputs: RunChangeInput[]): void {
 
 /** Subscribe to the next change for a run via the shared subscriber. */
 export function subscribeToRunChanges(runId: string): RunChangeSubscription {
+  if (!notifierEnabled) {
+    throw new Error("Run change notifier is disabled");
+  }
   return getRunChangeNotifier().subscribeToRunChanges(runId);
 }

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

@@ -1,6 +1,6 @@
 import { type Prisma, type PrismaClient } from "@trigger.dev/database";
 import { BoundedTtlCache } from "./boundedTtlCache";
-import { type RealtimeRunRow } from "./electricStreamProtocol.server";
+import { RESERVED_COLUMNS, type RealtimeRunRow } from "./electricStreamProtocol.server";
 
 /**
  * RunReader — the pluggable read half of the notifier-backed realtime feed.
@@ -52,7 +52,7 @@ export const RUN_HYDRATOR_SELECT = {
  * `buildHydratorSelect`), so the replica doesn't ship large `payload`/`output`/
  * `metadata`/`error` columns the response will drop anyway.
  */
-const ALWAYS_HYDRATED_COLUMNS = new Set<string>(["id", "updatedAt"]);
+const ALWAYS_HYDRATED_COLUMNS = new Set<string>(["id", "updatedAt", ...RESERVED_COLUMNS]);
 
 /** Project `RUN_HYDRATOR_SELECT` down to the columns the client didn't skip (plus
  * the always-needed ones). An empty skip set returns the full select unchanged. */

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

@@ -95,13 +95,21 @@ export class RealtimeShadowComparator {
       diffs: [],
     };
 
+    // Bulk-hydrate every emitted run in one query rather than a per-message round
+    // trip, so shadow mode doesn't inflate the very replica load it's measuring.
+    const emittedIds = changes
+      .map((m) => m.value.id)
+      .filter((id): id is string => typeof id === "string");
+    const hydrated = await this.options.runReader.hydrateByIds(input.environment.id, emittedIds);
+    const rowsById = new Map(hydrated.map((row) => [row.id, row]));
+
     for (const message of changes) {
       const runId = message.value.id ?? undefined;
       if (!runId) {
         continue;
       }
 
-      const row = await this.options.runReader.getRunById(input.environment.id, runId);
+      const row = rowsById.get(runId);
       if (!row) {
         // Run no longer readable (deleted / replica miss). Not a serialization divergence.
         outcome.serializationSkew++;

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

@@ -180,7 +180,9 @@ export class ShadowRealtimeClient implements RealtimeStreamClient {
         membershipMatch: outcome.membershipMatch,
         missingInNotifier: outcome.missingInNotifier?.slice(0, 20),
         extraInNotifier: outcome.extraInNotifier?.slice(0, 20),
-        diffs: outcome.diffs,
+        // Log only which run/column diverged, never the raw cell values — they can
+        // include run payload/output/metadata and must not leak into logs.
+        diffs: outcome.diffs.map(({ runId, column }) => ({ runId, column })),
       });
     }
   }

apps/webapp/test/realtime/boundedTtlCache.test.ts

@@ -28,6 +28,17 @@ describe("BoundedTtlCache", () => {
     expect(cache.size).toBe(0);
   });
 
+  it("does not evict another entry when updating an existing key at capacity", () => {
+    const cache = new BoundedTtlCache<number>(60_000, 2);
+    cache.set("a", 1);
+    cache.set("b", 2);
+    // Updating an existing key doesn't grow the map, so it must not drop "b".
+    cache.set("a", 11);
+    expect(cache.get("a")).toBe(11);
+    expect(cache.get("b")).toBe(2);
+    expect(cache.size).toBe(2);
+  });
+
   it("drops the oldest entry when full of still-live entries", () => {
     const cache = new BoundedTtlCache<number>(60_000, 2);
     cache.set("a", 1);

apps/webapp/test/realtime/notifierRunSetCache.test.ts

@@ -171,3 +171,57 @@ describe("NotifierRealtimeClient tag-list createdAt bucketing", () => {
     }
   });
 });
+
+describe("NotifierRealtimeClient review fixes", () => {
+  const ready = { changed: Promise.resolve(), unsubscribe() {} };
+  const liveNotifier = { subscribeToRunChanges: () => ready, subscribeToEnvChanges: () => ready };
+
+  it("clamps a stale/crafted handle's createdAt up to the max-age floor", async () => {
+    const maxAge = 24 * 60 * 60 * 1000;
+    const { client, resolveSpy } = makeClient({
+      notifier: liveNotifier,
+      maximumCreatedAtFilterAgeMs: maxAge,
+      runSetCreatedAtBucketMs: 0,
+      livePollTimeoutMs: 50,
+    });
+    const before = Date.now();
+    // Handle encodes createdAt = 1ms epoch, far older than the 24h ceiling.
+    await client.streamRuns(
+      "http://localhost:3030/realtime/v1/runs?offset=123_1&live=true&handle=runs_1_7",
+      ENV,
+      { tags: ["t"] },
+      CURRENT_API_VERSION,
+      undefined,
+      "1.0.0"
+    );
+    const passed = resolveSpy.mock.calls[0][0].createdAtAfter as Date;
+    // Clamped to ~now - maxAge, not the epoch value encoded in the handle.
+    expect(passed.getTime()).toBeGreaterThan(before - maxAge - 1_000);
+  });
+
+  it("enforces a concurrency limit of 0 instead of failing with a 500", async () => {
+    let limitCheckedWith: number | undefined;
+    const { client } = makeClient({
+      notifier: liveNotifier,
+      cachedLimitProvider: { getCachedLimit: async () => 0 },
+      limiter: {
+        incrementAndCheck: async (_env: string, _id: string, limit: number) => {
+          limitCheckedWith = limit;
+          return true;
+        },
+        decrement: async () => {},
+      },
+      livePollTimeoutMs: 50,
+    });
+    const res = await client.streamBatch(
+      "http://localhost:3030/realtime/v1/batches/batch_1?offset=123_1&live=true",
+      ENV,
+      "batch_1",
+      CURRENT_API_VERSION,
+      undefined,
+      "1.0.0"
+    );
+    expect(res.status).toBe(200);
+    expect(limitCheckedWith).toBe(0);
+  });
+});

apps/webapp/test/realtime/runReaderProjection.test.ts

@@ -11,6 +11,24 @@ describe("buildHydratorSelect", () => {
     expect(select.error).toBe(true);
   });
 
+  it("keeps protocol-reserved columns even when asked to skip them", () => {
+    // Reserved columns are always emitted by the serializer, so hydration must keep
+    // them regardless of skipColumns or the output is null/incorrect.
+    const select = buildHydratorSelect([
+      "status",
+      "taskIdentifier",
+      "createdAt",
+      "friendlyId",
+      "payload",
+    ]);
+    expect(select.status).toBe(true);
+    expect(select.taskIdentifier).toBe(true);
+    expect(select.createdAt).toBe(true);
+    expect(select.friendlyId).toBe(true);
+    // A non-reserved skipped column is still dropped.
+    expect(select.payload).toBeUndefined();
+  });
+
   it("drops skipped columns but always keeps id + updatedAt", () => {
     const select = buildHydratorSelect(["payload", "output", "metadata", "error"]);
     expect(select.payload).toBeUndefined();

apps/webapp/test/realtime/shadowCompare.test.ts

@@ -49,7 +49,11 @@ function makeComparator(
   resolvedIds: string[] = []
 ) {
   return new RealtimeShadowComparator({
-    runReader: { getRunById: async (_env: string, id: string) => rowsById[id] ?? null } as any,
+    runReader: {
+      getRunById: async (_env: string, id: string) => rowsById[id] ?? null,
+      hydrateByIds: async (_env: string, ids: string[]) =>
+        ids.map((id) => rowsById[id]).filter((row): row is RealtimeRunRow => Boolean(row)),
+    } as any,
     runListResolver: { resolveMatchingRunIds: async (_f: RunListFilter) => resolvedIds } as any,
   });
 }