From e4568f6bdddfc38d0f8dfe8cd22bde155adac5dd Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 14:19:20 -0700 Subject: [PATCH] fix(ui): isolate gateway observer lifecycles (#117017) Co-authored-by: Peter Steinberger --- ui/src/app/gateway-store.observers.test.ts | 305 +++++++++++++++++++++ ui/src/app/gateway-store.ts | 65 +++-- 2 files changed, 343 insertions(+), 27 deletions(-) create mode 100644 ui/src/app/gateway-store.observers.test.ts diff --git a/ui/src/app/gateway-store.observers.test.ts b/ui/src/app/gateway-store.observers.test.ts new file mode 100644 index 000000000000..57b6f3360cf1 --- /dev/null +++ b/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; + stop: ReturnType; + }> = []; + 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"); + }); +}); diff --git a/ui/src/app/gateway-store.ts b/ui/src/app/gateway-store.ts index 189ec0a32a52..2ae0b5934fd9 100644 --- a/ui/src/app/gateway-store.ts +++ b/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( + 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 | null = null; const listeners = new Set<(next: ApplicationGatewaySnapshot) => void>(); const eventListeners = new Set(); 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 => { 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, 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 = {