fix(desktop): bound screencast delivery to slow consumers

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

@@ -9,6 +9,7 @@ import {
 const makeTarget = (
   send: CdpRelayTarget["send"],
   setDownloadDirectory: CdpRelayTarget["setDownloadDirectory"] = () => {},
+  ackScreencastFrame: CdpRelayTarget["ackScreencastFrame"] = () => {},
 ): CdpRelayTarget => ({
   send,
   targetId: async () => "GUEST-TARGET",
@@ -17,6 +18,7 @@ const makeTarget = (
   userAgent: () => "Electron",
   setDownloadDirectory,
   receivesScreencast: () => true,
+  ackScreencastFrame,
 });
 
 const makeRelay = (target: CdpRelayTarget, write: (raw: string) => void) =>
@@ -138,9 +140,13 @@ 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 () => ({}));
+    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 nativeAcks: Array<number> = [];
+    const relay = makeRelay(
+      makeTarget(send, undefined, (frameId) => nativeAcks.push(frameId)),
+      (raw) => written.push(JSON.parse(raw)),
+    );
     relay.receive(JSON.stringify({ id: 1, method: "Target.setAutoAttach", params: {} }));
     relay.receive(JSON.stringify({ id: 2, method: "Target.attachToBrowserTarget" }));
     await settle();
@@ -170,8 +176,63 @@ 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 = (frameId: number) =>
+      relay.event("Page.screencastFrame", { data: "x", sessionId: frameId }, undefined);
+    const ack = (sessionId: string, frameId: number) =>
+      relay.receive(
+        JSON.stringify({
+          id: 6,
+          method: "Page.screencastFrameAck",
+          params: { sessionId: frameId },
+          sessionId,
+        }),
+      );
+    const delivered = () => {
+      const frames = written
+        .filter((m) => m["method"])
+        .map((m) => [m["sessionId"], (m["params"] as { sessionId?: number }).sessionId]);
+      written.length = 0;
+      return frames;
+    };
+    frame(1);
+    expect(delivered()).toEqual([
+      ["t3-preview-page", 1],
+      [extra, 1],
+    ]);
+    // A session that has not acknowledged its frame gets only the newest one, on its ack.
+    for (let i = 2; i < 100; i++) frame(i);
+    expect(delivered()).toEqual([]);
+    // Frames every session is busy for stay unacknowledged, so Chromium pauses at its in-flight limit.
+    expect(nativeAcks).toEqual([1]);
+    ack(extra, 1);
+    await settle();
+    expect(delivered()).toEqual([[extra, 99]]);
+    // Each consumer ack lets Chromium send one more frame.
+    expect(nativeAcks).toEqual([1, 2]);
+    ack(extra, 99);
+    await settle();
+    expect(delivered()).toEqual([]);
+    expect(nativeAcks).toEqual([1, 2, 3]);
+    frame(100);
+    expect(delivered()).toEqual([[extra, 100]]);
+    ack(extra, 99);
+    await settle();
+    frame(101);
+    expect(delivered()).toEqual([]);
+    expect(send.mock.calls.some(([method]) => method === "Page.screencastFrameAck")).toBe(false);
+    // A restart keeps the unacknowledged frame's slot until its own acknowledgement.
+    relay.receive(JSON.stringify({ id: 7, method: "Page.stopScreencast", sessionId: extra }));
+    relay.receive(JSON.stringify({ id: 8, method: "Page.startScreencast", sessionId: extra }));
+    await settle();
+    delivered();
+    frame(102);
+    expect(delivered()).toEqual([]);
+    ack(extra, 100);
+    await settle();
+    expect(delivered()).toEqual([[extra, 102]]);
+    // Other events still reach a session while its frame is unacknowledged.
+    relay.event("Page.loadEventFired", { timestamp: 1 }, undefined);
+    expect(delivered().map(([session]) => session)).toEqual(["t3-preview-page", extra]);
   });
 });
 

apps/desktop/src/preview/CdpRelay.ts

