From 02f7a5777111636f3d680e44b522fb0dd2157949 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 16:35:08 +0100 Subject: [PATCH 1/3] feat(webapp): add emission fan-out metrics to the native realtime feed --- .../realtime-emission-fanout-metrics.md | 6 +++++ .../realtime/envChangeRouter.server.ts | 23 +++++++++++++++++++ .../nativeRealtimeClientInstance.server.ts | 20 ++++++++++++++++ 3 files changed, 49 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 00000000000..bd9ef83afbc --- /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 95e25b7d0c0..849a8f437d8 100644 --- a/apps/webapp/app/services/realtime/envChangeRouter.server.ts +++ b/apps/webapp/app/services/realtime/envChangeRouter.server.ts @@ -51,6 +51,15 @@ 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. `runEmissions` = total (feed,run) rows resolved + * to feeds this batch (the baseline per-feed serialize count); `distinctSerializations` = distinct + * (columnSig,runId) among them (what a serialize-once-per-batch memo would compute). The gap + * (runEmissions - distinctSerializations) is the duplicate serialization such a memo would remove. */ + onEmissionFanout?: (stats: { + distinctSerializations: number; + runEmissions: 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 +618,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 runEmissions = 0; + const emittedSerializationKeys = 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 +642,22 @@ export class EnvChangeRouter { if (rows.length > 0) { feed.resolve({ reason: "notify", rows }); + runEmissions += rows.length; + for (const matched of rows) { + emittedSerializationKeys.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 (runEmissions > 0) { + this.options.onEmissionFanout?.({ + distinctSerializations: emittedSerializationKeys.size, + runEmissions, + 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 3f29f3faa47..3019d04ca6f 100644 --- a/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts +++ b/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts @@ -71,6 +71,22 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient { unit: "rows", }); + const emissionRunSerializations = meter.createCounter( + "realtime_native.emission_run_serializations", + { + description: + "Total per-feed row emissions per batch (the wire-value serializations the current path performs). Divide by realtime_native.emission_distinct_serializations for average feeds-per-run fan-out; the excess is duplicate serialization a serialize-once-per-batch memo would remove.", + } + ); + + const emissionDistinctSerializations = meter.createCounter( + "realtime_native.emission_distinct_serializations", + { + description: + "Distinct (columnSig,runId) emitted per batch, summed. A serialize-once-per-batch memo would perform exactly this many serializations vs realtime_native.emission_run_serializations today; 1 - (this / run_serializations) is the memo's serialization-work saving.", + } + ); + 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 +182,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: ({ distinctSerializations, runEmissions }) => { + emissionRunSerializations.add(runEmissions); + emissionDistinctSerializations.add(distinctSerializations); + }, replicaLag: lagEstimator ? { getLagMs: () => lagEstimator.getLagMs(), From 18fa151161b8ea9ace85fdd9e031fd22b810069b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 17:15:47 +0100 Subject: [PATCH 2/3] refactor(webapp): rename realtime emission fan-out metrics to what they count --- .../realtime/envChangeRouter.server.ts | 28 ++++++++++--------- .../nativeRealtimeClientInstance.server.ts | 28 ++++++++----------- 2 files changed, 26 insertions(+), 30 deletions(-) diff --git a/apps/webapp/app/services/realtime/envChangeRouter.server.ts b/apps/webapp/app/services/realtime/envChangeRouter.server.ts index 849a8f437d8..b3206887744 100644 --- a/apps/webapp/app/services/realtime/envChangeRouter.server.ts +++ b/apps/webapp/app/services/realtime/envChangeRouter.server.ts @@ -51,13 +51,15 @@ 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. `runEmissions` = total (feed,run) rows resolved - * to feeds this batch (the baseline per-feed serialize count); `distinctSerializations` = distinct - * (columnSig,runId) among them (what a serialize-once-per-batch memo would compute). The gap - * (runEmissions - distinctSerializations) is the duplicate serialization such a memo would remove. */ + /** 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: { - distinctSerializations: number; - runEmissions: number; + distinctRuns: number; + deliveries: number; feeds: number; }) => void; /** Read-your-writes gate over the replica: delays wake-path hydrates until the replica @@ -618,8 +620,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 runEmissions = 0; - const emittedSerializationKeys = new Set(); + 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 @@ -642,19 +644,19 @@ export class EnvChangeRouter { if (rows.length > 0) { feed.resolve({ reason: "notify", rows }); - runEmissions += rows.length; + deliveries += rows.length; for (const matched of rows) { - emittedSerializationKeys.add(`${feed.columnSig}${matched.row.id}`); + 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 (runEmissions > 0) { + if (deliveries > 0) { this.options.onEmissionFanout?.({ - distinctSerializations: emittedSerializationKeys.size, - runEmissions, + distinctRuns: distinctRunKeys.size, + deliveries, feeds: matchedRunIdsByFeed.size, }); } diff --git a/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts b/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts index 3019d04ca6f..77a9a02e5fc 100644 --- a/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts +++ b/apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts @@ -71,21 +71,15 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient { unit: "rows", }); - const emissionRunSerializations = meter.createCounter( - "realtime_native.emission_run_serializations", - { - description: - "Total per-feed row emissions per batch (the wire-value serializations the current path performs). Divide by realtime_native.emission_distinct_serializations for average feeds-per-run fan-out; the excess is duplicate serialization a serialize-once-per-batch memo would remove.", - } - ); + 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 emissionDistinctSerializations = meter.createCounter( - "realtime_native.emission_distinct_serializations", - { - description: - "Distinct (columnSig,runId) emitted per batch, summed. A serialize-once-per-batch memo would perform exactly this many serializations vs realtime_native.emission_run_serializations today; 1 - (this / run_serializations) is the memo's serialization-work saving.", - } - ); + 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: @@ -182,9 +176,9 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient { unsubscribeLingerMs: env.REALTIME_BACKEND_NATIVE_UNSUBSCRIBE_LINGER_MS, onReplay: (result) => replays.add(1, { result }), onReplayEviction: (reason) => replayEvictions.add(1, { reason }), - onEmissionFanout: ({ distinctSerializations, runEmissions }) => { - emissionRunSerializations.add(runEmissions); - emissionDistinctSerializations.add(distinctSerializations); + onEmissionFanout: ({ distinctRuns, deliveries }) => { + emissionFeedDeliveries.add(deliveries); + emissionDistinctRuns.add(distinctRuns); }, replicaLag: lagEstimator ? { From 2b5e75388135aaabb793a1d40f456180ad268bcf Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 17:20:48 +0100 Subject: [PATCH 3/3] chore(webapp): apply oxfmt formatting to realtime fan-out metrics --- apps/webapp/app/services/realtime/envChangeRouter.server.ts | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/apps/webapp/app/services/realtime/envChangeRouter.server.ts b/apps/webapp/app/services/realtime/envChangeRouter.server.ts index b3206887744..e183bf636a3 100644 --- a/apps/webapp/app/services/realtime/envChangeRouter.server.ts +++ b/apps/webapp/app/services/realtime/envChangeRouter.server.ts @@ -57,11 +57,7 @@ export type EnvChangeRouterOptions = { * (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; + 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). */