Merge pull request #13564 from esengine/fix/mux-open-timeout

fix(studio): panes read their own event streams when the shared one never opens

desktop/frontend-next/src/port/event_delivery.test.ts

@@ -0,0 +1,104 @@
+// @vitest-environment jsdom
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { SseHub, type RuntimeView } from "./hub";
+import type { WireEvent } from "./wire";
+
+// Every way a hub can be built has to put a frame the kernel sends into the
+// pane's reducer. A change to the event transport is held to this table, which
+// is where "the page stays connected and goes silent" gets caught.
+const rt = { id: "r1", base: "/rt/r1", root: "/w", name: "w" } as RuntimeView;
+
+interface Source { url: string; closed: boolean; say: (data: object) => void }
+
+function kernel(opts: { muxAnswers: boolean }) {
+  const sources: Source[] = [];
+  const puts: string[] = [];
+  vi.stubGlobal(
+    "EventSource",
+    class {
+      onmessage: ((m: { data: string }) => void) | null = null;
+      onopen: (() => void) | null = null;
+      readonly s: Source;
+      constructor(url: string) {
+        this.s = { url, closed: false, say: (data) => this.onmessage?.({ data: JSON.stringify(data) }) };
+        sources.push(this.s);
+        const answers = url.endsWith("/events/mux") ? opts.muxAnswers : true;
+        // A kernel that answers the mux path with the SPA shell opens a stream
+        // that never says anything, which is what an older kernel does.
+        if (answers && url.endsWith("/events/mux")) queueMicrotask(() => this.s.say({ kind: "mux_open", conn: "c1" }));
+      }
+      close() {
+        this.s.closed = true;
+      }
+    },
+  );
+  vi.stubGlobal("fetch", async (url: string, init: RequestInit = {}) => {
+    if (init.method === "PUT") puts.push(url);
+    return { ok: true, status: 204, json: async () => ({}) } as Response;
+  });
+  return {
+    sources,
+    puts,
+    live: (suffix: string) => sources.filter((s) => !s.closed && s.url.includes(suffix)),
+  };
+}
+
+beforeEach(() => vi.useFakeTimers());
+afterEach(() => {
+  vi.useRealTimers();
+  vi.unstubAllGlobals();
+});
+
+const tick = (ms = 0) => vi.advanceTimersByTimeAsync(ms);
+
+describe("a pane receives what its kernel sends", () => {
+  it("on a local hub, through the shared stream", async () => {
+    const k = kernel({ muxAnswers: true });
+    const got: WireEvent[] = [];
+    const stop = new SseHub().portFor(rt).subscribe((ev) => got.push(ev));
+    await tick();
+    expect(k.puts).toEqual(["/events/mux/c1/panes/r1"]);
+    k.live("/events/mux")[0].say({ pane: "r1", frame: { kind: "text", seq: 1, text: "hi" } });
+    expect(got.map((e) => e.kind)).toEqual(["text"]);
+    expect(k.live("/rt/r1/events")).toHaveLength(0);
+    stop();
+  });
+
+  it("on a hub reached over a relay, through the pane's own stream and never the shared one", async () => {
+    const k = kernel({ muxAnswers: true });
+    const got: WireEvent[] = [];
+    const stop = new SseHub({ mux: false }).portFor(rt).subscribe((ev) => got.push(ev));
+    await tick(20_000);
+    expect(k.sources.some((s) => s.url.endsWith("/events/mux"))).toBe(false);
+    k.live("/rt/r1/events")[0].say({ kind: "text", seq: 1, text: "hi" });
+    expect(got.map((e) => e.kind)).toEqual(["text"]);
+    stop();
+  });
+
+  it("on a kernel that answers the shared path with something else, through the pane's own stream", async () => {
+    const k = kernel({ muxAnswers: false });
+    const got: WireEvent[] = [];
+    const stop = new SseHub().portFor(rt).subscribe((ev) => got.push(ev));
+    await tick(11_000);
+    expect(k.live("/events/mux")).toHaveLength(0);
+    k.live("/rt/r1/events")[0].say({ kind: "text", seq: 1, text: "hi" });
+    expect(got.map((e) => e.kind)).toEqual(["text"]);
+    stop();
+  });
+
+  it("on every one of them, a second frame follows the first in order", async () => {
+    for (const [hub, muxAnswers] of [[new SseHub(), true], [new SseHub({ mux: false }), true], [new SseHub(), false]] as const) {
+      const k = kernel({ muxAnswers });
+      const got: number[] = [];
+      const stop = hub.portFor({ ...rt, id: `r${got.length}` } as RuntimeView).subscribe((ev) => got.push(ev.seq ?? 0));
+      await tick(11_000);
+      const source = k.live("/events/mux")[0] ?? k.live("/rt/")[0];
+      const say = (seq: number) => source.say(source.url.endsWith("/events/mux") ? { pane: "r0", frame: { kind: "tick", seq } } : { kind: "tick", seq });
+      say(1);
+      say(2);
+      expect(got).toEqual([1, 2]);
+      stop();
+      vi.unstubAllGlobals();
+    }
+  });
+});

