From e6008a22cd61fa90c4a2fa4c5892390c75431a59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20Kr=C3=BCger?= Date: Sun, 16 Aug 2026 00:02:57 +0200 Subject: [PATCH] Fix live output missing for fast-finishing scripts Scripts that finish in milliseconds (e.g. notify.js) could complete and broadcast all their output/status before a client's WS subscribe message even arrived, leaving the run page's terminal permanently blank with no way to catch up. The server now answers every subscribe with whatever log bytes were written past what the client's server-rendered page already had, plus the run's current status, before it starts streaming live broadcasts. Co-Authored-By: Claude Sonnet 5 --- app/src/app/(app)/runs/[runId]/page.tsx | 23 +++-------- app/src/components/runs/run-terminal.tsx | 50 +++++++++++++++--------- app/src/hooks/use-run-socket.ts | 13 +++++- app/src/lib/runner/log-file.ts | 49 +++++++++++++++++++++++ app/src/lib/ws/protocol.ts | 2 +- app/src/lib/ws/server.ts | 41 +++++++++++++++++-- 6 files changed, 137 insertions(+), 41 deletions(-) create mode 100644 app/src/lib/runner/log-file.ts diff --git a/app/src/app/(app)/runs/[runId]/page.tsx b/app/src/app/(app)/runs/[runId]/page.tsx index 5500455..dfd8312 100644 --- a/app/src/app/(app)/runs/[runId]/page.tsx +++ b/app/src/app/(app)/runs/[runId]/page.tsx @@ -1,16 +1,14 @@ export const dynamic = "force-dynamic"; -import fs from "node:fs"; import type { Metadata } from "next"; import { eq } from "drizzle-orm"; import { notFound } from "next/navigation"; import { getDb } from "@/lib/db/client"; import { runs } from "@/lib/db/schema"; +import { readLogTail } from "@/lib/runner/log-file"; import { RunTerminal } from "@/components/runs/run-terminal"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; -const MAX_INITIAL_LOG_BYTES = 200_000; - interface RunDetailPageProps { params: Promise<{ runId: string }>; } @@ -27,27 +25,15 @@ export async function generateMetadata({ return { title: run ? `${run.scriptName} run` : "Run not found" }; } -function readTail(logFilePath: string): string { - if (!fs.existsSync(logFilePath)) return ""; - const { size } = fs.statSync(logFilePath); - const start = Math.max(0, size - MAX_INITIAL_LOG_BYTES); - const fd = fs.openSync(logFilePath, "r"); - try { - const buffer = Buffer.alloc(size - start); - fs.readSync(fd, buffer, 0, buffer.length, start); - return (start > 0 ? "... (truncated)\n" : "") + buffer.toString("utf-8"); - } finally { - fs.closeSync(fd); - } -} - export default async function RunDetailPage({ params }: RunDetailPageProps) { const { runId } = await params; const db = getDb(); const run = db.select().from(runs).where(eq(runs.id, runId)).get(); if (!run) notFound(); - const initialLog = readTail(run.logFilePath); + const { text: initialLog, size: initialLogBytes } = readLogTail( + run.logFilePath, + ); return (
@@ -80,6 +66,7 @@ export default async function RunDetailPage({ params }: RunDetailPageProps) { runId={run.id} initialStatus={run.status} initialLog={initialLog} + initialLogBytes={initialLogBytes} initialExitCode={run.exitCode} /> diff --git a/app/src/components/runs/run-terminal.tsx b/app/src/components/runs/run-terminal.tsx index 0c11a59..54d7d7d 100644 --- a/app/src/components/runs/run-terminal.tsx +++ b/app/src/components/runs/run-terminal.tsx @@ -14,6 +14,7 @@ interface RunTerminalProps { runId: string; initialStatus: RunStatus; initialLog: string; + initialLogBytes: number; initialExitCode: number | null; } @@ -24,31 +25,44 @@ export function RunTerminal({ runId, initialStatus, initialLog, + initialLogBytes, initialExitCode, }: RunTerminalProps) { const [status, setStatus] = useState(initialStatus); const [exitCode, setExitCode] = useState(initialExitCode); const termRef = useRef(null); - const { cancel } = useRunSocket(runId, (message: ServerMessage) => { - if (message.type === "output") { - // Scripts that colorize their own output (via ANSI codes) render as-is; stderr additionally - // gets wrapped in red so failures stand out even from tools that don't colorize themselves. - const chunk = - message.stream === "stderr" - ? `${ANSI_RED}${message.chunk}${ANSI_RESET}` - : message.chunk; - termRef.current?.write(chunk); - } else if (message.type === "status") { - setStatus(message.status); - setExitCode(message.exitCode ?? null); - if (message.status !== "queued" && message.status !== "running") { - toast.info(`Run ${message.status.replace("_", " ")}`); + const { cancel } = useRunSocket( + runId, + (message: ServerMessage) => { + if (message.type === "output") { + // Scripts that colorize their own output (via ANSI codes) render as-is; stderr additionally + // gets wrapped in red so failures stand out even from tools that don't colorize themselves. + const chunk = + message.stream === "stderr" + ? `${ANSI_RED}${message.chunk}${ANSI_RESET}` + : message.chunk; + termRef.current?.write(chunk); + } else if (message.type === "status") { + // The server also resends the run's current status right after subscribing, so a client + // that connects after a fast run has already finished still gets the real status instead + // of being stuck on whatever was known at page render. Only toast when it's new info. + const changed = message.status !== status; + setStatus(message.status); + setExitCode(message.exitCode ?? null); + if ( + changed && + message.status !== "queued" && + message.status !== "running" + ) { + toast.info(`Run ${message.status.replace("_", " ")}`); + } + } else if (message.type === "error") { + toast.error(message.message); } - } else if (message.type === "error") { - toast.error(message.message); - } - }); + }, + initialLogBytes, + ); const isActive = status === "queued" || status === "running"; diff --git a/app/src/hooks/use-run-socket.ts b/app/src/hooks/use-run-socket.ts index 05da175..dc90867 100644 --- a/app/src/hooks/use-run-socket.ts +++ b/app/src/hooks/use-run-socket.ts @@ -6,10 +6,15 @@ import type { ServerMessage } from "@/lib/ws/protocol"; export function useRunSocket( runId: string, onMessage: (message: ServerMessage) => void, + /** Byte offset of the log content the caller already has (e.g. from server-rendered initial + * log) - lets the server catch the socket up on anything written/broadcast before it + * subscribed, instead of resending from the start. */ + afterBytes = 0, ) { const wsRef = useRef(null); const [connected, setConnected] = useState(false); const onMessageRef = useRef(onMessage); + const afterBytesRef = useRef(afterBytes); useEffect(() => { onMessageRef.current = onMessage; @@ -22,7 +27,13 @@ export function useRunSocket( ws.onopen = () => { setConnected(true); - ws.send(JSON.stringify({ type: "subscribe", runId })); + ws.send( + JSON.stringify({ + type: "subscribe", + runId, + afterBytes: afterBytesRef.current, + }), + ); }; ws.onmessage = (event) => { diff --git a/app/src/lib/runner/log-file.ts b/app/src/lib/runner/log-file.ts new file mode 100644 index 0000000..5ec8e6f --- /dev/null +++ b/app/src/lib/runner/log-file.ts @@ -0,0 +1,49 @@ +import fs from "node:fs"; + +const MAX_TAIL_BYTES = 200_000; + +export interface LogTail { + text: string; + /** Total file size at read time - callers use this as an offset to later fetch only what's + * been appended since (see `readLogSince`). */ + size: number; +} + +/** Reads up to the last `maxBytes` of a log file. */ +export function readLogTail( + logFilePath: string, + maxBytes = MAX_TAIL_BYTES, +): LogTail { + if (!fs.existsSync(logFilePath)) return { text: "", size: 0 }; + const { size } = fs.statSync(logFilePath); + const start = Math.max(0, size - maxBytes); + const fd = fs.openSync(logFilePath, "r"); + try { + const buffer = Buffer.alloc(size - start); + fs.readSync(fd, buffer, 0, buffer.length, start); + return { + text: (start > 0 ? "... (truncated)\n" : "") + buffer.toString("utf-8"), + size, + }; + } finally { + fs.closeSync(fd); + } +} + +/** Reads whatever has been appended to the log file after byte offset `fromByte`. Used to catch + * up a WS subscriber that connects after some output has already been written and broadcast - + * fast scripts can finish (and emit their whole output) before a client's WS subscribe message + * reaches the server, so a plain live-broadcast-only subscription would leave it stuck blank. */ +export function readLogSince(logFilePath: string, fromByte: number): string { + if (!fs.existsSync(logFilePath)) return ""; + const { size } = fs.statSync(logFilePath); + if (size <= fromByte) return ""; + const fd = fs.openSync(logFilePath, "r"); + try { + const buffer = Buffer.alloc(size - fromByte); + fs.readSync(fd, buffer, 0, buffer.length, fromByte); + return buffer.toString("utf-8"); + } finally { + fs.closeSync(fd); + } +} diff --git a/app/src/lib/ws/protocol.ts b/app/src/lib/ws/protocol.ts index 26f36e0..abeda0c 100644 --- a/app/src/lib/ws/protocol.ts +++ b/app/src/lib/ws/protocol.ts @@ -1,7 +1,7 @@ import type { RunStatus } from "../db/schema"; export type ClientMessage = - | { type: "subscribe"; runId: string } + | { type: "subscribe"; runId: string; afterBytes?: number } | { type: "unsubscribe"; runId: string } | { type: "cancel"; runId: string }; diff --git a/app/src/lib/ws/server.ts b/app/src/lib/ws/server.ts index 8870c2c..9d7360f 100644 --- a/app/src/lib/ws/server.ts +++ b/app/src/lib/ws/server.ts @@ -1,5 +1,6 @@ import type { IncomingMessage } from "node:http"; import { WebSocketServer, WebSocket } from "ws"; +import { eq } from "drizzle-orm"; import { getConfig } from "../config/load"; import { extractCookieValue, @@ -8,14 +9,48 @@ import { } from "../auth/session"; import { hashToken, verifyTokenHash } from "../auth/tokens"; import { getDb } from "../db/client"; -import { apiTokens } from "../db/schema"; +import { apiTokens, runs } from "../db/schema"; import { runEvents } from "../runner/events"; import { cancelRun } from "../runner/registry"; +import { readLogSince } from "../runner/log-file"; import { isClientMessage, type ServerMessage } from "./protocol"; const subscriptions = new Map>(); -function subscribe(runId: string, ws: WebSocket) { +/** A run can start and finish (emitting all its output+status over `runEvents`) before a + * client's `subscribe` message even reaches the server - fast scripts routinely beat the + * WS handshake + subscribe round trip. So every subscribe is answered with a catch-up: whatever + * log bytes exist past what the client already has (`afterBytes`, from its server-rendered + * initial log), plus the run's current status - before the socket starts receiving live + * broadcasts. Read-then-register (rather than register-then-read) trades a theoretical + * microsecond gap for guaranteeing no duplicated output. */ +function subscribe(runId: string, ws: WebSocket, afterBytes: number) { + const run = getDb().select().from(runs).where(eq(runs.id, runId)).get(); + if (run) { + const catchUp = readLogSince(run.logFilePath, afterBytes); + if (catchUp) { + ws.send( + JSON.stringify({ + type: "output", + runId, + stream: "stdout", + chunk: catchUp, + seq: -1, + ts: Date.now(), + } satisfies ServerMessage), + ); + } + ws.send( + JSON.stringify({ + type: "status", + runId, + status: run.status, + exitCode: run.exitCode, + ts: Date.now(), + } satisfies ServerMessage), + ); + } + let set = subscriptions.get(runId); if (!set) { set = new Set(); @@ -80,7 +115,7 @@ export function attachWsServer(wss: WebSocketServer) { switch (parsed.type) { case "subscribe": - subscribe(parsed.runId, ws); + subscribe(parsed.runId, ws, parsed.afterBytes ?? 0); break; case "unsubscribe": unsubscribe(parsed.runId, ws);