fix(desktop): bound screencast delivery to slow consumers

apps/desktop/src/preview/CdpRelay.test.ts

@@ -132,8 +132,71 @@ describe("CDP relay", () => {
 });
 
 describe("CDP relay extra sessions", () => {
-  it("gives a second page session the page's commands and events", async () => {
-    const send = vi.fn(async () => ({}));
+  it.each(
+    [
+      { committed: true, newerSucceeds: false, olderSucceeds: false },
+      { committed: true, newerSucceeds: true, olderSucceeds: false },
+      { committed: false, newerSucceeds: true, olderSucceeds: false },
+      { committed: false, newerSucceeds: false, olderSucceeds: true },
+      { committed: false, newerSucceeds: false, olderSucceeds: false },
+    ].flatMap((scenario) => [true, false].map((olderFirst) => ({ ...scenario, olderFirst }))),
+  )(
+    "preserves only committed streams across overlapping starts (%j)",
+    async ({ committed, newerSucceeds, olderSucceeds, olderFirst }) => {
+      const send = vi.fn<CdpRelayTarget["send"]>(async () => ({}));
+      const written: Array<Record<string, unknown>> = [];
+      const relay = makeRelay(makeTarget(send), (raw) => written.push(JSON.parse(raw)));
+      const command = (id: number, method: string, params = {}) =>
+        relay.receive(JSON.stringify({ id, method, params, sessionId: "t3-preview-page" }));
+      const frame = (sessionId: number) =>
+        relay.event("Page.screencastFrame", { data: "x", sessionId }, undefined);
+      const frames = () =>
+        written.filter((message) => message["method"] === "Page.screencastFrame");
+      relay.receive(JSON.stringify({ id: 1, method: "Target.setAutoAttach" }));
+      if (committed) {
+        command(2, "Page.startScreencast");
+        await settle();
+        frame(7);
+        expect(frames()).toHaveLength(1);
+      }
+      await settle();
+      written.length = 0;
+      const older = Promise.withResolvers<unknown>();
+      const newer = Promise.withResolvers<unknown>();
+      send.mockImplementationOnce(() => older.promise).mockImplementationOnce(() => newer.promise);
+      command(3, "Page.startScreencast");
+      command(4, "Page.startScreencast");
+      frame(8);
+      expect(frames()).toEqual([]);
+      for (const [pending, succeeds] of olderFirst
+        ? ([
+            [older, olderSucceeds],
+            [newer, newerSucceeds],
+          ] as const)
+        : ([
+            [newer, newerSucceeds],
+            [older, olderSucceeds],
+          ] as const)) {
+        if (succeeds) pending.resolve({});
+        else pending.reject(new Error("Refused"));
+        await settle();
+      }
+      if (committed) {
+        frame(9);
+        expect(frames()).toEqual([]);
+        command(5, "Page.screencastFrameAck", { sessionId: 7 });
+      }
+      frame(9);
+      frame(9);
+      expect(frames()).toHaveLength(committed || newerSucceeds || olderSucceeds ? 1 : 0);
+      expect(written.filter((message) => message["error"])).toHaveLength(
+        Number(!olderSucceeds) + Number(!newerSucceeds),
+      );
+    },
+  );
+
+  it("bounds screencast delivery independently for each page session", async () => {
+    const send = vi.fn<CdpRelayTarget["send"]>(async () => ({}));
     const written: Array<Record<string, unknown>> = [];
     const relay = makeRelay(makeTarget(send), (raw) => written.push(JSON.parse(raw)));
     relay.receive(JSON.stringify({ id: 1, method: "Target.setAutoAttach", params: {} }));
@@ -165,8 +228,79 @@ describe("CDP relay extra sessions", () => {
     await settle();
     expect(send).toHaveBeenLastCalledWith("Page.startScreencast", {}, undefined, extra);
     written.length = 0;
-    relay.event("Page.screencastFrame", { data: "x" }, undefined);
-    expect(written.map((m) => m["sessionId"])).toEqual(["t3-preview-page", extra]);
+    const frame = (sessionId = 7) =>
+      relay.event("Page.screencastFrame", { data: "x", sessionId }, undefined);
+    for (let i = 0; i < 100; i++) frame();
+    expect(written).toHaveLength(1);
+    expect(written.map((m) => m["sessionId"])).toEqual([extra]);
+
+    relay.receive(
+      JSON.stringify({
+        id: 6,
+        method: "Page.startScreencast",
+        sessionId: "t3-preview-page",
+      }),
+    );
+    await settle();
+    written.length = 0;
+    frame();
+    expect(written.map((m) => m["sessionId"])).toEqual(["t3-preview-page"]);
+    const acknowledge = (sessionId: string, frameId = 7) =>
+      relay.receive(
+        JSON.stringify({
+          id: 7,
+          method: "Page.screencastFrameAck",
+          params: { sessionId: frameId },
+          sessionId,
+        }),
+      );
+    acknowledge(extra);
+    await settle();
+    written.length = 0;
+    frame();
+    frame();
+    expect(written.map((m) => m["sessionId"])).toEqual([extra]);
+    expect(send.mock.calls.filter(([method]) => method === "Page.screencastFrameAck")).toEqual([]);
+
+    relay.receive(JSON.stringify({ id: 8, method: "Page.stopScreencast", sessionId: extra }));
+    relay.receive(JSON.stringify({ id: 9, method: "Page.startScreencast", sessionId: extra }));
+    await settle();
+    written.length = 0;
+    frame(8);
+    expect(written).toEqual([]);
+    acknowledge(extra);
+    await settle();
+    written.length = 0;
+    frame(8);
+    acknowledge(extra); // A late acknowledgement from the previous stream must not release this one.
+    await settle();
+    written.length = 0;
+    frame(8);
+    expect(written).toEqual([]);
+    acknowledge(extra, 8);
+    await settle();
+    written.length = 0;
+    frame(8);
+    expect(written.map((m) => m["sessionId"])).toEqual([extra]);
+
+    const restarting = Promise.withResolvers<unknown>();
+    send.mockReturnValueOnce(restarting.promise);
+    relay.receive(JSON.stringify({ id: 10, method: "Page.startScreencast", sessionId: extra }));
+    relay.receive(
+      JSON.stringify({
+        id: 11,
+        method: "Target.detachFromTarget",
+        params: { sessionId: extra },
+      }),
+    );
+    restarting.resolve({});
+    await settle();
+    written.length = 0;
+    frame(8);
+    expect(written).toEqual([]);
+    // Other protocol events still reach the main page session while its frame is pending.
+    relay.event("Page.loadEventFired", { timestamp: 1 }, undefined);
+    expect(written.map((m) => m["method"])).toEqual(["Page.loadEventFired"]);
   });
 });
 

apps/desktop/src/preview/CdpRelay.ts

@@ -101,6 +101,9 @@ export function createCdpRelayConnection(
   let downloadEventsEnabled = false;
   /** Relay-created page sessions share one real debugger; child frame sessions retain their own ids. */
   const sessions = new Map<string, "browser" | "page">([[PAGE_SESSION_ID, "page"]]);
+  // The manager acknowledges native frames immediately; each relay consumer must bound its own queue.
+  const screencasts = new Map<string, { pendingFrame: number | null; active: boolean }>();
+  const screencastStarts = new Map<string, object>();
   let sessionSequence = 0;
   const newSession = (kind: "browser" | "page") => {
     const sessionId = `t3-preview-${kind}-${++sessionSequence}`;
@@ -168,6 +171,8 @@ export function createCdpRelayConnection(
         const sessionId = command.params?.["sessionId"];
         if (typeof sessionId === "string" && sessionId !== PAGE_SESSION_ID) {
           sessions.delete(sessionId);
+          screencasts.delete(sessionId);
+          screencastStarts.delete(sessionId);
           target.sessionDetached?.(sessionId);
         }
         return {};
@@ -198,6 +203,33 @@ export function createCdpRelayConnection(
     }
     // Playwright asks for this on every page; the host emulates focus only while it is in use.
     if (command.method === "Emulation.setFocusEmulationEnabled") return Promise.resolve({});
+    if (child === undefined) {
+      const sessionId = command.sessionId!;
+      if (command.method === "Page.screencastFrameAck") {
+        const stream = screencasts.get(sessionId);
+        if (stream && stream.pendingFrame === command.params?.["sessionId"]) {
+          stream.pendingFrame = null;
+        }
+        return Promise.resolve({});
+      }
+      if (command.method === "Page.stopScreencast") {
+        const stream = screencasts.get(sessionId);
+        if (stream) stream.active = false;
+        screencastStarts.delete(sessionId);
+      }
+      if (command.method === "Page.startScreencast") {
+        const start = screencastStarts.get(sessionId) ?? {};
+        screencastStarts.set(sessionId, start);
+        return target.send(command.method, command.params ?? {}, child, owner).then((result) => {
+          if (screencastStarts.get(sessionId) === start) {
+            const stream = screencasts.get(sessionId);
+            if (stream) stream.active = true;
+            else screencasts.set(sessionId, { pendingFrame: null, active: true });
+          }
+          return result;
+        });
+      }
+    }
     return target.send(command.method, command.params ?? {}, child, owner);
   };
 
@@ -250,6 +282,16 @@ export function createCdpRelayConnection(
         send({ method, params, sessionId });
         return;
       }
+      if (method === "Page.screencastFrame") {
+        const frameId = (params as { sessionId?: unknown } | null)?.sessionId;
+        if (typeof frameId !== "number") return;
+        for (const [id, stream] of screencasts) {
+          if (!stream.active || stream.pendingFrame !== null) continue;
+          stream.pendingFrame = frameId;
+          send({ method, params, sessionId: id });
+        }
+        return;
+      }
       for (const [id, kind] of sessions)
         if (kind === "page") send({ method, params, sessionId: id });
     },

apps/desktop/src/preview/DesktopBrowserHost.test.ts

@@ -40,7 +40,7 @@ const makeDebuggee = (
   } = {},
 ) => {
   const emitter = new NodeEvents.EventEmitter();
-  const pending: Array<() => void> = [];
+  const pending: Array<{ resolve: () => void; reject: () => void }> = [];
   const focusChanges: Array<() => void> = [];
   const sent: Array<unknown> = [];
   const sendWaiters: Array<() => void> = [];
@@ -67,7 +67,10 @@ const makeDebuggee = (
           return;
         }
         sent.push(params?.id);
-        pending.push(() => resolve({ method }));
+        pending.push({
+          resolve: () => resolve({ method }),
+          reject: () => reject(new Error("Refused")),
+        });
         sendWaiters.shift()?.();
       }),
   });
@@ -96,8 +99,11 @@ const makeDebuggee = (
     sent,
     focusEmulation,
     nextSend: () => new Promise<void>((resolve) => sendWaiters.push(resolve)),
-    release: () => pending.splice(0).forEach((resolve) => resolve()),
-    releaseFirst: () => pending.shift()?.(),
+    release: () => pending.splice(0).forEach(({ resolve }) => resolve()),
+    releaseFirst: () => pending.shift()?.resolve(),
+    releaseLast: () => pending.pop()?.resolve(),
+    refuseFirst: () => pending.shift()?.reject(),
+    refuseLast: () => pending.pop()?.reject(),
     settleFocus: () => focusChanges.splice(0).forEach((resolve) => resolve()),
   };
 };
