fix(ui): isolate gateway observer lifecycles (#117017)

Co-authored-by: Peter Steinberger <steipete@macos.shared>

ui/src/app/gateway-store.observers.test.ts

@@ -0,0 +1,305 @@
+// @vitest-environment node
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type {
+  GatewayBrowserClient,
+  GatewayBrowserClientOptions,
+  GatewayEventFrame,
+  GatewayHelloOk,
+} from "../api/gateway.ts";
+import { createStorageMock } from "../test-helpers/storage.ts";
+import { createApplicationGateway } from "./gateway-store.ts";
+import { loadSettings } from "./settings.ts";
+
+vi.mock("../build-info.ts", () => ({
+  CONTROL_UI_BUILD_INFO: { version: "2026.7.2" },
+}));
+
+const HELLO: GatewayHelloOk = {
+  type: "hello-ok",
+  protocol: 1,
+  auth: { role: "operator", scopes: [] },
+};
+
+function createGatewayEvent(seq: number): GatewayEventFrame {
+  return {
+    type: "event",
+    event: "chat",
+    payload: { text: `event-${seq}` },
+    seq,
+    stateVersion: { presence: seq, health: seq },
+  };
+}
+
+function createGatewayStore() {
+  const clients: Array<{
+    opts: GatewayBrowserClientOptions;
+    start: ReturnType<typeof vi.fn>;
+    stop: ReturnType<typeof vi.fn>;
+  }> = [];
+  const gateway = createApplicationGateway(loadSettings(), "", "", (opts) => {
+    const client = {
+      opts,
+      instanceId: opts.instanceId ?? "",
+      request: vi.fn().mockRejectedValue(new Error("unexpected gateway request")),
+      start: vi.fn(),
+      stop: vi.fn(),
+    };
+    clients.push(client);
+    return client as unknown as GatewayBrowserClient;
+  });
+  return {
+    gateway,
+    clients,
+    current: () => {
+      const client = clients.at(-1);
+      if (!client) {
+        throw new Error("expected a gateway client");
+      }
+      return client;
+    },
+  };
+}
+
+describe("application gateway observer ownership", () => {
+  beforeEach(() => {
+    vi.stubGlobal("localStorage", createStorageMock());
+    vi.stubGlobal("sessionStorage", createStorageMock());
+    vi.stubGlobal("navigator", { language: "en-US" } as Navigator);
+    vi.stubGlobal("location", {
+      protocol: "http:",
+      host: "127.0.0.1:18789",
+      hostname: "127.0.0.1",
+      pathname: "/",
+    } as Location);
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+  });
+
+  it("isolates a failing snapshot observer during the actual hello callback", () => {
+    const { gateway, current } = createGatewayStore();
+    const failure = new Error("snapshot observer failed");
+    const reportError = vi.spyOn(console, "error").mockImplementation(() => {});
+    const healthy = vi.fn();
+    gateway.subscribe((snapshot) => {
+      if (snapshot.phase === "connected") {
+        throw failure;
+      }
+    });
+    gateway.subscribe(healthy);
+    gateway.start();
+
+    expect(() => current().opts.onHello?.(HELLO)).not.toThrow();
+
+    expect(gateway.snapshot.phase).toBe("connected");
+    expect(healthy.mock.calls.map(([snapshot]) => snapshot.phase)).toEqual([
+      "connecting",
+      "connected",
+    ]);
+    expect(reportError).toHaveBeenCalledExactlyOnceWith(
+      "[gateway] snapshot handler error:",
+      failure,
+    );
+  });
+
+  it("snapshots connection observers before subscription membership changes", () => {
+    const { gateway, current } = createGatewayStore();
+    const second = vi.fn();
+    const third = vi.fn();
+    let unsubscribeSecond = () => {};
+    gateway.subscribe((snapshot) => {
+      if (snapshot.phase === "connecting") {
+        unsubscribeSecond();
+        gateway.subscribe(third);
+      }
+    });
+    unsubscribeSecond = gateway.subscribe(second);
+
+    gateway.start();
+
+    expect(second).toHaveBeenCalledOnce();
+    expect(second.mock.calls[0]?.[0].phase).toBe("connecting");
+    expect(third).not.toHaveBeenCalled();
+
+    current().opts.onHello?.(HELLO);
+
+    expect(second).toHaveBeenCalledOnce();
+    expect(third).toHaveBeenCalledOnce();
+    expect(third.mock.calls[0]?.[0].phase).toBe("connected");
+  });
+
+  it("retires a snapshot immediately when an observer replaces its gateway client", () => {
+    const { gateway, current } = createGatewayStore();
+    const healthy = vi.fn();
+    let replaced = false;
+    gateway.subscribe((snapshot) => {
+      if (snapshot.phase === "connected" && !replaced) {
+        replaced = true;
+        gateway.connect();
+      }
+    });
+    gateway.subscribe(healthy);
+    gateway.start();
+
+    current().opts.onHello?.(HELLO);
+
+    expect(gateway.snapshot.phase).toBe("reconnecting");
+    expect(healthy.mock.calls.map(([snapshot]) => snapshot.phase)).toEqual([
+      "connecting",
+      "reconnecting",
+    ]);
+  });
+
+  it("isolates a failing event-log observer from the remaining log observers", () => {
+    const { gateway, current } = createGatewayStore();
+    const failure = new Error("event log observer failed");
+    const reportError = vi.spyOn(console, "error").mockImplementation(() => {});
+    const healthy = vi.fn();
+    gateway.subscribeEventLog(() => {
+      throw failure;
+    });
+    gateway.subscribeEventLog(healthy);
+    gateway.start();
+
+    current().opts.onEvent?.(createGatewayEvent(1));
+
+    expect(healthy).toHaveBeenCalledExactlyOnceWith(gateway.eventLog);
+    expect(reportError).toHaveBeenCalledExactlyOnceWith("[gateway] event handler error:", failure);
+  });
+
+  it("snapshots event-log observers before subscription membership changes", () => {
+    const { gateway, current } = createGatewayStore();
+    const second = vi.fn();
+    const third = vi.fn();
+    let unsubscribeSecond = () => {};
+    gateway.subscribeEventLog(() => {
+      unsubscribeSecond();
+      gateway.subscribeEventLog(third);
+    });
+    unsubscribeSecond = gateway.subscribeEventLog(second);
+    gateway.start();
+
+    current().opts.onEvent?.(createGatewayEvent(1));
+
+    expect(second).toHaveBeenCalledOnce();
+    expect(third).not.toHaveBeenCalled();
+
+    current().opts.onEvent?.(createGatewayEvent(2));
+
+    expect(second).toHaveBeenCalledOnce();
+    expect(third).toHaveBeenCalledExactlyOnceWith(gateway.eventLog);
+  });
+
+  it("never logs a presence event after its snapshot observer replaces the client", () => {
+    const { gateway, current } = createGatewayStore();
+    const logged = vi.fn();
+    const delivered = vi.fn();
+    let replaced = false;
+    gateway.subscribe((snapshot) => {
+      if (snapshot.selfUser?.id === "retired-owner" && !replaced) {
+        replaced = true;
+        gateway.connect();
+      }
+    });
+    gateway.subscribeEventLog(logged);
+    gateway.subscribeEvents(delivered);
+    gateway.start();
+    const retired = current();
+    retired.opts.onHello?.(HELLO);
+
+    retired.opts.onEvent?.({
+      ...createGatewayEvent(1),
+      event: "presence",
+      payload: {
+        presence: [{ instanceId: retired.opts.instanceId, user: { id: "retired-owner" } }],
+      },
+    });
+
+    expect(replaced).toBe(true);
+    expect(gateway.snapshot.selfUser).toBeNull();
+    expect(gateway.eventLog).toEqual([]);
+    expect(logged).not.toHaveBeenCalled();
+    expect(delivered).not.toHaveBeenCalled();
+  });
+
+  it("does not start a client replaced by a connecting snapshot observer", () => {
+    const { gateway, clients } = createGatewayStore();
+    let replaced = false;
+    gateway.subscribe((snapshot) => {
+      if (snapshot.phase === "connecting" && !replaced) {
+        replaced = true;
+        gateway.connect();
+      }
+    });
+
+    gateway.start();
+
+    expect(clients).toHaveLength(2);
+    expect(clients[0]?.stop).toHaveBeenCalledOnce();
+    expect(clients[0]?.start).not.toHaveBeenCalled();
+    expect(clients[1]?.start).toHaveBeenCalledOnce();
+  });
+
+  it("does not start a client stopped by a connecting snapshot observer", () => {
+    const { gateway, clients } = createGatewayStore();
+    let halted = false;
+    gateway.subscribe((snapshot) => {
+      if (snapshot.phase === "connecting" && !halted) {
+        halted = true;
+        gateway.stop();
+      }
+    });
+
+    gateway.start();
+
+    expect(clients).toHaveLength(1);
+    expect(clients[0]?.stop).toHaveBeenCalledOnce();
+    expect(clients[0]?.start).not.toHaveBeenCalled();
+    expect(gateway.snapshot.phase).toBe("stopped");
+  });
+
+  it("does not reconnect again after a gap observer replaces its client", () => {
+    const { gateway, clients } = createGatewayStore();
+    gateway.start();
+    const retired = clients[0];
+    retired?.opts.onHello?.(HELLO);
+    let replaced = false;
+    gateway.subscribe((snapshot) => {
+      if (snapshot.lastError?.startsWith("event gap detected") && !replaced) {
+        replaced = true;
+        gateway.connect();
+      }
+    });
+
+    retired?.opts.onGap?.({ expected: 2, received: 5 });
+
+    expect(replaced).toBe(true);
+    expect(clients).toHaveLength(2);
+    expect(clients[1]?.start).toHaveBeenCalledOnce();
+    expect(gateway.snapshot.client).toBe(clients[1]);
+  });
+
+  it("does not reconnect after a gap observer stops its client", () => {
+    const { gateway, clients } = createGatewayStore();
+    gateway.start();
+    const retired = clients[0];
+    retired?.opts.onHello?.(HELLO);
+    let halted = false;
+    gateway.subscribe((snapshot) => {
+      if (snapshot.lastError?.startsWith("event gap detected") && !halted) {
+        halted = true;
+        gateway.stop();
+      }
+    });
+
+    retired?.opts.onGap?.({ expected: 2, received: 5 });
+
+    expect(halted).toBe(true);
+    expect(clients).toHaveLength(1);
+    expect(retired?.start).toHaveBeenCalledOnce();
+    expect(gateway.snapshot.phase).toBe("stopped");
+  });
+});

ui/src/app/gateway-store.ts

@@ -30,6 +30,25 @@ const defaultClientFactory: GatewayClientFactory = (opts) => new GatewayBrowserC
 // Grace window before offline presentation appears; reconnects never wait.
 const OFFLINE_INDICATOR_DELAY_MS = 2_000;
 
+function notifyGatewayObservers<T>(
+  listeners: ReadonlySet<(value: T) => void>,
+  value: T,
+  errorLabel: string,
+  isCurrent?: (value: T) => boolean,
+): void {
+  // Snapshot membership because callbacks may mutate subscriptions or replace their owner.
+  for (const listener of Array.from(listeners)) {
+    if (isCurrent && !isCurrent(value)) {
+      return;
+    }
+    try {
+      listener(value);
+    } catch (error) {
+      console.error(`[gateway] ${errorLabel} handler error:`, error);
+    }
+  }
+}
+
 function sameSelfUser(
   left: ApplicationGatewaySnapshot["selfUser"],
   right: ApplicationGatewaySnapshot["selfUser"],
@@ -80,16 +99,14 @@ export function createApplicationGateway(
   // kicking the operator back to the login gate.
   let everConnected = false;
   let stopped = true;
+  // Snapshot observers can synchronously stop or replace their publishing client.
+  const isCurrentClient = (expected: GatewayBrowserClient | null) =>
+    !stopped && client === expected;
   let offlineIndicatorTimer: ReturnType<typeof globalThis.setTimeout> | null = null;
   const listeners = new Set<(next: ApplicationGatewaySnapshot) => void>();
   const eventListeners = new Set<GatewayEventListener>();
   const eventLogListeners = new Set<(events: readonly EventLogEntry[]) => void>();
   let eventLog: EventLogEntry[] = [];
-  const notify = () => {
-    for (const listener of listeners) {
-      listener(snapshot);
-    }
-  };
   const clearOfflineIndicatorTimer = () => {
     if (offlineIndicatorTimer !== null) {
       globalThis.clearTimeout(offlineIndicatorTimer);
@@ -120,7 +137,7 @@ export function createApplicationGateway(
       snapshot = next;
       scheduleOfflineIndicator();
     }
-    notify();
+    notifyGatewayObservers(listeners, snapshot, "snapshot", (current) => current === snapshot);
   };
   const loadCanvasSurfaceLease = (): Promise<CanvasSurfaceLease> => {
     if (canvasSurfaceLease) {
@@ -205,11 +222,6 @@ export function createApplicationGateway(
     // frame that remounts after the socket closes.
     bumpCanvasWidgetFrameConnectionGeneration();
   };
-  const publishEventLog = () => {
-    for (const listener of eventLogListeners) {
-      listener(eventLog);
-    }
-  };
   const updateSettings = (patch: Partial<typeof settings>, selectGateway = false) => {
     const next = { ...settings, ...patch };
     if (!persistConnectionSettings && !selectGateway) {
@@ -230,7 +242,12 @@ export function createApplicationGateway(
         // A live connection owns its authenticated identity until onClose. Older
         // gateways can omit still-connected clients after presence TTL pruning.
         if (selfUser && !sameSelfUser(snapshot.selfUser, selfUser)) {
+          const eventClient = client;
           setSnapshot({ ...snapshot, selfUser });
+          // A presence observer can replace its client before this event reaches the log.
+          if (!isCurrentClient(eventClient)) {
+            return;
+          }
         }
       }
     }
@@ -238,7 +255,7 @@ export function createApplicationGateway(
       0,
       250,
     );
-    publishEventLog();
+    notifyGatewayObservers(eventLogListeners, eventLog, "event");
   };
 
   const connect = (overrides: ApplicationGatewayConnectOptions = {}) => {
@@ -377,7 +394,7 @@ export function createApplicationGateway(
         });
       },
       onGap: ({ expected, received }) => {
-        if (client !== nextClient) {
+        if (!isCurrentClient(nextClient)) {
           return;
         }
         setSnapshot({
@@ -385,7 +402,9 @@ export function createApplicationGateway(
           lastError: `event gap detected (expected seq ${expected}, got ${received}); reconnecting`,
           lastErrorCode: null,
         });
-        connect();
+        if (isCurrentClient(nextClient)) {
+          connect();
+        }
       },
       onEvent: (event) => {
         // A replaced socket can still deliver queued events; never let it
@@ -400,18 +419,8 @@ export function createApplicationGateway(
           // not prevent chat, approvals, or the remaining app from updating.
           console.error("[gateway] event handler error:", error);
         }
-        // Snapshot listeners so subscriptions changed during delivery affect
-        // only the next frame, not sibling consumers of the current frame.
-        for (const listener of Array.from(eventListeners)) {
-          if (client !== nextClient) {
-            return;
-          }
-          try {
-            listener(event);
-          } catch (error) {
-            console.error("[gateway] event listener handler error:", error);
-          }
-        }
+        const isActiveClient = () => isCurrentClient(nextClient);
+        notifyGatewayObservers(eventListeners, event, "event listener", isActiveClient);
       },
     });
     client = nextClient;
@@ -429,7 +438,9 @@ export function createApplicationGateway(
       lastError: null,
       lastErrorCode: null,
     });
-    nextClient.start();
+    if (isCurrentClient(nextClient)) {
+      nextClient.start();
+    }
   };
 
   const gateway: ApplicationGateway = {