feat(webapp): add delivery-lag and health metrics to the realtime backend

New metrics covering the questions an operator asks during rollout: an
end-to-end delivery-lag histogram per emission path, a backstop outcome
split (sustained delivered counts mean live wakes are being missed),
publish success/failure and coalesce-ratio counters on the pub/sub
side, held-feed and active-environment gauges as the capacity signal,
concurrency-limiter rejections, replay-buffer evictions split by cause,
and a rows-per-emission histogram that exposes full-set re-emission
regressions.

Author: ericallam

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

@@ -48,6 +48,9 @@ export type EnvChangeRouterOptions = {
   unsubscribeLingerMs?: number;
   /** Observability: a replay scan found candidates and delivered rows (or none survived). */
   onReplay?: (result: "delivered" | "empty") => void;
+  /** Observability: a buffered record was evicted. `cap` evictions mean the env churns more
+   * runs inside the window than the buffer holds (the replay guarantee is degrading). */
+  onReplayEviction?: (reason: "cap" | "window") => void;
 };
 
 const DEFAULT_REPLAY_WINDOW_MS = 2_000;
@@ -210,6 +213,17 @@ export class EnvChangeRouter {
     return this.#envs.size;
   }
 
+  /** Currently-held feeds by kind (for metrics) — the system's capacity unit. */
+  get heldFeedCounts(): { run: number; tag: number; batch: number } {
+    const counts = { run: 0, tag: 0, batch: 0 };
+    for (const env of this.#envs.values()) {
+      for (const feed of env.feeds) {
+        counts[feed.filter.kind]++;
+      }
+    }
+    return counts;
+  }
+
   #ensureEnv(environmentId: string): EnvState {
     const existing = this.#envs.get(environmentId);
     if (existing) {
@@ -282,6 +296,7 @@ export class EnvChangeRouter {
       if (entry.receivedAtMs >= cutoff && env.recent.size <= maxRuns) {
         break;
       }
+      this.options.onReplayEviction?.(entry.receivedAtMs < cutoff ? "window" : "cap");
       env.recent.delete(runId);
     }
   }

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

@@ -116,6 +116,14 @@ export type NativeRealtimeClientOptions = {
   onRunSetQuery?: (stage: "resolve" | "hydrate", ms: number) => void;
   /** Observability hook: a fresh resolve waited `ms` for an admission permit (only when the gate engaged). */
   onResolveAdmissionWait?: (ms: number) => void;
+  /** Observability hook: a live emission left the server — lag is now minus the newest
+   * emitted row's updatedAt (the end-to-end delivery SLI), rowCount the delta size. */
+  onEmit?: (path: LivePollPath, lagMs: number, rowCount: number) => void;
+  /** Observability hook: a backstop resolve found missed changes (delivered) or nothing
+   * (empty). Sustained `delivered` means the notify/replay path is leaking. */
+  onBackstopResult?: (result: "delivered" | "empty") => void;
+  /** Observability hook: a poll was rejected by the per-env concurrency limiter (429). */
+  onConcurrencyRejected?: () => void;
 };
 
 const DEFAULT_CONCURRENCY_LIMIT = 100_000;
