feat(workspace-client,desktop): relay WS outage telemetry + version-stamped desktop events

apps/desktop/src/renderer/lib/posthog.ts

@@ -1,3 +1,4 @@
+import { setRelaySocketTelemetry } from "@superset/workspace-client";
 import posthogFull from "posthog-js/dist/module.full.no-external";
 import type { PostHog } from "posthog-js/react";
 import { env } from "../env.renderer";
@@ -23,8 +24,26 @@ export function initPostHog() {
 		loaded: (ph) => {
 			ph.register({
 				app_name: "desktop",
+				// Event-level version (person-profile desktop_version reflects the
+				// current install, not the build that emitted a given event).
+				app_version: window.App.appVersion,
 				platform: window.navigator.platform,
 			});
 		},
 	});
+
+	// Relay socket health (event bus / workspace "disconnected" surface). At
+	// most one event per outage episode plus one on recovery.
+	setRelaySocketTelemetry((event) => {
+		posthogFull.capture(`relay_ws_${event.kind}`, {
+			socket_name: event.socketName,
+			endpoint: event.endpoint,
+			preflight_status: event.preflightStatus,
+			tunnel_region: event.tunnelRegion,
+			close_code: event.closeCode,
+			close_reason: event.closeReason,
+			reconnect_attempts: event.failedAttempts,
+			outage_ms: event.outageMs,
+		});
+	});
 }

apps/web/src/app/workspaces/[workspaceId]/components/WebTerminal/TerminalConnection.ts