@@ -11,6 +11,8 @@ export interface CdpRelayTarget {
   readonly sessionDetached?: (owner: string) => void;
   /** Whether a page session, by owner as `send` names it, gets the page's screencast frames. */
   readonly receivesScreencast: (owner: string | undefined) => boolean;
+  /** Acknowledges a native screencast frame, letting Chromium send the next one. */
+  readonly ackScreencastFrame: (frameId: number) => void;
   /** The page's own CDP target id; Playwright keys the main frame by it. */
   readonly targetId: () => Promise<string>;
   readonly url: () => string;
@@ -103,6 +105,20 @@ 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"]]);
+  // Each page session holds one unacknowledged frame and keeps the newest one skipped meanwhile.
+  const unackedFrames = new Map<string, number>();
+  const skippedFrames = new Map<string, Record<string, unknown>>();
+  // Chromium keeps a few frames in flight and reuses one frame id per run, so pacing it means
+  // leaving frames every session is busy for unacknowledged and releasing one per consumer ack.
+  const heldNativeFrames: Array<number> = [];
+  const releaseHeldFramesWithoutHolders = () => {
+    if (skippedFrames.size === 0)
+      for (const frameId of heldNativeFrames.splice(0)) target.ackScreencastFrame(frameId);
+  };
+  const sendFrame = (sessionId: string, params: Record<string, unknown>) => {
+    unackedFrames.set(sessionId, params["sessionId"] as number);
+    send({ method: "Page.screencastFrame", params, sessionId });
+  };
   let sessionSequence = 0;
   const newSession = (kind: "browser" | "page") => {
     const sessionId = `t3-preview-${kind}-${++sessionSequence}`;
@@ -170,6 +186,9 @@ export function createCdpRelayConnection(
         const sessionId = command.params?.["sessionId"];
         if (typeof sessionId === "string" && sessionId !== PAGE_SESSION_ID) {
           sessions.delete(sessionId);
+          unackedFrames.delete(sessionId);
+          skippedFrames.delete(sessionId);
+          releaseHeldFramesWithoutHolders();
           target.sessionDetached?.(sessionId);
         }
         return {};
@@ -202,6 +221,27 @@ 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 (command.method === "Page.screencastFrameAck" && child === undefined) {
+      // A restart keeps the slot too, or restarting would let frames past an unacknowledged one.
+      const sessionId = command.sessionId!;
+      if (unackedFrames.get(sessionId) === command.params?.["sessionId"]) {
+        unackedFrames.delete(sessionId);
+        const skipped = skippedFrames.get(sessionId);
+        skippedFrames.delete(sessionId);
+        if (
+          skipped !== undefined &&
+          target.receivesScreencast(sessionId === PAGE_SESSION_ID ? undefined : sessionId)
+        )
+          sendFrame(sessionId, skipped);
+        const released = heldNativeFrames.shift();
+        if (released !== undefined) target.ackScreencastFrame(released);
+      }
+      return Promise.resolve({});
+    }
+    if (command.method === "Page.stopScreencast" && child === undefined) {
+      skippedFrames.delete(command.sessionId!);
+      releaseHeldFramesWithoutHolders();
+    }
     // The shared debugger keeps Runtime enabled after an earlier connection, and a repeated
     // enable reports no existing contexts; Playwright waits for them, so reset once per connection.
     if (command.method === "Runtime.enable" && child === undefined) {
@@ -262,15 +302,29 @@ export function createCdpRelayConnection(
         send({ method, params, sessionId });
         return;
       }
-      for (const [id, kind] of sessions) {
-        if (kind !== "page") continue;
-        if (
-          method === "Page.screencastFrame" &&
-          !target.receivesScreencast(id === PAGE_SESSION_ID ? undefined : id)
-        )
-          continue;
-        send({ method, params, sessionId: id });
+      if (method === "Page.screencastFrame") {
+        const frame = (params ?? {}) as Record<string, unknown>;
+        const frameId = frame["sessionId"];
+        if (typeof frameId !== "number") return;
+        let held = false;
+        let delivered = false;
+        for (const [id, kind] of sessions) {
+          if (kind !== "page") continue;
+          if (!target.receivesScreencast(id === PAGE_SESSION_ID ? undefined : id)) continue;
+          if (unackedFrames.has(id)) {
+            skippedFrames.set(id, frame);
+            held = true;
+          } else {
+            sendFrame(id, frame);
+            delivered = true;
+          }
+        }
+        if (held && !delivered) heldNativeFrames.push(frameId);
+        else target.ackScreencastFrame(frameId);
+        return;
       }
+      for (const [id, kind] of sessions)
+        if (kind === "page") send({ method, params, sessionId: id });
     },
   };
 }

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

@@ -1157,6 +1157,13 @@ describe("DesktopBrowserHost", () => {
       debuggee.refuse(6);
       debuggee.release();
       expect([yield* Queue.take(replies), yield* Queue.take(replies)].toSorted()).toEqual([5, 6]);
+      yield* raw({
+        id: 7,
+        method: "Page.screencastFrameAck",
+        params: { sessionId: 1 },
+        sessionId: "t3-preview-page-1",
+      });
+      expect(yield* Queue.take(replies)).toBe(7);
       yield* settle;
       frames.length = 0;
       debuggee.emit("Page.screencastFrame", { data: "frame", sessionId: 2 });

apps/desktop/src/preview/DesktopBrowserHost.ts

@@ -421,6 +421,10 @@ export const make = Effect.gen(function* () {
           dropSessionHolds(tab, owner);
         },
         receivesScreencast: (owner) => tab.screencasts.has(owner),
+        ackScreencastFrame: (frameId) =>
+          void debuggee
+            .sendCommand("Page.screencastFrameAck", { sessionId: frameId })
+            .catch(() => {}),
         targetId: () =>
           debuggee
             .sendCommand("Target.getTargetInfo")

apps/desktop/src/preview/Manager.ts

@@ -1108,28 +1108,13 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function
             ),
             scope,
           );
-          const handleDebuggerMessage = Effect.fnUntraced(function* (
-            method: string,
-            params: Record<string, unknown>,
-          ) {
-            if (method === "Page.screencastFrame") {
-              const sessionId = params["sessionId"];
-              if (typeof sessionId === "number") {
-                yield* attemptPromise(
-                  {
-                    operation: "ackScreencastFrame",
-                    webContentsId: wc.id,
-                  },
-                  () => wcDebugger.sendCommand("Page.screencastFrameAck", { sessionId }),
-                ).pipe(Effect.ignore);
-              }
-            }
+          const handleDebuggerMessage = Effect.fnUntraced(function* (method: string) {
             if (method === "Runtime.consoleAPICalled" || method === "Runtime.exceptionThrown") {
               yield* Queue.offer(consoleReleases, undefined);
             }
           });
-          const onMessage: BrowserControlSession["onMessage"] = (_event, method, params) => {
-            runFork(handleDebuggerMessage(method, params));
+          const onMessage: BrowserControlSession["onMessage"] = (_event, method) => {
+            runFork(handleDebuggerMessage(method));
           };
           const control: BrowserControlSession = {
             webContentsId: wc.id,