@@ -414,6 +422,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
           const probed = await this.options.runReader.getRunById(environment.id, runId);
           if (probed && probed.updatedAt.getTime() > lastSeenMs) {
             const seq = this.#nextSeq();
+            this.options.onEmit?.("cold-resolve", Date.now() - probed.updatedAt.getTime(), 1);
             return this.#buildResponse(
               buildUpdateBody(probed, skipColumns),
               apiVersion,
@@ -450,6 +459,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
             const updatedAtMs = matched.row.updatedAt.getTime();
             if (updatedAtMs > lastSeenMs) {
               const seq = this.#nextSeq();
+              this.options.onEmit?.("fast-hydrate", Date.now() - updatedAtMs, 1);
               return this.#buildResponse(
                 buildRowsBodyFromSerialized([
                   { runId: matched.row.id, value: matched.value, operation: "update" },
@@ -469,6 +479,8 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
           const row = await this.options.runReader.getRunById(environment.id, runId);
           const seq = this.#nextSeq();
           if (row && row.updatedAt.getTime() > lastSeenMs) {
+            this.options.onBackstopResult?.("delivered");
+            this.options.onEmit?.("full-resolve", Date.now() - row.updatedAt.getTime(), 1);
             return this.#buildResponse(
               buildUpdateBody(row, skipColumns),
               apiVersion,
@@ -480,6 +492,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
               }
             );
           }
+          this.options.onBackstopResult?.("empty");
           return this.#buildResponse(buildUpToDateBody(), apiVersion, clientVersion, {
             offset,
             handle,
@@ -601,6 +614,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
             this.#workingSetCache.set(workingSetKey, touched);
             prevSeen = touched;
             if (changes.length > 0) {
+              this.options.onEmit?.("cold-resolve", Date.now() - maxUpdatedAt, changes.length);
               return emitFromRows(changes, maxUpdatedAt);
             }
             continue; // nothing was missed — hold as usual
@@ -633,6 +647,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
             prevSeen = merged;
 
             if (changes.length > 0) {
+              this.options.onEmit?.("fast-hydrate", Date.now() - maxUpdatedAt, changes.length);
               return emitFromSerialized(changes, maxUpdatedAt);
             }
             // Matched but no row advanced (already seen). Keep holding.
@@ -655,10 +670,13 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
           prevSeen = touched;
 
           if (changes.length > 0) {
+            this.options.onBackstopResult?.("delivered");
+            this.options.onEmit?.("full-resolve", Date.now() - maxUpdatedAt, changes.length);
             return emitFromRows(changes, maxUpdatedAt);
           }
           // Empty backstop diff: timeout returns up-to-date; (holdOnEmpty never reaches
           // here on a notify — those are handled in the fast path above).
+          this.options.onBackstopResult?.("empty");
           return emitUpToDate(maxUpdatedAt);
         }
       } finally {
@@ -946,6 +964,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
     );
 
     if (!canProceed) {
+      this.options.onConcurrencyRejected?.();
       return json({ error: "Too many concurrent requests" }, { status: 429 });
     }
 

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

@@ -63,6 +63,41 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     registers: [metricsRegister],
   });
 
+  const deliveryLagMs = new Histogram({
+    name: "realtime_native_delivery_lag_ms",
+    help: "Live emissions: now minus the newest emitted row's updatedAt (PG clock vs app clock, so approximate). The end-to-end delivery SLI — a p99 near the backstop hold means wakes are being missed.",
+    labelNames: ["path"] as const,
+    buckets: [5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000],
+    registers: [metricsRegister],
+  });
+
+  const emittedRows = new Histogram({
+    name: "realtime_native_emitted_rows",
+    help: "Rows per live emission. Deltas should be small; a fat tail means working-set/offset-floor fallbacks are re-emitting full sets.",
+    buckets: [1, 2, 5, 10, 25, 50, 100, 250, 1_000],
+    registers: [metricsRegister],
+  });
+
+  const backstops = new Counter({
+    name: "realtime_native_backstop_total",
+    help: "Backstop full resolves by outcome. 'empty' is normal idle behavior; sustained 'delivered' means the notify/replay path missed changes — alert on it.",
+    labelNames: ["result"] as const,
+    registers: [metricsRegister],
+  });
+
+  const concurrencyRejections = new Counter({
+    name: "realtime_native_concurrency_rejections_total",
+    help: "Polls rejected (429) by the per-env concurrency limiter.",
+    registers: [metricsRegister],
+  });
+
+  const replayEvictions = new Counter({
+    name: "realtime_native_replay_evictions_total",
+    help: "Replay-buffer evictions. 'window' expiry is normal; 'cap' means an env churns more runs inside the window than the buffer holds (replay guarantee degrading — retune the knobs).",
+    labelNames: ["reason"] as const,
+    registers: [metricsRegister],
+  });
+
   const limiter = new RealtimeConcurrencyLimiter({
     keyPrefix: "tr:realtime:native:concurrency",
     redis: {
@@ -90,6 +125,7 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     replayMaxRunsPerEnv: env.REALTIME_BACKEND_NATIVE_REPLAY_MAX_RUNS,
     unsubscribeLingerMs: env.REALTIME_BACKEND_NATIVE_UNSUBSCRIBE_LINGER_MS,
     onReplay: (result) => replays.inc({ result }),
+    onReplayEviction: (reason) => replayEvictions.inc({ reason }),
   });
 
   const client = new NativeRealtimeClient({
@@ -128,6 +164,12 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     onRunSetResolve: (result) => runSetResolves.inc({ result }),
     onRunSetQuery: (stage, ms) => runSetQueryMs.observe({ stage }, ms),
     onResolveAdmissionWait: () => resolveAdmissionWaits.inc(),
+    onEmit: (path, lagMs, rowCount) => {
+      deliveryLagMs.observe({ path }, Math.max(lagMs, 0));
+      emittedRows.observe(rowCount);
+    },
+    onBackstopResult: (result) => backstops.inc({ result }),
+    onConcurrencyRejected: () => concurrencyRejections.inc(),
   });
 
   new Gauge({
@@ -148,6 +190,28 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     },
   });
 
