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..e183bf636a3 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 3f29f3faa47..77a9a02e5fc 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(),