@@ -739,6 +745,108 @@ describe("DesktopBrowserHost", () => {
     }).pipe(Effect.scoped),
   );
 
+  it.effect("preserves committed painting holds across failed and overlapping starts", () =>
+    Effect.gen(function* () {
+      const host = yield* DesktopBrowserHost.make;
+      const debuggee = makeDebuggee();
+      host.attach(key, debuggee.tab);
+      const replies = yield* watchReplies(host);
+      const painting = yield* watchPainting(host);
+
+      yield* command(host, 1, "Page.startScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(1);
+      yield* command(host, 2, "Page.startScreencast");
+      yield* settle;
+      debuggee.refuseFirst();
+      expect(yield* Queue.take(replies)).toBe(2);
+      expect(painting).toEqual([true]);
+
+      yield* command(host, 3, "Page.stopScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(3);
+      expect(painting).toEqual([true, false]);
+
+      yield* command(host, 4, "Page.startScreencast");
+      yield* command(host, 5, "Page.startScreencast");
+      yield* settle;
+      debuggee.releaseLast();
+      expect(yield* Queue.take(replies)).toBe(5);
+      debuggee.refuseFirst();
+      expect(yield* Queue.take(replies)).toBe(4);
+      expect(painting).toEqual([true, false, true]);
+
+      yield* command(host, 6, "Page.stopScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(6);
+      expect(painting).toEqual([true, false, true, false]);
+
+      yield* command(host, 7, "Page.startScreencast");
+      yield* command(host, 8, "Page.startScreencast");
+      yield* settle;
+      debuggee.refuseLast();
+      expect(yield* Queue.take(replies)).toBe(8);
+      debuggee.refuseFirst();
+      expect(yield* Queue.take(replies)).toBe(7);
+      expect(painting).toEqual([true, false, true, false, true, false]);
+
+      yield* command(host, 9, "Page.startScreencast");
+      yield* settle;
+      yield* command(host, 10, "Page.stopScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(9);
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(10);
+      expect(painting).toEqual([true, false, true, false, true, false, true, false]);
+      yield* command(host, 11, "Page.startScreencast");
+      yield* command(host, 12, "Page.startScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(11);
+      debuggee.refuseLast();
+      expect(yield* Queue.take(replies)).toBe(12);
+      expect(painting.at(-1)).toBe(true);
+
+      yield* command(host, 13, "Page.stopScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(13);
+      expect(painting.at(-1)).toBe(false);
+
+      yield* command(host, 14, "Page.startScreencast");
+      yield* command(host, 15, "Page.stopScreencast");
+      yield* command(host, 16, "Page.startScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(14);
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(15);
+      debuggee.refuseLast();
+      expect(yield* Queue.take(replies)).toBe(16);
+      expect(painting.at(-1)).toBe(false);
+      yield* command(host, 17, "Page.startScreencast");
+      yield* command(host, 18, "Page.startScreencast");
+      yield* settle;
+      debuggee.refuseLast();
+      expect(yield* Queue.take(replies)).toBe(18);
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(17);
+      expect(painting.at(-1)).toBe(true);
+
+      yield* command(host, 19, "Page.stopScreencast");
+      yield* settle;
+      debuggee.releaseFirst();
+      expect(yield* Queue.take(replies)).toBe(19);
+      expect(painting.at(-1)).toBe(false);
+      yield* TestClock.adjust("1 minute");
+      expect(debuggee.focusEmulation.at(-1)).toBe(false);
+    }).pipe(Effect.scoped),
+  );
+
   it.effect("lets a guest throttle after a person's input through a viewer that then closed", () =>
     Effect.gen(function* () {
       const host = yield* DesktopBrowserHost.make;