perf(desktop): fix log ingest freeze and wrap log lines

desktop/src/main/__tests__/log-store.test.ts

@@ -25,10 +25,11 @@ describe("LogStore", () => {
     expect(logPath).toMatch(/\.log$/)
   })
 
-  it("appends lines to a log file", () => {
+  it("appends lines to a log file", async () => {
     const logPath = store.createLogFile(CTX, "ws-1")
     store.appendLog(logPath, "line 1")
     store.appendLog(logPath, "line 2")
+    await store.closeLog(logPath)
     const content = store.readLogByPath(logPath)
     expect(content).toContain("line 1")
     expect(content).toContain("line 2")
@@ -54,9 +55,10 @@ describe("LogStore", () => {
     expect(store.listLogs(CTX, "nonexistent")).toEqual([])
   })
 
-  it("reads a log file by workspace and filename", () => {
+  it("reads a log file by workspace and filename", async () => {
     const logPath = store.createLogFile(CTX, "ws-1")
     store.appendLog(logPath, "test content")
+    await store.closeLog(logPath)
     const entries = store.listLogs(CTX, "ws-1")
     const content = store.readLog(CTX, "ws-1", entries[0].filename)
     expect(content).toContain("test content")

desktop/src/main/cli.ts

@@ -1,13 +1,26 @@
 import type { ChildProcess } from "node:child_process"
 import { execFile as execFileCb, spawn } from "node:child_process"
 import { createInterface } from "node:readline"
+import type { Readable } from "node:stream"
 import { promisify } from "node:util"
 import type { CLIError, CliLogLine } from "../shared/cli-error.js"
 import { getAnalyticsDistinctId } from "./analytics.js"
 
 const execFile = promisify(execFileCb)
 const MAX_CONCURRENT = 50
 
+function backpressureController(source: Readable): (r: void | Promise<void>) => void {
+  let pending = 0
+  return (result) => {
+    if (!result || typeof (result as Promise<void>).then !== "function") return
+    pending++
+    if (pending === 1) source.pause()
+    void (result as Promise<void>).finally(() => {
+      if (--pending === 0) source.resume()
+    })
+  }
+}
+
 export interface StreamLine {
   /** Raw line text from the CLI (already ANSI-stripped upstream is not assumed). */
   raw: string
@@ -190,7 +203,7 @@ export class CliRunner {
       line: string,
       stream: "stdout" | "stderr",
       meta?: StreamLine,
-    ) => void,
+    ) => void | Promise<void>,
     onExit: (code: number, cliError?: CLIError) => void,
     workspaceId?: string,
   ): Promise<ChildProcess> {
@@ -214,11 +227,13 @@ export class CliRunner {
     let lastCliError: CLIError | undefined
 
     if (child.stdout) {
+      const applyBackpressure = backpressureController(child.stdout)
       const rl = createInterface({ input: child.stdout })
-      rl.on("line", (line) => onLine(line, "stdout"))
+      rl.on("line", (line) => applyBackpressure(onLine(line, "stdout")))
     }
 
     if (child.stderr) {
+      const applyBackpressure = backpressureController(child.stderr)
       const rl = createInterface({ input: child.stderr })
       rl.on("line", (line) => {
         const parsed = parseStderrLine(line)
@@ -231,7 +246,7 @@ export class CliRunner {
           cliError: parsed?.cliError,
           level: normalizeLevel(parsed?.level),
         }
-        onLine(line, "stderr", meta)
+        applyBackpressure(onLine(line, "stderr", meta))
       })
     }
 

desktop/src/main/ipc.ts

@@ -83,6 +83,61 @@ function formatLogLine(line: string, level: "INFO" | "ERROR" = "INFO"): string {
   return `${new Date().toISOString()}\t${level}\t${line}`
 }
 
+interface ProgressSink {
+  line(formatted: string): boolean
+  done(
+    finalLine: string,
+    extra?: { level?: "info" | "warn" | "error"; cliError?: CLIError },
+  ): Promise<void>
+}
+
+function createLogSink(
+  getWin: () => BrowserWindow | null,
+  commandId: string,
+  appendLog?: (line: string) => boolean,
+  flush?: () => Promise<void>,
+): ProgressSink {
+  const FLUSH_MS = 64
+  const MAX_BATCH = 250
+  let buf: string[] = []
+  let timer: ReturnType<typeof setTimeout> | null = null
+
+  function post(
+    done: boolean,
+    extra?: { message?: string; level?: "info" | "warn" | "error"; cliError?: CLIError },
+  ): void {
+    if (timer) {
+      clearTimeout(timer)
+      timer = null
+    }
+    if (!done && buf.length === 0) return
+    const lines = buf
+    buf = []
+    getWin()?.webContents.send("command-progress", {
+      commandId,
+      lines,
+      done,
+      ...extra,
+    })
+  }
+
+  return {
+    line(formatted) {
+      const ok = appendLog?.(formatted) ?? true
+      buf.push(formatted)
+      if (buf.length >= MAX_BATCH) post(false)
+      else if (!timer) timer = setTimeout(() => post(false), FLUSH_MS)
+      return ok
+    },
+    async done(finalLine, extra) {
+      appendLog?.(finalLine)
+      buf.push(finalLine)
+      await flush?.()
+      post(true, { message: finalLine, ...extra })
+    },
+  }
+}
+
 export function registerIpcHandlers(deps: IpcDependencies): {
   tunnelProcesses: Map<string, import("node:child_process").ChildProcess>
   scheduleProviderUpdateCheck: () => void
@@ -549,7 +604,12 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       const wsId = args.workspaceId ?? args.source
       const cmdId = crypto.randomUUID()
       const logPath = logStore.createLogFile(state.workspaceContext(wsId), wsId)
-      const win = deps.getMainWindow()
+      const sink = createLogSink(
+        deps.getMainWindow,
+        cmdId,
+        (line) => logStore.appendLog(logPath, line),
+        () => logStore.closeLog(logPath),
+      )
       let signalledDone = false
 
       // Kill any existing tunnel process for this workspace before starting a new one
@@ -562,45 +622,27 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       const child = await cli.runStreaming(
         cliArgs,
         (line) => {
+          if (signalledDone) return
           const formatted = formatLogLine(line)
 
-          if (!signalledDone && line.includes('"outcome":"success"')) {
-            logStore.appendLog(logPath, formatted)
+          if (line.includes('"outcome":"success"')) {
             signalledDone = true
             // Track this as a tunnel process (it stays alive for the tunnel)
             tunnelProcesses.set(wsId, child)
-            win?.webContents.send("command-progress", {
-              commandId: cmdId,
-              message: formatted,
-              done: true,
-            })
+            void sink.done(formatted)
             return
           }
 
-          if (!signalledDone) {
-            logStore.appendLog(logPath, formatted)
-            win?.webContents.send("command-progress", {
-              commandId: cmdId,
-              message: formatted,
-              done: false,
-            })
-          }
+          if (!sink.line(formatted)) return logStore.onDrain(logPath)
         },
         (code) => {
           if (tunnelProcesses.get(wsId) === child) {
             tunnelProcesses.delete(wsId)
           }
           if (signalledDone) return
-          const exitMsg = formatLogLine(
-            `Exit code: ${code}`,
-            code === 0 ? "INFO" : "ERROR",
+          void sink.done(
+            formatLogLine(`Exit code: ${code}`, code === 0 ? "INFO" : "ERROR"),
           )
-          logStore.appendLog(logPath, exitMsg)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: exitMsg,
-            done: true,
-          })
         },
         wsId,
       )
@@ -616,33 +658,25 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       await quiesceWorkspace(args.workspaceId)
       const cmdId = crypto.randomUUID()
       const logPath = logStore.createLogFile(state.workspaceContext(args.workspaceId), args.workspaceId)
-      const win = deps.getMainWindow()
+      const sink = createLogSink(
+        deps.getMainWindow,
+        cmdId,
+        (line) => logStore.appendLog(logPath, line),
+        () => logStore.closeLog(logPath),
+      )
 
       const cliArgs = ["workspace", "stop", args.workspaceId]
       if (args.debug) cliArgs.push("--debug")
 
       cli.runStreaming(
         cliArgs,
         (line) => {
-          const formatted = formatLogLine(line)
-          logStore.appendLog(logPath, formatted)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: formatted,
-            done: false,
-          })
+          if (!sink.line(formatLogLine(line))) return logStore.onDrain(logPath)
         },
         (code) => {
-          const exitMsg = formatLogLine(
-            `Exit code: ${code}`,
-            code === 0 ? "INFO" : "ERROR",
+          void sink.done(
+            formatLogLine(`Exit code: ${code}`, code === 0 ? "INFO" : "ERROR"),
           )
-          logStore.appendLog(logPath, exitMsg)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: exitMsg,
-            done: true,
-          })
         },
         args.workspaceId,
       )
@@ -663,7 +697,12 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       await quiesceWorkspace(args.workspaceId)
       const cmdId = crypto.randomUUID()
       const logPath = logStore.createLogFile(state.workspaceContext(args.workspaceId), args.workspaceId)
-      const win = deps.getMainWindow()
+      const sink = createLogSink(
+        deps.getMainWindow,
+        cmdId,
+        (line) => logStore.appendLog(logPath, line),
+        () => logStore.closeLog(logPath),
+      )
 
       const cliArgs = ["workspace", "delete", args.workspaceId]
       if (args.debug) cliArgs.push("--debug")
@@ -672,25 +711,12 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       cli.runStreaming(
         cliArgs,
         (line) => {
-          const formatted = formatLogLine(line)
-          logStore.appendLog(logPath, formatted)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: formatted,
-            done: false,
-          })
+          if (!sink.line(formatLogLine(line))) return logStore.onDrain(logPath)
         },
         (code) => {
-          const exitMsg = formatLogLine(
-            `Exit code: ${code}`,
-            code === 0 ? "INFO" : "ERROR",
+          void sink.done(
+            formatLogLine(`Exit code: ${code}`, code === 0 ? "INFO" : "ERROR"),
           )
-          logStore.appendLog(logPath, exitMsg)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: exitMsg,
-            done: true,
-          })
         },
         args.workspaceId,
       )
@@ -705,33 +731,25 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       trackEvent("workspace_rebuild")
       const cmdId = crypto.randomUUID()
       const logPath = logStore.createLogFile(state.workspaceContext(args.workspaceId), args.workspaceId)
-      const win = deps.getMainWindow()
+      const sink = createLogSink(
+        deps.getMainWindow,
+        cmdId,
+        (line) => logStore.appendLog(logPath, line),
+        () => logStore.closeLog(logPath),
+      )
 
       const cliArgs = ["workspace", "up", args.workspaceId, "--recreate"]
       if (args.debug) cliArgs.push("--debug")
 
       cli.runStreaming(
         cliArgs,
         (line) => {
-          const formatted = formatLogLine(line)
-          logStore.appendLog(logPath, formatted)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: formatted,
-            done: false,
-          })
+          if (!sink.line(formatLogLine(line))) return logStore.onDrain(logPath)
         },
         (code) => {
-          const exitMsg = formatLogLine(
-            `Exit code: ${code}`,
-            code === 0 ? "INFO" : "ERROR",
+          void sink.done(
+            formatLogLine(`Exit code: ${code}`, code === 0 ? "INFO" : "ERROR"),
           )
-          logStore.appendLog(logPath, exitMsg)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: exitMsg,
-            done: true,
-          })
         },
         args.workspaceId,
       )
@@ -746,33 +764,25 @@ export function registerIpcHandlers(deps: IpcDependencies): {
       trackEvent("workspace_reset")
       const cmdId = crypto.randomUUID()
       const logPath = logStore.createLogFile(state.workspaceContext(args.workspaceId), args.workspaceId)
-      const win = deps.getMainWindow()
+      const sink = createLogSink(
+        deps.getMainWindow,
+        cmdId,
+        (line) => logStore.appendLog(logPath, line),
+        () => logStore.closeLog(logPath),
+      )
 
       const cliArgs = ["workspace", "up", args.workspaceId, "--reset"]
       if (args.debug) cliArgs.push("--debug")
 
       cli.runStreaming(
         cliArgs,
         (line) => {
-          const formatted = formatLogLine(line)
-          logStore.appendLog(logPath, formatted)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: formatted,
-            done: false,
-          })
+          if (!sink.line(formatLogLine(line))) return logStore.onDrain(logPath)
         },
         (code) => {
-          const exitMsg = formatLogLine(
-            `Exit code: ${code}`,
-            code === 0 ? "INFO" : "ERROR",
+          void sink.done(
+            formatLogLine(`Exit code: ${code}`, code === 0 ? "INFO" : "ERROR"),
           )
-          logStore.appendLog(logPath, exitMsg)
-          win?.webContents.send("command-progress", {
-            commandId: cmdId,
-            message: exitMsg,
-            done: true,
-          })
         },
         args.workspaceId,
       )