+  new Gauge({
+    name: "realtime_native_held_feeds",
+    help: "Long-polls currently held, by feed kind — the system's capacity unit.",
+    labelNames: ["kind"] as const,
+    registers: [metricsRegister],
+    collect() {
+      const counts = router.heldFeedCounts;
+      this.set({ kind: "run" }, counts.run);
+      this.set({ kind: "tag" }, counts.tag);
+      this.set({ kind: "batch" }, counts.batch);
+    },
+  });
+
+  new Gauge({
+    name: "realtime_native_active_envs",
+    help: "Environments currently routed on this instance (held feeds + lingering subscriptions).",
+    registers: [metricsRegister],
+    collect() {
+      this.set(router.activeEnvCount);
+    },
+  });
+
   return client;
 }
 

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

@@ -60,6 +60,12 @@ export type RunChangeNotifierOptions = {
   envWakeCoalesceWindowMs?: number;
   /** Use Redis sharded pub/sub (SSUBSCRIBE/SPUBLISH); cluster-only and requires `clusterOptions.shardedSubscribers`. Defaults to false (classic). */
   shardedPubSub?: boolean;
+  /** Observability hook: a publish settled (ok) or failed (the leading degradation signal). */
+  onPublishResult?: (ok: boolean) => void;
+  /** Observability hook: a raw channel message arrived (pre-coalesce). */
+  onMessageReceived?: () => void;
+  /** Observability hook: a coalesced batch was delivered to listeners (records per batch). */
+  onBatchDelivered?: (recordCount: number) => void;
 };
 
 const DEFAULT_CHANNEL_PREFIX = "realtime:";
