`68a4c05`

perf: per-pty event channel + spawn-lock dedupe + chunked replay

Authored by

espadonne 1 month ago

SHA: 68a4c05db53fa6fa0a2cce1f8886472195c80f48
Parents: b85ff40
Tree: 22d6cd6

3 changed files

Status	File	+	-
M	`src-tauri/src/commands.rs`	14	5
M	`src/components/TerminalPane.tsx`	141	59
M	`src/lib/ipc/client.ts`	7	4

src-tauri/src/commands.rsmodified

      );
      // Forwarder task: base64-encode each chunk off `data_rx` and emit
 -    // `pty:data`. When `data_rx` closes (reader task hit EOF), wait
 -    // on `exit_rx` for the final exit code and emit `pty:exit`. Then
 -    // remove our entry from active_ptys — if close_pty already
 -    // removed it, the remove here is a harmless no-op.
 +    // a per-PTY event named `pty:data:<pty_id>`. Per-PTY events
 +    // avoid the O(N) fan-out that the previous single `pty:data`
 +    // channel had — every TerminalPane subscribes only to events
 +    // for its own ptyId, so one PTY's stdout chunk doesn't wake up
 +    // every listener in the webview.
 +    //
 +    // When `data_rx` closes (reader task hit EOF), wait on
 +    // `exit_rx` for the final exit code and emit `pty:exit`
 +    // (globally, since the store subscribes to it to clean up
 +    // `ptyIds` and doesn't want to listen per-pty). Then remove
 +    // our entry from active_ptys — if close_pty already removed
 +    // it, the remove here is a harmless no-op.
      let app_clone = app.clone();
      let active_ptys = state.active_ptys.clone();
      let pid = pty_id.clone();
 +    let data_event = format!("pty:data:{pid}");
      tauri::async_runtime::spawn(async move {
          while let Some(chunk) = data_rx.recv().await {
              let b64 = BASE64_STANDARD.encode(&chunk);
                  pty_id: pid.clone(),
                  base64: b64,
              };
 -            if let Err(err) = app_clone.emit("pty:data", &payload) {
 +            if let Err(err) = app_clone.emit(&data_event, &payload) {
                  tracing::warn!(error = %err, "failed to emit pty:data");
+             }
+         }

src/components/TerminalPane.tsxmodified

    void logFrontend("debug", "TerminalPane", payload);
  };
 +/** Module-level dedup lock for concurrent spawn attempts on the
 + *  same `sessionId`. React 19 StrictMode double-invokes effects
 + *  (mount → cleanup → mount), and both invocations used to race
 + *  into `spawn_pty`, producing TWO claude subprocesses per click.
 + *  The first one was then orphaned — its ptyId never landed in
 + *  the store so the frontend lost track of it, but the backend
 + *  kept the subprocess alive until window-destroy.
 + *
 + *  With this map, the second mount awaits the first mount's
 + *  promise and reuses the same ptyId. Exactly one subprocess per
 + *  session, StrictMode-safe. */
 +const spawnLocks = new Map<string, Promise<string>>();
++
 +async function getOrSpawnPty(
 +  sessionId: string,
 +  cwd: string,
 +  claudeArgs: string[],
 +  cols: number,
 +  rows: number,
 +): Promise<string> {
 +  const existing = useSessionStore.getState().ptyIds.get(sessionId);
 +  if (existing) return existing;
++
 +  const inflight = spawnLocks.get(sessionId);
 +  if (inflight) return inflight;
++
 +  const promise = (async () => {
 +    const newId = crypto.randomUUID();
 +    trace("spawn path", { ptyId: newId, cols, rows });
 +    await spawnPty({
 +      ptyId: newId,
 +      sessionId,
 +      cwd,
 +      args: claudeArgs,
 +      cols,
 +      rows,
 +    });
 +    // Register in the store before returning so any second mount
 +    // awaiting the same promise can look up the ptyId via the
 +    // `existing` check above on its next call.
 +    useSessionStore.getState().registerPty(sessionId, {
 +      ptyId: newId,
 +      sessionId,
 +      cwd,
 +      startedAt: new Date().toISOString(),
 +    });
 +    trace("spawn_pty ok", { ptyId: newId });
 +    return newId;
 +  })();
++
 +  spawnLocks.set(sessionId, promise);
 +  try {
 +    return await promise;
 +  } finally {
 +    spawnLocks.delete(sessionId);
 +  }
 +}
++
 +/** Write a large base64 payload into xterm in paced chunks so we
 + *  don't block the UI thread. xterm.write accepts a callback that
 + *  fires once the write has been parsed and rendered, which we use
 + *  to schedule the next chunk. For a fresh mount with a 200 KB
 + *  ring buffer replay this keeps the main thread responsive
 + *  (keystrokes still register) instead of dropping a multi-second
 + *  stall while xterm parses the whole blob. */
 +const REPLAY_CHUNK_BYTES = 8 * 1024;
++
 +function writeBase64Chunked(term: Terminal, b64: string): void {
 +  const binary = atob(b64);
 +  const total = binary.length;
 +  if (total === 0) return;
 +  if (total <= REPLAY_CHUNK_BYTES) {
 +    term.write(decodeBinary(binary));
 +    return;
 +  }
 +  let offset = 0;
 +  const step = () => {
 +    const end = Math.min(offset + REPLAY_CHUNK_BYTES, total);
 +    const slice = decodeBinary(binary.slice(offset, end));
 +    offset = end;
 +    if (offset >= total) {
 +      term.write(slice);
 +    } else {
 +      term.write(slice, step);
 +    }
 +  };
 +  step();
 +}
++
 +function decodeBinary(binary: string): Uint8Array {
 +  const bytes = new Uint8Array(binary.length);
 +  for (let i = 0; i < binary.length; i++) {
 +    bytes[i] = binary.charCodeAt(i);
 +  }
 +  return bytes;
 +}
++
  /** Theme matching the claudex design tokens in `src/index.css`. */
  const TERMINAL_THEME = {
    foreground: "#d4d4d8",
  export function TerminalPane({ sessionId, cwd, claudeArgs }: TerminalPaneProps) {
    const containerRef = useRef<HTMLDivElement | null>(null);
 -  const registerPty = useSessionStore((s) => s.registerPty);
    useEffect(() => {
      const container = containerRef.current;
      let cancelled = false;
      let disposeListeners: Array<() => void> = [];
 -    const writeBase64 = (b64: string) => {
 -      const binary = atob(b64);
 -      const bytes = new Uint8Array(binary.length);
 -      for (let i = 0; i < binary.length; i++) {
 -        bytes[i] = binary.charCodeAt(i);
 -      }
 -      term.write(bytes);
 -    };
+-
      const attach = async () => {
 +      // Resolve a ptyId — either reattach to an already-running
 +      // subprocess or dedupe-safely spawn a new one. The spawn
 +      // lock guarantees at most one spawn per sessionId even under
 +      // React 19 StrictMode's double-mount.
        const existing = useSessionStore.getState().ptyIds.get(sessionId);
 -      if (existing) {
 -        trace("reattach path", { ptyId: existing });
 -        ptyId = existing;
 +      const reattach = !!existing;
 +      let resolved: string;
 +      try {
 +        resolved = await getOrSpawnPty(
 +          sessionId,
 +          cwd,
 +          claudeArgs,
 +          term.cols,
 +          term.rows,
 +        );
 +      } catch (err) {
 +        trace("spawn_pty failed", { error: String(err) });
 +        if (!cancelled) {
 +          term.write(
 +            `\r\n\x1b[31m[claudex] spawn_pty failed: ${formatErr(err)}\x1b[0m\r\n`,
 +          );
 +        }
 +        return;
 +      }
 +      if (cancelled) return;
 +      ptyId = resolved;
++
 +      // On reattach, replay the ring buffer into xterm in paced
 +      // chunks so a 200 KB scrollback doesn't stall the main
 +      // thread for multiple seconds.
 +      if (reattach) {
 +        trace("reattach path", { ptyId: resolved });
          try {
 -          const snapshot = await getPtyBuffer(existing);
 +          const snapshot = await getPtyBuffer(resolved);
            if (cancelled) return;
 -          if (snapshot.length > 0) writeBase64(snapshot);
 +          if (snapshot.length > 0) writeBase64Chunked(term, snapshot);
            trace("ring buffer replayed", { bytes: snapshot.length });
          } catch (err) {
            trace("get_pty_buffer failed during reattach", {
              error: String(err),
            });
+         }
 -      } else {
 -        const newId = crypto.randomUUID();
 -        trace("spawn path", {
 -          ptyId: newId,
 -          cols: term.cols,
 -          rows: term.rows,
 -        });
 -        try {
 -          await spawnPty({
 -            ptyId: newId,
 -            sessionId,
 -            cwd,
 -            args: claudeArgs,
 -            cols: term.cols,
 -            rows: term.rows,
 -          });
 -          trace("spawn_pty ok", { ptyId: newId });
 -        } catch (err) {
 -          trace("spawn_pty failed", { error: String(err) });
 -          if (!cancelled) {
 -            term.write(
 -              `\r\n\x1b[31m[claudex] spawn_pty failed: ${formatErr(err)}\x1b[0m\r\n`,
 -            );
 -          }
 -          return;
 -        }
 -        if (cancelled) return;
 -        ptyId = newId;
 -        registerPty(sessionId, {
 -          ptyId: newId,
 -          sessionId,
 -          cwd,
 -          startedAt: new Date().toISOString(),
 -        });
+       }
 -      // Listen for stdout. `pty:data` is a single shared channel, so
 -      // we filter by our own pty id inside the callback.
 +      // Listen for stdout — per-pty event name, so this listener
 +      // only wakes up for our own terminal's bytes.
        try {
 -        const un = await onPtyData((ev: PtyDataEvent) => {
 -          if (ev.ptyId !== ptyId) return;
 -          writeBase64(ev.base64);
 +        const pid = resolved;
 +        const un = await onPtyData(pid, (ev: PtyDataEvent) => {
 +          writeBase64Chunked(term, ev.base64);
          });
          if (cancelled) {
            un();
            unlistenData = un;
+         }
        } catch (err) {
 -        console.warn("[pty] onPtyData listen failed", err);
 +        trace("onPtyData listen failed", { error: String(err) });
+       }
        // Wire keystrokes → PTY writer.
        const dataDispose = term.onData((data) => {
          if (!ptyId) return;
          void writePty(ptyId, data).catch((err) => {
 -          console.warn("[pty] write_pty failed", err);
 +          trace("write_pty failed", { error: String(err) });
          });
        });
        disposeListeners.push(() => dataDispose.dispose());
 -      // Propagate xterm resize events (triggered by fit.fit() below)
 -      // down to the PTY master so claude's TUI redraws properly.
 +      // Propagate xterm resize events (triggered by fit.fit() or
 +      // container ResizeObserver) down to the PTY master so
 +      // claude's TUI redraws properly.
        const resizeDispose = term.onResize(({ cols, rows }) => {
          if (!ptyId) return;
          void resizePty(ptyId, cols, rows).catch((err) => {
 -          console.warn("[pty] resize_pty failed", err);
 +          trace("resize_pty failed", { error: String(err) });
          });
        });
        disposeListeners.push(() => resizeDispose.dispose());

src/lib/ipc/client.tsmodified

    return invoke<PtyInfo[]>("list_ptys");
+ }
 -/** Subscribe to `pty:data` payloads. The frontend filters by its
 - *  own `ptyId` inside the callback — one event channel is shared
 - *  across every live PTY. Returns an unlisten function. */
 +/** Subscribe to stdout chunks for one specific live PTY. The Rust
 + *  backend emits per-PTY events named `pty:data:<ptyId>` so each
 + *  `TerminalPane` listener only wakes up for its own stdout — no
 + *  fan-out cost as the number of live PTYs grows. Returns an
 + *  unlisten function. */
  export async function onPtyData(
 +  ptyId: string,
    cb: (ev: PtyDataEvent) => void,
  ): Promise<UnlistenFn> {
 -  return listen<PtyDataEvent>("pty:data", (e) => cb(e.payload));
 +  return listen<PtyDataEvent>(`pty:data:${ptyId}`, (e) => cb(e.payload));
+ }
  /** Subscribe to `pty:exit` payloads. Fires once per PTY when its