diff --git a/desktop/src/main/__tests__/log-store.test.ts b/desktop/src/main/__tests__/log-store.test.ts index 1a1ad80a7..7c2792526 100644 --- a/desktop/src/main/__tests__/log-store.test.ts +++ b/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") diff --git a/desktop/src/main/cli.ts b/desktop/src/main/cli.ts index c57d8e648..76fd4fb3c 100644 --- a/desktop/src/main/cli.ts +++ b/desktop/src/main/cli.ts @@ -1,6 +1,7 @@ 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" @@ -8,6 +9,18 @@ import { getAnalyticsDistinctId } from "./analytics.js" const execFile = promisify(execFileCb) const MAX_CONCURRENT = 50 +function backpressureController(source: Readable): (r: void | Promise) => void { + let pending = 0 + return (result) => { + if (!result || typeof (result as Promise).then !== "function") return + pending++ + if (pending === 1) source.pause() + void (result as Promise).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, onExit: (code: number, cliError?: CLIError) => void, workspaceId?: string, ): Promise { @@ -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)) }) } diff --git a/desktop/src/main/ipc.ts b/desktop/src/main/ipc.ts index 178e5194d..90ad010d1 100644 --- a/desktop/src/main/ipc.ts +++ b/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 +} + +function createLogSink( + getWin: () => BrowserWindow | null, + commandId: string, + appendLog?: (line: string) => boolean, + flush?: () => Promise, +): ProgressSink { + const FLUSH_MS = 64 + const MAX_BATCH = 250 + let buf: string[] = [] + let timer: ReturnType | 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 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,7 +658,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", "stop", args.workspaceId] if (args.debug) cliArgs.push("--debug") @@ -624,25 +671,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, ) @@ -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,7 +731,12 @@ 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") @@ -713,25 +744,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, ) @@ -746,7 +764,12 @@ 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") @@ -754,25 +777,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, ) diff --git a/desktop/src/main/log-store.ts b/desktop/src/main/log-store.ts index 1e018d30b..a73d44e24 100644 --- a/desktop/src/main/log-store.ts +++ b/desktop/src/main/log-store.ts @@ -1,11 +1,12 @@ import { - appendFileSync, + createWriteStream, existsSync, mkdirSync, readdirSync, readFileSync, statSync, unlinkSync, + type WriteStream, writeFileSync, } from "node:fs" import { basename, join } from "node:path" @@ -52,6 +53,8 @@ let counter = 0 export class LogStore { constructor(private logsDir: string) {} + private streams = new Map() + private workspaceLogDir(context: string, workspaceId: string): string { return join(this.logsDir, "workspaces", context, workspaceId) } @@ -67,15 +70,31 @@ export class LogStore { return filePath } - appendLog(logPath: string, line: string): void { - try { - appendFileSync(logPath, `${line}\n`) - } catch (err) { - // Defensive: the workspace dir under the desktop logs root is owned by - // this process, but a manual `rm -rf` of the logs tree (or any other - // out-of-band removal) shouldn't crash the main process. - if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err + appendLog(logPath: string, line: string): boolean { + let stream = this.streams.get(logPath) + if (!stream) { + stream = createWriteStream(logPath, { flags: "a" }) + stream.on("error", (err) => { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") { + console.error(`log write failed for ${logPath}:`, err) + } + }) + this.streams.set(logPath, stream) } + return stream.write(`${line}\n`) + } + + onDrain(logPath: string): Promise { + const stream = this.streams.get(logPath) + if (!stream) return Promise.resolve() + return new Promise((resolve) => stream.once("drain", resolve)) + } + + closeLog(logPath: string): Promise { + const stream = this.streams.get(logPath) + if (!stream) return Promise.resolve() + this.streams.delete(logPath) + return new Promise((resolve) => stream.end(resolve)) } readLogByPath(logPath: string): string { diff --git a/desktop/src/renderer/src/lib/components/log/LogTable.svelte b/desktop/src/renderer/src/lib/components/log/LogTable.svelte index 2166fe3b8..52c90ef7b 100644 --- a/desktop/src/renderer/src/lib/components/log/LogTable.svelte +++ b/desktop/src/renderer/src/lib/components/log/LogTable.svelte @@ -7,64 +7,38 @@ let { lines, class: className = "", maxHeightClass = "max-h-96", - rowHeight = 28, - overscan = 12, follow = false, + maxLines = 5000, }: { lines: string[] class?: string maxHeightClass?: string - rowHeight?: number - overscan?: number follow?: boolean + maxLines?: number } = $props() let viewport = $state(null) -let scrollTop = $state(0) -let viewportHeight = $state(0) let pinnedToBottom = $state(true) -let total = $derived(lines.length) -let totalHeight = $derived(total * rowHeight) +let hiddenCount = $derived(Math.max(0, lines.length - maxLines)) +let visible = $derived(hiddenCount > 0 ? lines.slice(-maxLines) : lines) -let startIndex = $derived( - Math.max(0, Math.floor(scrollTop / rowHeight) - overscan), -) -let endIndex = $derived( - Math.min( - total, - Math.ceil((scrollTop + viewportHeight) / rowHeight) + overscan, - ), -) - -let visible = $derived( - lines.slice(startIndex, endIndex).map((raw, i) => ({ - index: startIndex + i, - line: parseLogLine(raw), - })), -) +function levelRowClass(level: string | undefined): string { + if (level === "fatal" || level === "error") return "bg-destructive/5" + if (level === "warn") return "bg-amber-500/5" + return "" +} function onScroll() { if (!viewport) return - scrollTop = viewport.scrollTop const distanceFromBottom = viewport.scrollHeight - viewport.scrollTop - viewport.clientHeight - pinnedToBottom = distanceFromBottom <= rowHeight + pinnedToBottom = distanceFromBottom <= 8 } -$effect(() => { - if (!viewport) return - const observer = new ResizeObserver(() => { - if (viewport) viewportHeight = viewport.clientHeight - }) - observer.observe(viewport) - viewportHeight = viewport.clientHeight - return () => observer.disconnect() -}) - $effect(() => { // Follow the tail as new lines arrive, but only while the user is at the bottom. - void total + void lines.length if (follow && pinnedToBottom && viewport) { viewport.scrollTop = viewport.scrollHeight } @@ -75,50 +49,53 @@ $effect(() => { bind:this={viewport} onscroll={onScroll} role="table" - aria-rowcount={total} + aria-rowcount={lines.length + 1} class={cn("relative overflow-auto rounded-md border", maxHeightClass, className)} >
-
Time
-
Level
-
Message
+
Time
+
Level
+
Message
-
- {#each visible as row (row.index)} -
-
{row.line.time}
-
- {#if row.line.level} - - {row.line.level} - - {/if} -
-
{row.line.message}
-
- {/each} -
+ {#if hiddenCount > 0} +
+ {hiddenCount.toLocaleString()} earlier {hiddenCount === 1 ? "line" : "lines"} hidden — open the full log to see everything +
+ {/if} + + {#each visible as raw, i (i)} + {@const line = parseLogLine(raw)} +
+ {line.time} + + {#if line.level} + + {line.level} + + {/if} + + {line.message} +
+ {/each} diff --git a/desktop/src/renderer/src/lib/components/workspace/WorkspaceWizard.svelte b/desktop/src/renderer/src/lib/components/workspace/WorkspaceWizard.svelte index fd9b2f9ab..0f056e28b 100644 --- a/desktop/src/renderer/src/lib/components/workspace/WorkspaceWizard.svelte +++ b/desktop/src/renderer/src/lib/components/workspace/WorkspaceWizard.svelte @@ -426,16 +426,23 @@ onDestroy(() => { } }) +const MAX_LOG_LINES = 5000 +const TRIM_TARGET = 4000 + function flushLines() { flushHandle = null if (pendingLines.length === 0) return outputLines.push(...pendingLines) pendingLines.length = 0 + if (outputLines.length > MAX_LOG_LINES) { + outputLines.splice(0, outputLines.length - TRIM_TARGET) + } } function handleProgress(progress: CommandProgress, wsId: string | undefined) { - if (progress.message) { - pendingLines.push(progress.message) + const incoming = progress.lines ?? (progress.message ? [progress.message] : []) + if (incoming.length > 0) { + pendingLines.push(...incoming) if (flushHandle === null) { flushHandle = requestAnimationFrame(flushLines) } diff --git a/desktop/src/renderer/src/lib/types/index.ts b/desktop/src/renderer/src/lib/types/index.ts index 64e9c8b73..d700fcaa8 100644 --- a/desktop/src/renderer/src/lib/types/index.ts +++ b/desktop/src/renderer/src/lib/types/index.ts @@ -123,7 +123,8 @@ export interface Context { export interface CommandProgress { commandId: string - message: string + lines?: string[] + message?: string /** Optional log level; populated when the underlying stderr line was JSON. */ level?: "info" | "warn" | "error" done: boolean diff --git a/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte b/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte index 95f48bfc0..6d3634931 100644 --- a/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte +++ b/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte @@ -158,11 +158,17 @@ let filteredIdes = $derived( ), ) +const MAX_LOG_LINES = 5000 +const TRIM_TARGET = 4000 + function flushLines() { flushHandle = null if (pendingLines.length === 0) return outputLines.push(...pendingLines) pendingLines.length = 0 + if (outputLines.length > MAX_LOG_LINES) { + outputLines.splice(0, outputLines.length - TRIM_TARGET) + } } async function copyToClipboard(text: string) { @@ -178,8 +184,9 @@ onMount(async () => { try { unlisten = await onCommandProgress((progress) => { if (commandId && progress.commandId === commandId) { - if (progress.message) { - pendingLines.push(progress.message) + const incoming = progress.lines ?? (progress.message ? [progress.message] : []) + if (incoming.length > 0) { + pendingLines.push(...incoming) if (flushHandle === null) { flushHandle = requestAnimationFrame(flushLines) }