@@ -80,6 +80,7 @@ export class TerminalConnection {
 		}
 
 		const socket = createRelaySocket({
+			name: "web-terminal",
 			buildUrl: () => this.buildUrl(),
 			getToken: () => this.deps.getToken(),
 			// Definitive access denial: retrying can't change the answer.

packages/workspace-client/src/index.ts

@@ -19,6 +19,8 @@ export {
 	createRelaySocket,
 	type RelaySocket,
 	type RelaySocketOptions,
+	type RelaySocketTelemetryEvent,
+	setRelaySocketTelemetry,
 } from "./lib/relaySocket";
 export {
 	useWorkspaceClient,

packages/workspace-client/src/lib/eventBus.ts

@@ -210,6 +210,7 @@ function getOrCreateConnection(
 	// inside partysocket. Buffering is disabled so command semantics stay
 	// "send only while open" — watches are replayed from state on each open.
 	const socket = createRelaySocket({
+		name: "event-bus",
 		buildUrl: () => `${hostUrl.replace(/\/$/, "")}/events`,
 		getToken: getWsToken,
 		accessDeniedRetryMs: ACCESS_DENIED_RETRY_MS,

packages/workspace-client/src/lib/relaySocket/index.ts

@@ -1,3 +1,7 @@
+export {
+	type RelaySocketTelemetryEvent,
+	setRelaySocketTelemetry,
+} from "./outageReporter";
 export {
 	createRelaySocket,
 	type RelaySocket,

packages/workspace-client/src/lib/relaySocket/outageReporter.test.ts

@@ -0,0 +1,118 @@
+import { afterEach, describe, expect, it } from "bun:test";
+import {
+	createOutageReporter,
+	type RelaySocketTelemetryEvent,
+	setRelaySocketTelemetry,
+} from "./outageReporter";
+
+let telemetry: RelaySocketTelemetryEvent[] = [];
+
+function collect(): void {
+	telemetry = [];
+	setRelaySocketTelemetry((e) => telemetry.push(e));
+}
+
+afterEach(() => {
+	setRelaySocketTelemetry(null);
+	telemetry = [];
+});
+
+describe("createOutageReporter", () => {
+	it("reports degraded once at the attempt threshold, then recovered", () => {
+		collect();
+		const reporter = createOutageReporter("bus");
+		reporter.attempt("ws://relay/hosts/h/events?token=secret", {
+			status: 200,
+			region: "iad",
+		});
+		for (let i = 1; i <= 8; i++) {
+			reporter.failed(i, { code: 1006, reason: "" });
+		}
+		expect(telemetry.map((e) => e.kind)).toEqual(["degraded"]);
+		expect(telemetry[0]).toMatchObject({
+			socketName: "bus",
+			endpoint: "ws://relay/hosts/h/events",
+			preflightStatus: 200,
+			tunnelRegion: "iad",
+			closeCode: 1006,
+			failedAttempts: 5,
+		});
+
+		reporter.opened(9);
+		expect(telemetry.map((e) => e.kind)).toEqual(["degraded", "recovered"]);
+		expect(telemetry[1]?.outageMs).not.toBeNull();
+	});
+
+	it("attaches the outage's last observed close info when the threshold-crossing call has none", () => {
+		collect();
+		const reporter = createOutageReporter("bus");
+		reporter.attempt("ws://relay/hosts/h/events", null);
+		// Dial failures surface as close (with code) and error (without) pairs;
+		// the error side can be the one that crosses the threshold.
+		reporter.failed(4, { code: 1006, reason: "abnormal" });
+		reporter.failed(5);
+		expect(telemetry).toHaveLength(1);
+		expect(telemetry[0]).toMatchObject({
+			kind: "degraded",
+			closeCode: 1006,
+			closeReason: "abnormal",
+		});
+
+		// The stashed close info doesn't leak into the next outage.
+		reporter.opened(6);
+		for (let i = 1; i <= 6; i++) {
+			reporter.failed(i);
+		}
+		const second = telemetry.filter((e) => e.kind === "degraded")[1];
+		expect(second?.closeCode).toBeNull();
+	});
+
+	it("never lets a throwing sink escape into the caller", () => {
+		setRelaySocketTelemetry(() => {
+			throw new Error("analytics exploded");
+		});
+		const reporter = createOutageReporter("bus");
+		reporter.attempt("ws://relay/hosts/h/events", {
+			status: 403,
+			region: null,
+		});
+		expect(() => reporter.accessDenied(1)).not.toThrow();
+		expect(() => reporter.failed(6)).not.toThrow();
+		expect(() => reporter.opened(7)).not.toThrow();
+	});
+
+	it("stays silent for short blips and clean opens", () => {
+		collect();
+		const reporter = createOutageReporter("bus");
+		reporter.attempt("ws://relay/hosts/h/events", null);
+		reporter.failed(1);
+		reporter.failed(2);
+		reporter.opened(3);
+		expect(telemetry).toEqual([]);
+	});
+
+	it("reports access_denied once per episode and suppresses degraded", () => {
+		collect();
+		const reporter = createOutageReporter("bus");
+		reporter.attempt("ws://relay/hosts/h/events?token=secret", {
+			status: 403,
+			region: null,
+		});
+		for (let i = 1; i <= 10; i++) {
+			reporter.accessDenied(i);
+			reporter.failed(i); // partysocket error event after provider rejection
+		}
+		expect(telemetry.map((e) => e.kind)).toEqual(["access_denied"]);
+		expect(telemetry[0]?.preflightStatus).toBe(403);
+		expect(telemetry[0]?.endpoint).not.toContain("secret");
+
+		// Regrant: a successful open ends the episode, next denial reports again.
+		reporter.opened(11);
+		reporter.accessDenied(1);
+		expect(telemetry.map((e) => e.kind)).toEqual([
+			"access_denied",
+			"recovered",
+			"access_denied",
+		]);
+	});
+});

packages/workspace-client/src/lib/relaySocket/outageReporter.ts

@@ -0,0 +1,117 @@
+import type { RelayAffinityProbe } from "../primeRelayAffinity";
+
+export interface RelaySocketTelemetryEvent {
+	kind: "access_denied" | "degraded" | "recovered";
+	socketName: string;
+	/** Attempt URL without its query string — the token never leaves this module. */
+	endpoint: string | null;
+	/** `_whoowns` preflight status; null when the probe failed (relay unreachable). */
+	preflightStatus: number | null;
+	tunnelRegion: string | null;
+	closeCode: number | null;
+	closeReason: string | null;
+	/** Dial attempts in the current outage at emit time. */
+	failedAttempts: number;
+	/** Outage duration; only on `recovered`. */
+	outageMs: number | null;
+}
+
+type RelaySocketTelemetrySink = (event: RelaySocketTelemetryEvent) => void;
+
+let telemetrySink: RelaySocketTelemetrySink | null = null;
+
+/**
+ * Install a process-wide sink for socket-health events. Analytics wiring lives
+ * in the consuming app so this package stays analytics-vendor-free. At most one
+ * event per outage episode (plus one on recovery), so the sink can forward
+ * without sampling.
+ */
+export function setRelaySocketTelemetry(
+	sink: RelaySocketTelemetrySink | null,
+): void {
+	telemetrySink = sink;
+}
+
+// Below this many consecutive failed dials an outage is assumed transient
+// (host restart, network blip) and not worth an event.
+const DEGRADED_AFTER_ATTEMPTS = 5;
+
+export interface CloseInfo {
+	code: number;
+	reason: string;
+}
+
+/**
+ * Collapses a socket's connection churn into outage episodes: one
+ * `access_denied` or `degraded` event when an episode is worth reporting, one
+ * `recovered` when it ends. Pure policy — callers feed it dial outcomes.
+ */
+export function createOutageReporter(socketName: string) {
+	let probe: RelayAffinityProbe | null = null;
+	let endpoint: string | null = null;
+	let outageStartedAt: number | null = null;
+	// Failed dials surface as error and close events in either order, and only
+	// the close carries a code — keep the outage's last observed one so the
+	// threshold-crossing call doesn't have to be the one that saw it.
+	let lastClose: CloseInfo | null = null;
+	// One telemetry event per outage episode; cleared on the next open.
+	let reported = false;
+
+	const emit = (
+		kind: RelaySocketTelemetryEvent["kind"],
+		failedAttempts: number,
+		close: CloseInfo | null,
+	): void => {
+		try {
+			telemetrySink?.({
+				kind,
+				socketName,
+				endpoint,
+				preflightStatus: probe?.status ?? null,
+				tunnelRegion: probe?.region ?? null,
+				closeCode: close?.code ?? null,
+				closeReason: close?.reason || null,
+				failedAttempts,
+				outageMs:
+					kind === "recovered" && outageStartedAt !== null
+						? Date.now() - outageStartedAt
+						: null,
+			});
+		} catch {
+			// A throwing sink must never break the socket lifecycle.
+		}
+	};
+
+	return {
+		/** A dial is starting: remember its target and preflight result. */
+		attempt(signedUrl: string, attemptProbe: RelayAffinityProbe | null): void {
+			endpoint = signedUrl.split("?")[0] ?? null;
+			probe = attemptProbe;
+		},
+
+		/** The preflight returned a definitive 403. */
+		accessDenied(failedAttempts: number): void {
+			outageStartedAt ??= Date.now();
+			if (reported) return;
+			reported = true;
+			emit("access_denied", failedAttempts, null);
+		},
+
+		/** A dial failed or an established connection dropped. */
+		failed(failedAttempts: number, close?: CloseInfo): void {
+			outageStartedAt ??= Date.now();
+			if (close) lastClose = close;
+			if (reported || failedAttempts < DEGRADED_AFTER_ATTEMPTS) return;
+			reported = true;
+			emit("degraded", failedAttempts, lastClose);
+		},
+
+		/** The socket (re)connected; closes out a reported episode. */
+		opened(failedAttempts: number): void {
+			if (reported) emit("recovered", failedAttempts, null);
+			outageStartedAt = null;
+			lastClose = null;
+			reported = false;
+		},
+	};
+}

packages/workspace-client/src/lib/relaySocket/relaySocket.test.ts

@@ -1,4 +1,8 @@
 import { afterEach, describe, expect, it } from "bun:test";
+import {
+	type RelaySocketTelemetryEvent,
+	setRelaySocketTelemetry,
+} from "./outageReporter";
 import { createRelaySocket, type RelaySocket } from "./relaySocket";
 
 // Local WS server whose /hosts/:id/_whoowns preflight status is scriptable.
@@ -34,6 +38,7 @@ let socket: RelaySocket | null = null;
 afterEach(() => {
 	socket?.close();
 	socket = null;
+	setRelaySocketTelemetry(null);
 });
 
 function waitFor(cond: () => boolean, timeoutMs = 3_000): Promise<void> {
@@ -107,6 +112,68 @@ describe("createRelaySocket", () => {
 		server.stop(true);
 	});
 
+	it("emits degraded once per outage, then recovered with outage stats", async () => {
+		const { server, port } = makeServer(() => 200);
+		server.stop(true); // outage: every dial is refused
+		const telemetry: RelaySocketTelemetryEvent[] = [];
+		setRelaySocketTelemetry((e) => telemetry.push(e));
+		socket = createRelaySocket({
+			name: "test-bus",
+			buildUrl: () => `ws://localhost:${port}${HOST_PATH}`,
+			getToken: () => "secret-token",
+			minReconnectionDelay: 10,
+			maxReconnectionDelay: 20,
+		});
+		await waitFor(() => telemetry.some((e) => e.kind === "degraded"), 8_000);
+		// Keep failing: still one degraded per outage.
+		await Bun.sleep(150);
+		expect(telemetry.filter((e) => e.kind === "degraded").length).toBe(1);
+		const degraded = telemetry.find((e) => e.kind === "degraded");
+		expect(degraded?.socketName).toBe("test-bus");
+		expect(degraded?.endpoint).not.toContain("secret-token");
+
+		const revived = Bun.serve({
+			port,
+			fetch(req, srv) {
+				const url = new URL(req.url);
+				if (url.pathname.endsWith("/_whoowns")) {
+					return new Response(JSON.stringify({ ok: true }), { status: 200 });
+				}
+				if (srv.upgrade(req)) return;
+				return new Response("no", { status: 400 });
+			},
+			websocket: { open() {}, message() {} },
+		});
+		await waitFor(() => telemetry.some((e) => e.kind === "recovered"), 8_000);
+		const recovered = telemetry.find((e) => e.kind === "recovered");
+		expect(recovered?.failedAttempts).toBeGreaterThanOrEqual(5);
+		expect(recovered?.outageMs).not.toBeNull();
+		revived.stop(true);
+	}, 20_000);
+
+	it("emits access_denied once per denial episode and recovered on regrant", async () => {
+		let status = 403;
+		const { server, port } = makeServer(() => status);
+		const telemetry: RelaySocketTelemetryEvent[] = [];
+		setRelaySocketTelemetry((e) => telemetry.push(e));
+		let denied = 0;
+		socket = createRelaySocket({
+			buildUrl: () => `ws://localhost:${port}${HOST_PATH}`,
+			getToken: () => "tok",
+			onAccessDenied: () => denied++,
+			accessDeniedRetryMs: 30,
+			minReconnectionDelay: 10,
+			maxReconnectionDelay: 20,
+		});
+		await waitFor(() => denied >= 3);
+		expect(telemetry.filter((e) => e.kind === "access_denied").length).toBe(1);
+		expect(telemetry.filter((e) => e.kind === "degraded").length).toBe(0);
+		expect(telemetry[0]?.preflightStatus).toBe(403);
+		status = 200;
+		await waitFor(() => telemetry.some((e) => e.kind === "recovered"), 8_000);
+		server.stop(true);
+	}, 20_000);
+
 	it("dials local (non-/hosts) URLs without a preflight, converting http to ws", async () => {
 		const { server, tokensSeen, port } = makeServer(() => 503);
 		socket = createRelaySocket({