feat(webapp): add emission fan-out metrics to the native realtime feed (#4341)

## Summary

Adds two counters to the native realtime backend so we can see how much
duplicate row serialization the change router does per batch. When a run
changes it can match several held feeds at once (a run subscription plus
one or more tag/list feeds), and today each matching feed serializes
that run's wire value independently. These counters quantify that
fan-out so we can decide whether a shared serialization step is worth
it.

## What they measure

- `realtime_native.emission_run_serializations`: total wire-value
serializations performed across feeds per batch (what the current path
does).
- `realtime_native.emission_distinct_serializations`: distinct (columns,
run) rows those serializations cover (what a serialize-once-per-batch
step would do).

Average feeds-per-run is `run_serializations / distinct_serializations`,
and `1 - distinct / run_serializations` is the serialization work a
shared step could save. Wired through a new optional `onEmissionFanout`
callback on the router. No behavior change.

Author: ericallam

.server-changes/realtime-emission-fanout-metrics.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Add metrics to the realtime backend that measure how often a single changed run is served to multiple subscriptions in one batch.

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

@@ -51,6 +51,13 @@ export type EnvChangeRouterOptions = {
   /** 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;
+  /** Observability: per-batch emission fan-out. `deliveries` = total (feed,run) rows matched and
+   * resolved to feeds this batch. It is an upper bound on the per-feed wire serializations, since
+   * the client working-set diff drops already-seen rows before encoding. `distinctRuns` = distinct
+   * (columnSig,runId) among them (the serialize-once-per-batch floor). `deliveries / distinctRuns`
+   * is the average number of feeds a changed run is delivered to; a shared-serialization step would
+   * save at most `deliveries - distinctRuns` encodings. */
+  onEmissionFanout?: (stats: { distinctRuns: number; deliveries: number; feeds: number }) => void;
   /** Read-your-writes gate over the replica: delays wake-path hydrates until the replica
    * should have applied the change (record.updatedAtMs + lag + margin), and re-hydrates
    * rows the tripwire still finds stale. Omit to hydrate immediately (legacy behavior). */
@@ -609,6 +616,8 @@ export class EnvChangeRouter {
 
     // 4. Assemble each feed's matched rows (post-filtering tag feeds against the
     //    authoritative hydrated row) and resolve its pending wait.
+    let deliveries = 0;
+    const distinctRunKeys = new Set<string>();
     for (const [feed, runIds] of matchedRunIdsByFeed) {
       if (!feed.resolve) {
         continue; // stopped waiting while we hydrated; its next poll/backstop covers it
@@ -631,10 +640,22 @@ export class EnvChangeRouter {
 
       if (rows.length > 0) {
         feed.resolve({ reason: "notify", rows });
+        deliveries += rows.length;
+        for (const matched of rows) {
+          distinctRunKeys.add(`${feed.columnSig}${matched.row.id}`);
+        }
       }
       // No surviving rows (e.g. a partial-record candidate that didn't actually match):
       // leave the feed waiting; nothing relevant changed for it.
     }
+
+    if (deliveries > 0) {
+      this.options.onEmissionFanout?.({
+        distinctRuns: distinctRunKeys.size,
+        deliveries,
+        feeds: matchedRunIdsByFeed.size,
+      });
+    }
   }
 
   /** Runs whose hydrated row is provably behind its record's watermark (stale content),

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

@@ -71,6 +71,16 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     unit: "rows",
   });
 
+  const emissionFeedDeliveries = meter.createCounter("realtime_native.emission_feed_deliveries", {
+    description:
+      "Matched (feed,run) rows resolved to feeds per batch, summed. Upper bound on per-feed wire serializations, since the client working-set diff drops already-seen rows before encoding. Divide by realtime_native.emission_distinct_runs for average feeds-per-run fan-out.",
+  });
+
+  const emissionDistinctRuns = meter.createCounter("realtime_native.emission_distinct_runs", {
+    description:
+      "Distinct (columnSig,run) among the deliveries per batch, summed. The serialize-once-per-batch floor: a shared-serialization step would encode at most this many rows, saving at most (feed_deliveries minus distinct_runs) encodings.",
+  });
+
   const backstops = meter.createCounter("realtime_native.backstops", {
     description:
       "Backstop full resolves by outcome. 'empty' is normal idle behavior; sustained 'delivered' means the notify/replay path missed changes — alert on it.",
@@ -166,6 +176,10 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     unsubscribeLingerMs: env.REALTIME_BACKEND_NATIVE_UNSUBSCRIBE_LINGER_MS,
     onReplay: (result) => replays.add(1, { result }),
     onReplayEviction: (reason) => replayEvictions.add(1, { reason }),
+    onEmissionFanout: ({ distinctRuns, deliveries }) => {
+      emissionFeedDeliveries.add(deliveries);
+      emissionDistinctRuns.add(distinctRuns);
+    },
     replicaLag: lagEstimator
       ? {
           getLagMs: () => lagEstimator.getLagMs(),