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 <noreply@anthropic.com>
This commit is contained in:
2026-08-16 00:02:57 +02:00
co-authored by Claude Sonnet 5
parent c3312f5dc6
commit e6008a22cd
6 changed files with 137 additions and 41 deletions
+5 -18
View File
@@ -1,16 +1,14 @@
export const dynamic = "force-dynamic"; export const dynamic = "force-dynamic";
import fs from "node:fs";
import type { Metadata } from "next"; import type { Metadata } from "next";
import { eq } from "drizzle-orm"; import { eq } from "drizzle-orm";
import { notFound } from "next/navigation"; import { notFound } from "next/navigation";
import { getDb } from "@/lib/db/client"; import { getDb } from "@/lib/db/client";
import { runs } from "@/lib/db/schema"; import { runs } from "@/lib/db/schema";
import { readLogTail } from "@/lib/runner/log-file";
import { RunTerminal } from "@/components/runs/run-terminal"; import { RunTerminal } from "@/components/runs/run-terminal";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
const MAX_INITIAL_LOG_BYTES = 200_000;
interface RunDetailPageProps { interface RunDetailPageProps {
params: Promise<{ runId: string }>; params: Promise<{ runId: string }>;
} }
@@ -27,27 +25,15 @@ export async function generateMetadata({
return { title: run ? `${run.scriptName} run` : "Run not found" }; 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) { export default async function RunDetailPage({ params }: RunDetailPageProps) {
const { runId } = await params; const { runId } = await params;
const db = getDb(); const db = getDb();
const run = db.select().from(runs).where(eq(runs.id, runId)).get(); const run = db.select().from(runs).where(eq(runs.id, runId)).get();
if (!run) notFound(); if (!run) notFound();
const initialLog = readTail(run.logFilePath); const { text: initialLog, size: initialLogBytes } = readLogTail(
run.logFilePath,
);
return ( return (
<div className="mx-auto flex max-w-3xl flex-col gap-4"> <div className="mx-auto flex max-w-3xl flex-col gap-4">
@@ -80,6 +66,7 @@ export default async function RunDetailPage({ params }: RunDetailPageProps) {
runId={run.id} runId={run.id}
initialStatus={run.status} initialStatus={run.status}
initialLog={initialLog} initialLog={initialLog}
initialLogBytes={initialLogBytes}
initialExitCode={run.exitCode} initialExitCode={run.exitCode}
/> />
</CardContent> </CardContent>
+32 -18
View File
@@ -14,6 +14,7 @@ interface RunTerminalProps {
runId: string; runId: string;
initialStatus: RunStatus; initialStatus: RunStatus;
initialLog: string; initialLog: string;
initialLogBytes: number;
initialExitCode: number | null; initialExitCode: number | null;
} }
@@ -24,31 +25,44 @@ export function RunTerminal({
runId, runId,
initialStatus, initialStatus,
initialLog, initialLog,
initialLogBytes,
initialExitCode, initialExitCode,
}: RunTerminalProps) { }: RunTerminalProps) {
const [status, setStatus] = useState<RunStatus>(initialStatus); const [status, setStatus] = useState<RunStatus>(initialStatus);
const [exitCode, setExitCode] = useState<number | null>(initialExitCode); const [exitCode, setExitCode] = useState<number | null>(initialExitCode);
const termRef = useRef<XtermViewHandle>(null); const termRef = useRef<XtermViewHandle>(null);
const { cancel } = useRunSocket(runId, (message: ServerMessage) => { const { cancel } = useRunSocket(
if (message.type === "output") { runId,
// Scripts that colorize their own output (via ANSI codes) render as-is; stderr additionally (message: ServerMessage) => {
// gets wrapped in red so failures stand out even from tools that don't colorize themselves. if (message.type === "output") {
const chunk = // Scripts that colorize their own output (via ANSI codes) render as-is; stderr additionally
message.stream === "stderr" // gets wrapped in red so failures stand out even from tools that don't colorize themselves.
? `${ANSI_RED}${message.chunk}${ANSI_RESET}` const chunk =
: message.chunk; message.stream === "stderr"
termRef.current?.write(chunk); ? `${ANSI_RED}${message.chunk}${ANSI_RESET}`
} else if (message.type === "status") { : message.chunk;
setStatus(message.status); termRef.current?.write(chunk);
setExitCode(message.exitCode ?? null); } else if (message.type === "status") {
if (message.status !== "queued" && message.status !== "running") { // The server also resends the run's current status right after subscribing, so a client
toast.info(`Run ${message.status.replace("_", " ")}`); // 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"; const isActive = status === "queued" || status === "running";
+12 -1
View File
@@ -6,10 +6,15 @@ import type { ServerMessage } from "@/lib/ws/protocol";
export function useRunSocket( export function useRunSocket(
runId: string, runId: string,
onMessage: (message: ServerMessage) => void, 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<WebSocket | null>(null); const wsRef = useRef<WebSocket | null>(null);
const [connected, setConnected] = useState(false); const [connected, setConnected] = useState(false);
const onMessageRef = useRef(onMessage); const onMessageRef = useRef(onMessage);
const afterBytesRef = useRef(afterBytes);
useEffect(() => { useEffect(() => {
onMessageRef.current = onMessage; onMessageRef.current = onMessage;
@@ -22,7 +27,13 @@ export function useRunSocket(
ws.onopen = () => { ws.onopen = () => {
setConnected(true); setConnected(true);
ws.send(JSON.stringify({ type: "subscribe", runId })); ws.send(
JSON.stringify({
type: "subscribe",
runId,
afterBytes: afterBytesRef.current,
}),
);
}; };
ws.onmessage = (event) => { ws.onmessage = (event) => {
+49
View File
@@ -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);
}
}
+1 -1
View File
@@ -1,7 +1,7 @@
import type { RunStatus } from "../db/schema"; import type { RunStatus } from "../db/schema";
export type ClientMessage = export type ClientMessage =
| { type: "subscribe"; runId: string } | { type: "subscribe"; runId: string; afterBytes?: number }
| { type: "unsubscribe"; runId: string } | { type: "unsubscribe"; runId: string }
| { type: "cancel"; runId: string }; | { type: "cancel"; runId: string };
+38 -3
View File
@@ -1,5 +1,6 @@
import type { IncomingMessage } from "node:http"; import type { IncomingMessage } from "node:http";
import { WebSocketServer, WebSocket } from "ws"; import { WebSocketServer, WebSocket } from "ws";
import { eq } from "drizzle-orm";
import { getConfig } from "../config/load"; import { getConfig } from "../config/load";
import { import {
extractCookieValue, extractCookieValue,
@@ -8,14 +9,48 @@ import {
} from "../auth/session"; } from "../auth/session";
import { hashToken, verifyTokenHash } from "../auth/tokens"; import { hashToken, verifyTokenHash } from "../auth/tokens";
import { getDb } from "../db/client"; import { getDb } from "../db/client";
import { apiTokens } from "../db/schema"; import { apiTokens, runs } from "../db/schema";
import { runEvents } from "../runner/events"; import { runEvents } from "../runner/events";
import { cancelRun } from "../runner/registry"; import { cancelRun } from "../runner/registry";
import { readLogSince } from "../runner/log-file";
import { isClientMessage, type ServerMessage } from "./protocol"; import { isClientMessage, type ServerMessage } from "./protocol";
const subscriptions = new Map<string, Set<WebSocket>>(); const subscriptions = new Map<string, Set<WebSocket>>();
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); let set = subscriptions.get(runId);
if (!set) { if (!set) {
set = new Set(); set = new Set();
@@ -80,7 +115,7 @@ export function attachWsServer(wss: WebSocketServer) {
switch (parsed.type) { switch (parsed.type) {
case "subscribe": case "subscribe":
subscribe(parsed.runId, ws); subscribe(parsed.runId, ws, parsed.afterBytes ?? 0);
break; break;
case "unsubscribe": case "unsubscribe":
unsubscribe(parsed.runId, ws); unsubscribe(parsed.runId, ws);