feat(stats)!: publish groups as tracks of one node broadcast

demo/relay/localhost.toml

@@ -30,7 +30,7 @@ public = "**"
 #
 #   .stats/node/local/host
 #
-# carrying totals.json, publisher.json, subscriber.json, and sessions.json tracks. See
+# carrying totals.json and group.json tracks. See
 # doc/bin/relay/config.md for the field semantics.
 [stats]
 enabled = true

demo/web/src/stats.ts

@@ -8,23 +8,24 @@
  * let you drill into one. A restarted relay announces a new epoch at the same
  * path, which arrives as an end then a start, so its history starts over.
  *
- * Per-node tracks we read:
- *   publisher.json   egress  (relay -> downstream viewers)
- *   subscriber.json  ingress (upstream publishers -> relay)
- *   sessions.json    sessions by stats label
+ * Per node we read `group.json`, the whole node's group at the relay's default
+ * `depth = 0`: `{ epoch, tiers: { <tier>: { paths, labels } } }`, a JSON snapshot
+ * with merge-patch deltas. From the default tier we take each path's `publisher`
+ * (egress, relay -> downstream viewers) and `subscriber` (ingress, upstream
+ * publishers -> relay), and each stats label's sessions.
  *
- * Each frame is `{ "<broadcast path>": Snapshot }`. Counters are cumulative;
- * "active" = started - ended. The relay includes every entry it still holds
- * counters for, idle ones too, so the latest frame is a snapshot of now. We sample the aggregate on an interval
- * to derive per-second throughput rates for the charts. A relay built after the
- * started/ended rename still writes the legacy names beside the new ones; this
- * dashboard prefers the canonical spelling and falls back so it also reads an
- * older relay.
+ * Counters are cumulative; "active" = started - ended. The relay includes every
+ * entry it still holds counters for, idle ones too, so the latest frame is a
+ * snapshot of now. We sample the aggregate on an interval to derive per-second
+ * throughput rates for the charts. A relay built after the started/ended rename
+ * still writes the legacy names beside the new ones; this dashboard prefers the
+ * canonical spelling and falls back so it also reads an older relay.
  */
 
 import "./highlight";
 import { Milli, Nano, wall } from "@moq/clock";
 import { Net, Signals } from "@moq/hang";
+import * as Json from "@moq/json";
 
 const RELAY_URL = import.meta.env.VITE_RELAY_URL ?? "http://localhost:4443";
 