desktop/frontend-next/src/port/muxclient.test.ts

@@ -164,6 +164,42 @@ describe("panes sharing one stream", () => {
   });
 });
 
+describe("a shared stream that never opens", () => {
+  it("gives each pane a stream of its own once it has waited long enough", async () => {
+    const hub = stubHub();
+    const client = new MuxClient();
+    const a = pane(client, "r1");
+    const b = pane(client, "r2");
+    await vi.advanceTimersByTimeAsync(9_000);
+    expect(a.own).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(2_000);
+    expect(a.own).toHaveBeenCalledOnce();
+    expect(b.own).toHaveBeenCalledOnce();
+    expect(open(hub.sources)).toHaveLength(0);
+    expect(client.carried()).toEqual([]);
+  });
+
+  it("keeps the shared stream when it says it is open in time", async () => {
+    const hub = stubHub();
+    const client = new MuxClient();
+    const a = pane(client, "r1");
+    await vi.advanceTimersByTimeAsync(5_000);
+    hub.sources[0].open("c1");
+    await vi.advanceTimersByTimeAsync(20_000);
+    expect(a.own).not.toHaveBeenCalled();
+    expect(client.carried()).toEqual(["r1"]);
+  });
+
+  it("opens a pane attached afterwards on its own stream straight away", async () => {
+    stubHub();
+    const client = new MuxClient();
+    pane(client, "r1");
+    await vi.advanceTimersByTimeAsync(11_000);
+    const late = pane(client, "r2");
+    expect(late.own).toHaveBeenCalledOnce();
+  });
+});
+
 describe("a pane's subscription over the shared stream", () => {
   it("loses no frame across a reconnect, even when the replay overlaps", async () => {
     const hub = stubHub();

desktop/frontend-next/src/port/muxclient.ts

@@ -3,6 +3,7 @@ import type { MuxAttachRequest, MuxEnvelope, MuxOpened } from "./mux";
 
 const UNKNOWN_CONNECTION = "mux.unknown_connection";
 const RETRY_MS = 2_000;
+const OPEN_WAIT_MS = 10_000;
 
 /** What a pane gives the shared stream: where it stands, where its frames go,
  *  and how to open the stream of its own when the hub cannot carry it. */
@@ -26,12 +27,14 @@ interface Pane {
  *  stood. A pane the hub does not run locally keeps a stream of its own. */
 export class MuxClient {
   private conn: string | null = null;
+  private opening: ReturnType<typeof setTimeout> | null = null;
+  private refused = false;
   private stream: LiveStream | null = null;
   private readonly panes = new Map<string, Pane>();
   private readonly listeners = new Set<() => void>();
 
   attach(id: string, feed: PaneFeed): LiveStream {
-    const pane: Pane = { feed, paused: false, own: null };
+    const pane: Pane = { feed, paused: false, own: this.refused ? feed.own() : null };
     this.panes.set(id, pane);
     this.sync();
     void this.subscribe(id, pane);
@@ -79,10 +82,12 @@ export class MuxClient {
     const needed = this.carried().length > 0;
     if (needed && !this.stream) {
       this.stream = openLiveStream(() => "/events/mux", (raw) => this.hear(raw));
+      this.opening = setTimeout(() => this.giveUp(), OPEN_WAIT_MS);
     } else if (!needed && this.stream) {
       this.stream();
       this.stream = null;
       this.conn = null;
+      this.settled();
     }
   }
 
@@ -90,9 +95,32 @@ export class MuxClient {
     this.stream?.();
     this.stream = null;
     this.conn = null;
+    this.settled();
     this.sync();
   }
 
+  private settled() {
+    if (this.opening) clearTimeout(this.opening);
+    this.opening = null;
+  }
+
+  // A stream that never says it is open is one nothing is subscribed through: an
+  // older kernel, or a proxy that answers the path with something else. Every
+  // pane then reads its own stream, as it did before there was a shared one.
+  private giveUp() {
+    this.opening = null;
+    if (this.conn || !this.stream) return;
+    this.refused = true;
+    this.stream();
+    this.stream = null;
+    for (const pane of this.panes.values()) {
+      if (pane.own) continue;
+      pane.own = pane.feed.own();
+      if (pane.paused) pane.own.pause();
+    }
+    this.listeners.forEach((l) => l());
+  }
+
   private hear(raw: string) {
     let msg: Partial<MuxOpened & MuxEnvelope>;
     try {
@@ -102,6 +130,7 @@ export class MuxClient {
     }
     if (msg.kind === "mux_open" && msg.conn) {
       this.conn = msg.conn;
+      this.settled();
       for (const [id, pane] of this.panes) void this.subscribe(id, pane);
       return;
     }

release-notes/studio/unreleased/13537-mux-open-wait.md

@@ -0,0 +1,2 @@
+- 修复：共享事件流迟迟打不开时（旧版内核或代理不认这条路径），窗格 10 秒后改读各自的事件流，页面不再停在“已连接却收不到事件”。关联 #13562。
+- Fixed: when the shared event stream never opens (an older kernel, or a proxy), panes read their own streams after 10 seconds instead of staying silent. Refs #13562.