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);