[supervisor] Make sure we read the last chunk of message when task is closed (#20062)

* [supervisor] Make sure we read the last chunk of message when task is closed

* [server, dashboard] Prebuild logs: stream as UInt8Array instead of string, and make sure we send the last chunk before closing the response

* Make log buffer more efficient

Merging done with `TypedArray.prototype.set()` as suggested here:
https://stackoverflow.com/a/73074813/10199319

* [dashboard] PrebuildTaskTab: use hash(err.message) as toastId

* [server] prebuild logs: Only call res.end after a timeout, so clients can continue reading from the stream

* [dashboard] prebuild logs emitter: If we receive code "200", forward prefixChunk, and stop streaming

* fixup! [dashboard] PrebuildTaskTab: use hash(err.message) as toastId

---------

Co-authored-by: Filip Troníček <[email protected]>

Author: geropl

components/dashboard/src/components/PrebuildLogs.tsx

@@ -103,7 +103,7 @@ export default function PrebuildLogs(props: PrebuildLogsProps) {
                         if (!content) {
                             return;
                         }
-                        logsEmitter.emit("logs", content.text);
+                        logsEmitter.emit("logs", content.data);
                     },
                 }),
             );
@@ -173,7 +173,12 @@ export default function PrebuildLogs(props: PrebuildLogsProps) {
         <div className="rounded-xl overflow-hidden bg-gray-100 dark:bg-gray-800 flex flex-col mb-8">
             <div className="h-96 flex">
                 <Suspense fallback={<div />}>
-                    <WorkspaceLogs classes="h-full w-full" logsEmitter={logsEmitter} errorMessage={error?.message} />
+                    <WorkspaceLogs
+                        taskId="undefined"
+                        classes="h-full w-full"
+                        logsEmitter={logsEmitter}
+                        errorMessage={error?.message}
+                    />
                 </Suspense>
             </div>
             <div className="w-full bottom-0 h-20 px-6 bg-gray-50 dark:bg-gray-800 border-t border-gray-200 dark:border-gray-600 flex flex-row items-center space-x-2">

components/dashboard/src/components/WorkspaceLogs.tsx

@@ -26,6 +26,7 @@ const lightTheme: ITheme = {
 };
 
 export interface Props {
+    taskId: string;
     logsEmitter: EventEmitter;
     errorMessage?: string;
     classes?: string;
@@ -56,14 +57,14 @@ export default function WorkspaceLogs({ logsEmitter, errorMessage, classes, xter
         terminal.loadAddon(fitAddon);
         terminal.open(xTermParentRef.current);
 
-        let logBuffer = "";
+        let logBuffer = new Uint8Array();
         let isWriting = false;
 
         const processNextLog = () => {
             if (isWriting || logBuffer.length === 0) return;
 
-            const logs = logBuffer.slice(0, MAX_CHUNK_SIZE);
-            logBuffer = logBuffer.slice(logs.length);
+            const logs = logBuffer.subarray(0, MAX_CHUNK_SIZE);
+            logBuffer = logBuffer.subarray(logs.length);
             if (logs) {
                 isWriting = true;
                 terminal.write(logs, () => {
@@ -73,16 +74,20 @@ export default function WorkspaceLogs({ logsEmitter, errorMessage, classes, xter
             }
         };
 
-        const logListener = (logs: string) => {
+        const logListener = (logs: Uint8Array) => {
             if (!logs) return;
 
-            logBuffer += logs;
+            const newBuffer = new Uint8Array(logBuffer.length + logs.length);
+            newBuffer.set(logBuffer);
+            newBuffer.set(logs, logBuffer.length);
+            logBuffer = newBuffer;
+
             processNextLog();
         };
 
         const resetListener = () => {
             terminal.clear();
-            logBuffer = "";
+            logBuffer = new Uint8Array();
             isWriting = false;
         };
 

components/dashboard/src/data/prebuilds/prebuild-logs-emitter.ts

@@ -80,16 +80,14 @@ export function usePrebuildLogsEmitter(prebuild: PlainMessage<Prebuild>, taskId:
         const disposables = new DisposableCollection();
         disposables.push(
             streamPrebuildLogs(
+                taskId,
                 task.logUrl,
-                (msg) => {
-                    const error = matchPrebuildError(msg);
-                    if (!error) {
-                        emitter.emit("logs", msg);
-                    } else {
-                        emitter.emit("logs-error", error);
-                    }
+                (chunk) => {
+                    emitter.emit("logs", chunk);
+                },
+                (err) => {
+                    emitter.emit("logs-error", err);
                 },
-                async () => false,
                 () => {
                     emitter.markReachedEnd();
                 },
@@ -110,9 +108,10 @@ export function usePrebuildLogsEmitter(prebuild: PlainMessage<Prebuild>, taskId:
 }
 
 function streamPrebuildLogs(
+    taskId: string,
     streamUrl: string,
-    onLog: (chunk: string) => void,
-    checkIsDone: () => Promise<boolean>,
+    onLog: (chunk: Uint8Array) => void,
+    onError: (err: Error) => void,
     onEnd?: () => void,
 ): DisposableCollection {
     const disposables = new DisposableCollection();
@@ -127,10 +126,6 @@ function streamPrebuildLogs(
     let delayInSeconds = initialDelaySeconds;
 
     const startWatchingLogs = async () => {
-        if (await checkIsDone()) {
-            return;
-        }
-
         const retryBackoff = async (reason: string, err?: Error) => {
             delayInSeconds = Math.min(delayInSeconds * backoffFactor, maxBackoffSeconds);
 
@@ -145,25 +140,21 @@ function streamPrebuildLogs(
         };
 
         let response: Response | undefined = undefined;
-        let abortController: AbortController | undefined = undefined;
         let reader: ReadableStreamDefaultReader<Uint8Array> | undefined = undefined;
         try {
-            abortController = new AbortController();
             disposables.push({
                 dispose: async () => {
-                    abortController?.abort();
                     await reader?.cancel();
                 },
             });
             console.debug("fetching from streamUrl: " + streamUrl);
             response = await fetch(streamUrl, {
                 method: "GET",
-                cache: "reload",
+                cache: "no-store", // we don't want the browser to a) look at the cache, or b) update the cache (which would interrupt any running fetches to that resource!)
                 credentials: "include",
                 headers: {
                     TE: "trailers", // necessary to receive stream status code
                 },
-                signal: abortController.signal,
                 redirect: "follow",
             });
             reader = response.body?.getReader();
@@ -174,41 +165,51 @@ function streamPrebuildLogs(
 
             const decoder = new TextDecoder("utf-8");
             let chunk = await reader.read();
+            let received200 = false;
             while (!chunk.done) {
                 if (disposables.disposed) {
                     // stop reading when disposed
                     return;
                 }
-                const msg = decoder.decode(chunk.value, { stream: true });
 
                 // In an ideal world, we'd use res.addTrailers()/response.trailer here. But despite being introduced with HTTP/1.1 in 1999, trailers are not supported by popular proxies (nginx, for example).
                 // So we resort to this hand-written solution:
+                const msg = decoder.decode(chunk.value, { stream: true });
                 const matches = msg.match(HEADLESS_LOG_STREAM_STATUS_CODE_REGEX);
                 const prebuildMatches = matchPrebuildError(msg);
                 if (matches) {
                     if (matches.length < 2) {
                         console.debug("error parsing log stream status code. msg: " + msg);
                     } else {
+                        const prefix = msg.substring(0, matches.index);
+                        if (prefix) {
+                            const prefixChunk = new TextEncoder().encode(prefix);
+                            onLog(prefixChunk);
+                        }
                         const code = parseStatusCode(matches[1]);
                         if (code !== 200) {
                             throw new StreamError(code);
                         }
+                        if (code === 200) {
+                            received200 = true;
+                            break;
+                        }
+                    }
+                } else if (prebuildMatches) {
+                    if (prebuildMatches.code === ErrorCodes.HEADLESS_LOG_NOT_YET_AVAILABLE) {
+                        // reset backoff because this error is expected
+                        delayInSeconds = initialDelaySeconds;
+                        throw prebuildMatches;
                     }
-                } else if (prebuildMatches && prebuildMatches.code === ErrorCodes.HEADLESS_LOG_NOT_YET_AVAILABLE) {
-                    // reset backoff because this error is expected
-                    delayInSeconds = initialDelaySeconds;
-                    throw prebuildMatches;
+                    onError(prebuildMatches);
                 } else {
-                    onLog(msg);
+                    onLog(chunk.value);
                 }
 
                 chunk = await reader.read();
             }
+            console.info("[stream] end of stream", { received200 });
             reader.cancel();
-
-            if (await checkIsDone()) {
-                return;
-            }
         } catch (err) {
             if (err instanceof DOMException && err.name === "AbortError") {
                 console.debug("stopped watching headless logs, not retrying: method got disposed of");

components/dashboard/src/prebuilds/detail/PrebuildTaskTab.tsx

@@ -30,35 +30,21 @@ export const PrebuildTaskTab = memo(({ taskId, prebuild }: Props) => {
     const history = useHistory();
 
     useEffect(() => {
-        const errorListener = (err: Error) => {
-            if (err?.name === "AbortError") {
-                return;
-            }
-            if (err instanceof ApplicationError && err.code === ErrorCodes.NOT_FOUND) {
-                // We don't want to show a toast for this error, we handle it in the UI
-                return;
-            }
-            if (err?.message) {
-                toast("Fetching logs failed: " + err.message);
-            }
-        };
-
-        const logErrorListener = (err: ApplicationError) => {
+        const logErrorListener = async (err: ApplicationError) => {
             if (err.code === ErrorCodes.NOT_FOUND) {
                 setError(err);
                 return;
             }
 
-            const toastId = crypto.randomUUID();
+            const digest = await crypto.subtle.digest("sha256", new TextEncoder().encode(err.message + ":" + err.code));
+            const toastId = new TextDecoder().decode(digest);
             toast("Fetching logs failed: " + err.message, { autoHide: false, id: toastId });
             setActiveToasts((prev) => new Set(prev).add(toastId));
         };
 
-        logEmitter.on("error", errorListener);
         logEmitter.on("logs-error", logErrorListener);
 
         return () => {
-            logEmitter.removeListener("error", errorListener);
             logEmitter.removeListener("logs-error", logErrorListener);
             setError(undefined);
         };
@@ -109,6 +95,7 @@ export const PrebuildTaskTab = memo(({ taskId, prebuild }: Props) => {
                     key={prebuild.id + taskId}
                     classes="w-full h-full"
                     xtermClasses="absolute top-0 left-0 bottom-0 right-0 ml-6 my-0 mt-4"
+                    taskId={taskId}
                     logsEmitter={logEmitter}
                 />
             </Suspense>

components/dashboard/src/start/StartWorkspace.tsx

@@ -794,7 +794,7 @@ function ImageBuildView(props: ImageBuildViewProps) {
                 if (!content) {
                     return;
                 }
-                logsEmitter.emit("logs", content.text);
+                logsEmitter.emit("logs", content.data);
             },
         });
 
@@ -807,7 +807,7 @@ function ImageBuildView(props: ImageBuildViewProps) {
     return (
         <StartPage title="Building Image" phase={props.phase} workspaceId={props.workspaceId}>
             <Suspense fallback={<div />}>
-                <WorkspaceLogs logsEmitter={logsEmitter} errorMessage={props.error?.message} />
+                <WorkspaceLogs taskId="image-build" logsEmitter={logsEmitter} errorMessage={props.error?.message} />
             </Suspense>
             {!!props.onStartWithDefaultImage && (
                 <>

components/gitpod-protocol/src/protocol.ts

@@ -974,9 +974,7 @@ export namespace WorkspaceImageBuild {
         maxSteps?: number;
     }
     export interface LogContent {
-        text: string;
-        upToLine?: number;
-        isDiff?: boolean;
+        data: Uint8Array;
     }
     export type LogCallback = (info: StateInfo, content: LogContent | undefined) => void;
     export namespace LogLine {

components/server/src/prebuilds/prebuild-manager.ts

@@ -654,7 +654,7 @@ export class PrebuildManager {
         userId: string,
         prebuildId: string,
         taskId: string,
-        onLog: (message: string) => Promise<void>,
+        onLog: (chunk: Uint8Array) => Promise<void>,
     ): Promise<{ taskUrl: string } | undefined> {
         const prebuild = await this.getPrebuild({}, userId, prebuildId);
         const organizationId = prebuild?.info.teamId;