backend/project: stream exec typing, debugging

Author: haraldschilly

src/packages/backend/exec-stream.ts

@@ -3,14 +3,21 @@
  * Core streaming logic that can be used by different services.
  */
 
-import { executeCode, asyncCache } from "./execute-code";
-import { abspath } from "./misc_node";
-import getLogger from "./logger";
+import { unreachable } from "@cocalc/util/misc";
 import {
   ExecuteCodeOutputAsync,
   ExecuteCodeStats,
   ExecuteCodeStreamEvent,
 } from "@cocalc/util/types/execute-code";
+import { asyncCache, executeCode } from "./execute-code";
+import getLogger from "./logger";
+import { abspath } from "./misc_node";
+
+export type StreamEvent = {
+  type?: "job" | ExecuteCodeStreamEvent["type"];
+  data?: ExecuteCodeStreamEvent["data"];
+  error?: string;
+};
 
 const logger = getLogger("backend:exec-stream");
 
@@ -32,11 +39,11 @@ export interface ExecuteStreamOptions {
   verbose?: boolean;
   project_id?: string;
   debug?: string;
-  stream: (event: any) => void;
+  stream: (event: StreamEvent | null) => void;
 }
 
 export async function executeStream(options: ExecuteStreamOptions) {
-  const { stream, debug, project_id: reqProjectId, ...opts } = options;
+  const { stream, debug, project_id, ...opts } = options;
 
   // Log debug message for debugging purposes
   if (debug) {
@@ -47,7 +54,7 @@ export async function executeStream(options: ExecuteStreamOptions) {
     let done = false;
     let stats: ExecuteCodeStats = [];
 
-    // Create streaming callback
+    // Create streaming callback, passed into execute-code::executeCode call
     const streamCB = (event: ExecuteCodeStreamEvent) => {
       if (done) {
         logger.debug(
@@ -65,12 +72,14 @@ export async function executeStream(options: ExecuteStreamOptions) {
             data: event.data,
           });
           break;
+
         case "stderr":
           stream({
             type: "stderr",
             data: event.data,
           });
           break;
+
         case "stats":
           // Stats are accumulated in the stats array for the final result
           if (
@@ -89,6 +98,7 @@ export async function executeStream(options: ExecuteStreamOptions) {
             });
           }
           break;
+
         case "done":
           logger.debug(`executeStream: processing done event`);
           const result = event.data as ExecuteCodeOutputAsync;
@@ -101,12 +111,16 @@ export async function executeStream(options: ExecuteStreamOptions) {
           done = true;
           stream(null); // End the stream
           break;
+
         case "error":
           logger.debug(`executeStream: processing error event`);
           stream({ error: event.data as string });
           done = true;
           stream(null);
           break;
+
+        default:
+          unreachable(event.type);
       }
     };
 
@@ -161,8 +175,8 @@ export async function executeStream(options: ExecuteStreamOptions) {
 
     // Stats monitoring is now handled by execute-code.ts via streamCB
     // No need for duplicate monitoring here
-   } catch (err) {
-     stream({ error: `${err}` });
-     stream(null); // End the stream
-   }
+  } catch (err) {
+    stream({ error: `${err}` });
+    stream(null); // End the stream
+  }
 }

src/packages/backend/execute-code.ts

@@ -23,7 +23,13 @@ import { envToInt } from "@cocalc/backend/misc/env-to-number";
 import { aggregate } from "@cocalc/util/aggregate";
 import { callback_opts } from "@cocalc/util/async-utils";
 import { PROJECT_EXEC_DEFAULT_TIMEOUT_S } from "@cocalc/util/consts/project";
-import { to_json, trunc, uuid, walltime } from "@cocalc/util/misc";
+import {
+  to_json,
+  trunc,
+  trunc_middle,
+  uuid,
+  walltime,
+} from "@cocalc/util/misc";
 import {
   ExecuteCodeOutputAsync,
   ExecuteCodeOutputBlocking,
@@ -239,7 +245,15 @@ async function executeCodeNoAggregate(
       const child = doSpawn(
         { ...opts, origCommand, job_id, job_config },
         async (err, result) => {
-          log.debug("async/doSpawn returned", { err, result });
+          log.debug("async/doSpawn returned", {
+            err,
+            result: {
+              type: result?.type,
+              stdout: trunc_middle(result?.stdout),
+              stderr: trunc_middle(result?.stderr),
+              exit_code: result?.exit_code,
+            },
+          });
           try {
             const info: Omit<
               ExecuteCodeOutputAsync,
@@ -428,6 +442,45 @@ function doSpawn(
     log.debug("listening for stdout, stderr and exit_code...");
   }
 
+  // Batching mechanism for streaming to reduce message frequency -- otherwise there could be 100msg/s to process
+  let streamBatchTimer: NodeJS.Timeout | undefined;
+  const streamBuffer = { stdout: "", stderr: "" };
+
+  // Send batched stream data
+  const sendBatchedStream = () => {
+    if (!opts.streamCB) return;
+
+    const hasStdout = streamBuffer.stdout.length > 0;
+    const hasStderr = streamBuffer.stderr.length > 0;
+
+    if (hasStdout || hasStderr) {
+      // Send stdout if available
+      if (hasStdout) {
+        opts.streamCB({ type: "stdout", data: streamBuffer.stdout });
+        streamBuffer.stdout = "";
+      }
+      // Send stderr if available
+      if (hasStderr) {
+        opts.streamCB({ type: "stderr", data: streamBuffer.stderr });
+        streamBuffer.stderr = "";
+      }
+    }
+  };
+
+  // Flush any remaining buffered data and cleanup
+  const flushStreamBuffer = () => {
+    if (streamBatchTimer) {
+      clearInterval(streamBatchTimer);
+      streamBatchTimer = undefined;
+    }
+    sendBatchedStream();
+  };
+
+  // Start batch timer if streaming is enabled, every 100ms
+  if (opts.streamCB) {
+    streamBatchTimer = setInterval(sendBatchedStream, 100);
+  }
+
   function update_async(
     job_id: string | undefined,
     aspect: "stdout" | "stderr" | "pid",
@@ -456,16 +509,16 @@ function doSpawn(
       if (stdout.length < opts.max_output) {
         const newData = data.slice(0, opts.max_output - stdout.length);
         stdout += newData;
-        // Stream only the new portion
+        // Buffer the new portion for batched streaming
         if (opts.streamCB && stdout.length > prevLength) {
-          opts.streamCB({ type: "stdout", data: newData });
+          streamBuffer.stdout += newData;
         }
       }
     } else {
       stdout += data;
-      // Stream the new data
+      // Buffer the new data for batched streaming
       if (opts.streamCB) {
-        opts.streamCB({ type: "stdout", data });
+        streamBuffer.stdout += data;
       }
     }
     update_async(opts.job_id, "stdout", stdout);
@@ -478,16 +531,16 @@ function doSpawn(
       if (stderr.length < opts.max_output) {
         const newData = data.slice(0, opts.max_output - stderr.length);
         stderr += newData;
-        // Stream only the new portion
+        // Buffer the new portion for batched streaming
         if (opts.streamCB && stderr.length > prevLength) {
-          opts.streamCB({ type: "stderr", data: newData });
+          streamBuffer.stderr += newData;
         }
       }
     } else {
       stderr += data;
-      // Stream the new data
+      // Buffer the new data for batched streaming
       if (opts.streamCB) {
-        opts.streamCB({ type: "stderr", data });
+        streamBuffer.stderr += data;
       }
     }
     update_async(opts.job_id, "stderr", stderr);
@@ -524,8 +577,9 @@ function doSpawn(
     stderr += to_json(err);
     // a fundamental issue, we were not running some code
     ran_code = false;
-    // For streaming, send error event immediately
+    // For streaming, flush buffer and send error event
     if (opts.streamCB && opts.async_call && opts.job_id) {
+      flushStreamBuffer(); // Flush any buffered data first
       const errorResult: ExecuteCodeOutputAsync = {
         type: "async",
         job_id: opts.job_id,
@@ -579,6 +633,11 @@ function doSpawn(
     // finally finish up – this will also terminate the monitor
     callback_done = true;
 
+    // Flush any remaining buffered stream data before finishing
+    if (opts.streamCB) {
+      flushStreamBuffer();
+    }
+
     if (timer != null) {
       clearTimeout(timer);
       timer = undefined;

src/packages/frontend/frame-editors/latex-editor/build-command.tsx

@@ -7,18 +7,19 @@
 Customization and selection of the build command.
 */
 
-import { Alert, Select, Form, Input } from "antd";
-import { List } from "immutable";
 import { SaveOutlined } from "@ant-design/icons";
+import { Alert, Form, Input, Select } from "antd";
+import { List } from "immutable";
+
 import { Button } from "@cocalc/frontend/antd-bootstrap";
 import { React } from "@cocalc/frontend/app-framework";
 import { Icon, Loading, Paragraph } from "@cocalc/frontend/components";
 import { split } from "@cocalc/util/misc";
 import { Actions } from "./actions";
 import {
-  build_command as latexmk_build_command,
   Engine,
   ENGINES,
+  build_command as latexmk_build_command,
 } from "./latexmk";
 
 // cmd could be undefined -- https://github.com/sagemathinc/cocalc/issues/3290
@@ -185,7 +186,7 @@ export const BuildCommand: React.FC<Props> = React.memo((props: Props) => {
     return (
       <Button
         disabled={!dirty}
-        bsSize={"small"}
+        bsSize={"xsmall"}
         bsStyle={dirty ? "success" : undefined}
         title={"Saves the modified command (or just hit the 'Return' key)"}
         onClick={() => handle_build_change()}

src/packages/project/exec-stream.ts

@@ -3,14 +3,17 @@ Project-side exec-stream service that handles streaming execution requests.
 Similar to how the project API service works, but specifically for streaming exec.
 */
 
-import { executeStream as backendExecuteStream } from "@cocalc/backend/exec-stream";
+import { executeStream, StreamEvent  } from "@cocalc/backend/exec-stream";
+import { Message, Subscription } from "@cocalc/conat/core/client";
 import { projectSubject } from "@cocalc/conat/names";
 import { connectToConat } from "@cocalc/project/conat/connection";
 import { project_id } from "@cocalc/project/data";
 import { getLogger } from "@cocalc/project/logger";
 
 const logger = getLogger("project:exec-stream");
 
+
+
 export function init() {
   serve();
 }
@@ -24,31 +27,26 @@ async function serve() {
     service: "exec-stream",
   });
 
-  logger.debug(`serve: creating exec-stream service for project ${project_id}`);
+  logger.debug(
+    `serve: creating exec-stream service for project ${project_id} and subject='${subject}'`,
+  );
   const api = await cn.subscribe(subject, { queue: "q" });
-  logger.debug(`serve: subscribed to subject='${subject}'`);
   await listen(api, subject);
 }
 
-async function listen(api, _subject) {
+async function listen(api: Subscription, subject: string) {
+  logger.debug(`Listening on subject='${subject}'`);
+
   for await (const mesg of api) {
     handleMessage(mesg);
   }
 }
 
-async function handleMessage(mesg) {
+async function handleMessage(mesg: Message) {
   const options = mesg.data;
 
   let seq = 0;
-  const respond = ({
-    type,
-    data,
-    error,
-  }: {
-    type?: string;
-    data?: any;
-    error?: string;
-  }) => {
+  const respond = ({ type, data, error }: StreamEvent) => {
     mesg.respondSync({ type, data, error, seq });
     seq += 1;
   };
@@ -61,7 +59,7 @@ async function handleMessage(mesg) {
     mesg.respondSync(null);
   };
 
-  const stream = (event?) => {
+  const stream = (event: StreamEvent) => {
     if (done) return;
     if (event != null) {
       respond(event);
@@ -80,7 +78,7 @@ async function handleMessage(mesg) {
     const { stream: _, project_id: reqProjectId, ...opts } = options;
 
     // Call the backend executeStream function
-    await backendExecuteStream({
+    await executeStream({
       ...opts,
       project_id: reqProjectId,
       stream,