@@ -115,15 +121,22 @@ export class RunChangeNotifier {
       const result = this.#sharded
         ? publisher.spublish(channel, payload)
         : publisher.publish(channel, payload);
-      if (typeof (result as Promise<number>)?.catch === "function") {
-        (result as Promise<number>).catch((error) => {
-          logger.error("[runChangeNotifier] Failed to publish run-changed notification", {
-            error,
-            channel,
-          });
-        });
+      if (typeof (result as Promise<number>)?.then === "function") {
+        (result as Promise<number>).then(
+          () => this.options.onPublishResult?.(true),
+          (error) => {
+            this.options.onPublishResult?.(false);
+            logger.error("[runChangeNotifier] Failed to publish run-changed notification", {
+              error,
+              channel,
+            });
+          }
+        );
+      } else {
+        this.options.onPublishResult?.(true);
       }
     } catch (error) {
+      this.options.onPublishResult?.(false);
       logger.error("[runChangeNotifier] Failed to publish run-changed notification", {
         error,
         channel,
@@ -241,6 +254,7 @@ export class RunChangeNotifier {
   }
 
   #onMessage(channel: string, message: string) {
+    this.options.onMessageReceived?.();
     // Accumulate the decoded record (deduped by runId) before delivering, so a coalesced
     // wake carries every run that moved during the window.
     this.#addPending(channel, decodeChangeRecord(message));
@@ -274,6 +288,7 @@ export class RunChangeNotifier {
     if (!listeners || batch.length === 0) {
       return;
     }
+    this.options.onBatchDelivered?.(batch.length);
     for (const onBatch of [...listeners]) {
       onBatch(batch);
     }

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

@@ -1,4 +1,4 @@
-import { Gauge } from "prom-client";
+import { Counter, Gauge } from "prom-client";
 import { env } from "~/env.server";
 import { metricsRegister } from "~/metrics.server";
 import { singleton } from "~/utils/singleton";
@@ -16,6 +16,25 @@ function initializeRunChangeNotifier(): RunChangeNotifier {
   // broadcast every message to every node, so this is what actually shards load.
   const shardedPubSub = clusterMode && env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_SHARDED_ENABLED === "1";
 
+  const publishes = new Counter({
+    name: "realtime_run_change_notifier_publishes_total",
+    help: "Change-record publishes by outcome. Failures are the leading indicator that feeds are degrading to their backstops (pub/sub Redis trouble).",
+    labelNames: ["result"] as const,
+    registers: [metricsRegister],
+  });
+
+  const received = new Counter({
+    name: "realtime_run_change_notifier_messages_received_total",
+    help: "Raw channel messages received by this instance's subscriber, pre-coalesce.",
+    registers: [metricsRegister],
+  });
+
+  const delivered = new Counter({
+    name: "realtime_run_change_notifier_batches_delivered_total",
+    help: "Coalesced batches delivered to listeners. received/batches = the coalesce ratio (how hard a busy env is being collapsed).",
+    registers: [metricsRegister],
+  });
+
   const notifier = new RunChangeNotifier({
     redis: {
       host: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_HOST,
@@ -29,6 +48,9 @@ function initializeRunChangeNotifier(): RunChangeNotifier {
     },
     envWakeCoalesceWindowMs: env.REALTIME_BACKEND_NATIVE_ENV_WAKE_COALESCE_WINDOW_MS,
     shardedPubSub,
+    onPublishResult: (ok) => publishes.inc({ result: ok ? "ok" : "error" }),
+    onMessageReceived: () => received.inc(),
+    onBatchDelivered: () => delivered.inc(),
   });
 
   new Gauge({

apps/webapp/test/realtime/envChangeRouter.test.ts

@@ -317,7 +317,11 @@ describe("EnvChangeRouter", () => {
       ["r2", row("r2")],
       ["r3", row("r3")],
     ]);
-    const { router, src, hydrateSpy } = makeRouter(rows, { replayMaxRunsPerEnv: 2 });
+    const evictions: string[] = [];
+    const { router, src, hydrateSpy } = makeRouter(rows, {
+      replayMaxRunsPerEnv: 2,
+      onReplayEviction: (reason: string) => evictions.push(reason),
+    });
     const reg = router.register("env_1", { kind: "batch", batchId: "batch_1" }, []);
     src.push("env_1", [
       record("r1", { batchId: "batch_1" }),
@@ -329,6 +333,7 @@ describe("EnvChangeRouter", () => {
     expect(result.reason).toBe("notify");
     // r1 was evicted by the cap; only the newest two replay.
     expect(hydrateSpy).toHaveBeenCalledWith("env_1", ["r2", "r3"], []);
+    expect(evictions).toEqual(["cap"]);
     reg.close();
   });
 });

apps/webapp/test/realtime/nativeHoldOnEmpty.test.ts

@@ -120,7 +120,10 @@ const isUpToDate = (body: Awaited<ReturnType<typeof bodyOf>>) =>
 
 describe("NativeRealtimeClient multi-run live path over the router", () => {
   it("a matching change hydrates by id (no ClickHouse) and returns a delta", async () => {
-    const { client, src, hydrateSpy, resolveSpy, setRows } = makeClient();
+    const emits: Array<[string, number, number]> = [];
+    const { client, src, hydrateSpy, resolveSpy, setRows } = makeClient({
+      onEmit: (path: string, lagMs: number, rows: number) => emits.push([path, lagMs, rows]),
+    });
     setRows([row("run_1", FLOOR_MS + 5_000, { tags: ["t"] })]);
 
     const responsePromise = liveRuns(client);
@@ -132,6 +135,9 @@ describe("NativeRealtimeClient multi-run live path over the router", () => {
     expect(hasRowOp(await bodyOf(res))).toBe(true);
     expect(resolveSpy).not.toHaveBeenCalled(); // ClickHouse skipped
     expect(hydrateSpy).toHaveBeenCalledWith("env_1", ["run_1"], expect.anything());
+    expect(emits).toHaveLength(1);
+    expect(emits[0][0]).toBe("fast-hydrate");
+    expect(emits[0][2]).toBe(1); // one delta row
   });
 
   it("a change that doesn't match the filter never wakes the feed (no CH, no PG); a later match does", async () => {
@@ -175,11 +181,16 @@ describe("NativeRealtimeClient multi-run live path over the router", () => {
   });
 
   it("the backstop timeout does a full ClickHouse resolve and returns up-to-date", async () => {
-    const { client, resolveSpy } = makeClient({ livePollTimeoutMs: 50 });
+    const backstopResults: string[] = [];
+    const { client, resolveSpy } = makeClient({
+      livePollTimeoutMs: 50,
+      onBackstopResult: (r: string) => backstopResults.push(r),
+    });
     const res = await liveRuns(client); // never pushed -> backstop fires
     expect(res.status).toBe(200);
     expect(isUpToDate(await bodyOf(res))).toBe(true);
     expect(resolveSpy).toHaveBeenCalled();
+    expect(backstopResults).toEqual(["empty"]);
   });
 
   it("a cold env registration resolves immediately instead of holding blind", async () => {