From a2d382b2be3964c9b0ea7edc5fced8844258d8f2 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 17:48:44 +0100 Subject: [PATCH] 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. --- .../realtime-emission-fanout-metrics.md | 6 ++++++ .../realtime/envChangeRouter.server.ts | 21 +++++++++++++++++++ .../nativeRealtimeClientInstance.server.ts | 14 +++++++++++++ 3 files changed, 41 insertions(+) create mode 100644 .server-changes/realtime-emission-fanout-metrics.md diff --git a/.server-changes/realtime-emission-fanout-metrics.md b/.server-changes/realtime-emission-fanout-metrics.md new file mode 100644 index 000000000..bd9ef83af --- /dev/null +++ b/.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. diff --git a/apps/webapp/app/services/realtime/envChangeRouter.server.ts b/apps/webapp/app/services/realtime/envChangeRouter.server.ts index 95e25b7d0..e183bf636 100644 --- a/apps/webapp/app/services/realtime/envChangeRouter.server.ts +++ b/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(); 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), diff --git a/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts b/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts index 3f29f3faa..77a9a02e5 100644 --- a/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts +++ b/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(),