@@ -78,9 +79,32 @@ interface SessionCounters {
 type SessionFrame = Record<string, SessionCounters>;
 
 interface NodeStats {
-	egress: BroadcastFrame; // publisher.json
-	ingress: BroadcastFrame; // subscriber.json
-	sessions: SessionFrame; // sessions.json
+	egress: BroadcastFrame; // each path's `publisher`
+	ingress: BroadcastFrame; // each path's `subscriber`
+	sessions: SessionFrame; // `labels`
+}
+
+// One tier of a `group.json` frame.
+interface GroupTier {
+	paths?: Record<string, { publisher?: Snapshot; subscriber?: Snapshot }>;
+	labels?: SessionFrame;
+}
+
+interface GroupFrame {
+	tiers?: Record<string, GroupTier>;
+}
+
+// The default tier's egress, ingress, and sessions from a group frame.
+function nodeStatsOf(frame: GroupFrame): NodeStats {
+	const tier = frame.tiers?.[""] ?? {};
+	const egress: BroadcastFrame = {};
+	const ingress: BroadcastFrame = {};
+	for (const [path, usage] of Object.entries(tier.paths ?? {})) {
+		// The relay leaves out a side that never moved.
+		if (usage.publisher) egress[path] = usage.publisher;
+		if (usage.subscriber) ingress[path] = usage.subscriber;
+	}
+	return { egress, ingress, sessions: tier.labels ?? {} };
 }
 
 // Canonical spelling wins when a frame carries both names.
@@ -176,24 +200,18 @@ function subscribeNode(effect: Signals.Effect, origin: Net.Origin.Table, path: N
 	const consumer = request.active.peek();
 	if (!consumer) return;
 
-	const sub = <K extends keyof NodeStats>(trackName: string, key: K) => {
-		const track = consumer.track(trackName).subscribe().ordered();
-		effect.cleanup(() => track.close());
-		effect.spawn(async () => {
-			for (;;) {
-				const data = await effect.race(track.readJson());
-				if (data === undefined) break;
-				nodeStats.mutate((s) => {
-					const cur = s[node];
-					if (cur) cur[key] = (data ?? {}) as NodeStats[K];
-				});
-			}
-		});
-	};
-
-	sub("publisher.json", "egress");
-	sub("subscriber.json", "ingress");
-	sub("sessions.json", "sessions");
+	const track = consumer.track("group.json").subscribe();
+	effect.cleanup(() => track.close());
+	const group = new Json.Snapshot.Consumer<GroupFrame>({ track });
+	effect.spawn(async () => {
+		for (;;) {
+			const state = await effect.race(group.latest());
+			if (state === undefined) break;
+			nodeStats.mutate((s) => {
+				if (s[node]) s[node] = nodeStatsOf(state.value ?? {});
+			});
+		}
+	});
 }
 
 // ---- Aggregation ----------------------------------------------------------

doc/bin/inspect.md

@@ -69,15 +69,16 @@ Dial the `.stats` path, which the token must cover:
 
 ```bash
 moq --connect http://localhost:4443/.stats announced
-moq --connect http://localhost:4443/.stats --broadcast node/local/host fetch publisher.json | jq
-curl http://localhost:4443/fetch/.stats/node/local/host/publisher.json | jq
+moq --connect http://localhost:4443/.stats --broadcast node/local/host fetch totals.json | jq
+curl http://localhost:4443/fetch/.stats/node/local/host/totals.json | jq
 ```
 
 The node name (`local/host` here) is `stats.node`. Each fetch returns the
 newest snapshot of cumulative counters. `totals.json` sums each tier's traffic
-and sessions, `publisher.json` is egress per broadcast path,
-`subscriber.json` is ingress, and `sessions.json` counts sessions per stats
-label. A subscriber may also ask for `<path>/prefix.json`, one broadcast
-prefix rolled up with each child, produced only while held. Fetch twice and divide by the interval for a
-rate. The `.json.z` twins carry compressed patches, which these tools print as
-raw bytes. [Stats](/concept/stats) describes every field.
+and sessions. `group.json` (`<group>/group.json` at `stats.depth` 1 or more)
+carries a group's traffic per broadcast path and sessions per stats label: its
+newest group is a full snapshot followed by merge-patch deltas, one per line.
+A subscriber may also ask for `<path>/prefix.json`, one broadcast prefix rolled
+up with each child, produced only while held. Fetch twice and divide by the
+interval for a rate. The `.json.z` twins carry compressed patches, which these
+tools print as raw bytes. [Stats](/concept/stats) describes every field.

doc/bin/relay/config.md

@@ -350,20 +350,22 @@ group of every track is always kept, even past `capacity`.
 ```toml
 [stats]
 enabled = true
-prefix = ".stats"                    # Broadcasts appear under <prefix>/node/<node>.
+prefix = ".stats"                    # The broadcast appears at <prefix>/node/<node>.
 interval = "1s"                      # Time between snapshots. Default.
 node = "sjc/1"                       # Disambiguates relays sharing a cluster.
-depth = 1                            # Also bucket by the first N path segments (per tenant).
-linger = "5m"                        # Keep an empty group's broadcast announced this long. Default.
+depth = 1                            # One group track per first N path segments (per tenant).
+linger = "5m"                        # Keep an empty group's track listed this long. Default.
 ```
 
 Each node publishes its traffic and session counters as MoQ tracks, split by a
 **tier** label from the auth server's grant (`--cluster-tier` for links this
 relay dials and LAN peers it admits), which is what makes billing per customer or per region possible.
-Each run, and each group returning after its linger, announces under a fresh
-[epoch](/concept/moq-lite#publisher-epochs), so a restart is a new broadcast at
-the same path. Each group's totals count everything since its epoch began, so
-billing reads those; per-path traffic and per-label sessions and HTTP requests are for
+Each run announces under a fresh [epoch](/concept/moq-lite#publisher-epochs),
+so a restart is a new broadcast at the same path. Each group is a
+[track](/concept/stats#group-tracks) of that broadcast, and one returning after
+its linger restarts its counters under a new epoch without an announce. The
+node's totals count everything since its epoch began, so billing reads those;
+a group's per-path traffic and per-label sessions and HTTP requests are for
 dashboards. Session counters report how close sessions come to the
 per-session limits, past which one is closed with `TOO_MANY_REQUESTS`.
 [Stats](/concept/stats) describes the paths, tracks, and encodings; read them