Merge remote-tracking branch 'origin/master' into github-tests-report-status

Author: haraldschilly

src/.claude/settings.local.json

@@ -4,7 +4,18 @@
       "Bash(pnpm tsc:*)",
       "Bash(pnpm build:*)",
       "Bash(git add:*)",
-      "Bash(git commit:*)"
+      "Bash(git commit:*)",
+      "Bash(node:*)",
+      "Bash(grep:*)",
+      "Bash(find:*)",
+      "WebFetch(domain:github.com)",
+      "WebFetch(domain:cocalc.com)",
+      "WebFetch(domain:doc.cocalc.com)",
+      "Bash(npm show:*)",
+      "Bash(prettier -w:*)",
+      "Bash(npx tsc:*)",
+      "Bash(gh pr view:*)",
+      "Bash(gh:*)"
     ],
     "deny": []
   }

src/packages/conat/core/client.ts

@@ -510,7 +510,10 @@ export class Client extends EventEmitter {
       reconnection: true,
     });
 
-    this.conn.on("info", (info) => {
+    this.conn.on("info", (info, ack) => {
+      if (typeof ack == "function") {
+        ack();
+      }
       const firstTime = this.info == null;
       this.info = info;
       this.emit("info", info);

src/packages/conat/core/server.ts

@@ -53,7 +53,7 @@ import {
 import { Patterns } from "./patterns";
 import { is_array } from "@cocalc/util/misc";
 import { UsageMonitor } from "@cocalc/conat/monitor/usage";
-import { once } from "@cocalc/util/async-utils";
+import { once, until } from "@cocalc/util/async-utils";
 import {
   clusterLink,
   type ClusterLink,
@@ -940,7 +940,7 @@ export class ConatServer extends EventEmitter {
       this.subscriptions[id] = new Set<string>();
     }
 
-    socket.emit("info", { ...this.info(), user });
+    this.sendInfo(socket, user);
 
     socket.on("stats", ({ recv0 }) => {
       const s = this.stats[socket.id];
@@ -1086,6 +1086,38 @@ export class ConatServer extends EventEmitter {
     });
   };
 
+  sendInfo = async (socket, user) => {
+    // we send info with an ack because I think sometimes the initial "info"
+    // message gets dropped, leaving a broken hanging connection that never
+    // does anything (until the user explicitly refreshes their browser).
+    // I did see what is probably this in production frequently.
+    try {
+      await until(
+        async () => {
+          if (!socket.conn?.readyState.startsWith("o")) {
+            // logger.debug(`failed to send "info" message to ${socket.id}`);
+            // readyState not defined or not opened or opening, so connection must
+            // have been closed before success.
+            return true;
+          }
+          try {
+            await socket
+              .timeout(7500)
+              .emitWithAck("info", { ...this.info(), user });
+            return true;
+          } catch (err) {
+            // logger.debug(`error sending "info" message to ${socket.id}`, err);
+            return false;
+          }
+        },
+        { min: 5000, max: 30000, timeout: 120_000 },
+      );
+    } catch {
+      // never ack'd "info" after a few minutes -- could just be an old client,
+      // so don't do anything at this point.
+    }
+  };
+
   address = () => getServerAddress(this.options);
 
   // create new client in the same process connected to this server.
@@ -1349,8 +1381,8 @@ export class ConatServer extends EventEmitter {
     const usage = async () => {
       return { [this.id]: this.usage.stats() };
     };
-    // user has to explicitly refresh there browser after
-    // being disconnected this way
+    // user has to explicitly refresh their browser after
+    // being disconnected this way:
     const disconnect = async (ids: string | string[]) => {
       if (typeof ids == "string") {
         ids = [ids];

src/packages/conat/files/read.ts

@@ -41,6 +41,10 @@ for await (const chunk of await a.readFile({project_id:'00847397-d6a8-4cb0-96a8-
 import { conat } from "@cocalc/conat/client";
 import { projectSubject } from "@cocalc/conat/names";
 import { type Subscription } from "@cocalc/conat/core/client";
+import { delay } from "awaiting";
+import { getLogger } from "@cocalc/conat/client";
+
+const logger = getLogger("conat:files:read");
 
 let subs: { [name: string]: Subscription } = {};
 export async function close({ project_id, compute_server_id, name = "" }) {
@@ -72,6 +76,7 @@ export async function createServer({
     compute_server_id,
     name,
   });
+  logger.debug("createServer", { subject });
   const cn = await conat();
   const sub = await cn.subscribe(subject);
   subs[subject] = sub;
@@ -89,36 +94,65 @@ async function listen({ sub, createReadStream }) {
 }
 
 async function handleMessage(mesg, createReadStream) {
+  logger.debug("handleMessage", mesg.subject);
   try {
     await sendData(mesg, createReadStream);
     await mesg.respond(null, { headers: { done: true } });
   } catch (err) {
-    // console.log("sending ERROR", err);
+    logger.debug("handleMessage: ERROR", err);
     mesg.respondSync(null, { headers: { error: `${err}` } });
   }
 }
 
-const MAX_CHUNK_SIZE = 16384 * 16 * 3;
+// 4MB -- chunks may be slightly bigger
+const CHUNK_SIZE = 4194304;
+const CHUNK_INTERVAL = 250;
 
 function getSeqHeader(seq) {
   return { headers: { seq } };
 }
 
 async function sendData(mesg, createReadStream) {
   const { path } = mesg.data;
+  logger.debug("sendData: starting", { path });
   let seq = 0;
+  const chunks: Buffer[] = [];
+  let size = 0;
+  const sendChunks = async () => {
+    // Not only is waiting for the response useful to make sure somebody is listening,
+    // we also use await here partly to space out the messages to avoid saturing
+    // the websocket connection, since doing so would break everything
+    // (heartbeats, etc.) and disconnect us, when transfering a large file.
+    seq += 1;
+    logger.debug("sendData: sending", { path, seq });
+    const data = Buffer.concat(chunks);
+    const { count } = await mesg.respond(data, getSeqHeader(seq));
+    if (count == 0) {
+      logger.debug("sendData: nobody is listening");
+      // nobody is listening so don't waste effort sending...
+      throw Error("receiver is gone");
+    }
+    size = 0;
+    chunks.length = 0;
+    // Delay a little just to give other messages a chance, so we don't get disconnected
+    // e.g., due to lack of heartbeats. Also, this reduces the load on conat-router.
+    await delay(CHUNK_INTERVAL);
+  };
+
   for await (let chunk of createReadStream(path, {
-    highWaterMark: 16384 * 16 * 3,
+    highWaterMark: CHUNK_SIZE,
   })) {
-    // console.log("sending ", { seq, bytes: chunk.length });
-    // We must break the chunk into smaller messages or it will
-    // get bounced by conat...
-    while (chunk.length > 0) {
-      seq += 1;
-      mesg.respondSync(chunk.slice(0, MAX_CHUNK_SIZE), getSeqHeader(seq));
-      chunk = chunk.slice(MAX_CHUNK_SIZE);
+    chunks.push(chunk);
+    size += chunk.length;
+    if (size >= CHUNK_SIZE) {
+      // send it
+      await sendChunks();
     }
   }
+  if (size > 0) {
+    await sendChunks();
+  }
+  logger.debug("sendData: done", { path }, "successfully sent ", seq, "chunks");
 }
 
 export interface ReadFileOptions {
@@ -136,6 +170,7 @@ export async function* readFile({
   name = "",
   maxWait = 1000 * 60 * 10, // 10 minutes
 }: ReadFileOptions) {
+  logger.debug("readFile", { project_id, compute_server_id, path });
   const cn = await conat();
   const subject = getSubject({
     project_id,

src/packages/conat/files/write.ts

@@ -73,6 +73,8 @@ import {
 import { projectSubject } from "@cocalc/conat/names";
 import { type Subscription } from "@cocalc/conat/core/client";
 import { type Readable } from "node:stream";
+import { getLogger } from "@cocalc/conat/client";
+const logger = getLogger("conat:files:write");
 
 function getWriteSubject({ project_id, compute_server_id }) {
   return projectSubject({
@@ -106,6 +108,7 @@ export async function createServer({
   createWriteStream: (path: string) => any;
 }) {
   const subject = getWriteSubject({ project_id, compute_server_id });
+  logger.debug("createServer", { subject });
   let sub = subs[subject];
   if (sub != null) {
     return;
@@ -141,6 +144,7 @@ async function handleMessage({
   let writeStream: null | Awaited<ReturnType<typeof createWriteStream>> = null;
   try {
     const { path, name, maxWait } = mesg.data;
+    logger.debug("handleMessage", { path, name, maxWait });
     writeStream = await createWriteStream(path);
     // console.log("created writeStream");
     writeStream.on("error", (err) => {
@@ -166,12 +170,15 @@ async function handleMessage({
       writeStream.write(chunk);
       chunks += 1;
       bytes += chunk.length;
+      logger.debug("handleMessage -- wrote", { path, name, bytes });
       // console.log("wrote ", bytes);
     }
     writeStream.end();
     writeStream.emit("rename");
     mesg.respondSync({ status: "success", bytes, chunks });
+    logger.debug("handleMessage -- SUCCESS", { path, name });
   } catch (err) {
+    logger.debug("handleMessage: ERROR", err);
     if (!error) {
       mesg.respondSync({ error: `${err}`, status: "error" });
       writeStream?.emit("remove");
@@ -194,6 +201,7 @@ export async function writeFile({
   stream,
   maxWait = 1000 * 60 * 10, // 10 minutes
 }): Promise<{ bytes: number; chunks: number }> {
+  logger.debug("writeFile", { project_id, compute_server_id, path, maxWait });
   const name = randomId();
   try {
     function createReadStream() {

src/packages/conat/service/service.ts

@@ -210,7 +210,12 @@ export class ConatService extends EventEmitter {
         await mesg.respond(resp);
       } catch (err) {
         const data = { error: `${err}` };
-        await mesg.respond(data);
+        try {
+          await mesg.respond(data);
+        } catch (err2) {
+          // do not crash on sending an error report:
+          logger.debug("WARNING: unable to send error", this.name, err, err2);
+        }
       }
     }
   };

src/packages/conat/socket/util.ts

@@ -13,7 +13,7 @@ export type Role = "client" | "server";
 // socketio and use those to manage things.  This ping
 // is entirely a "just in case" backup if some event
 // were missed (e.g., a kill -9'd process...)
-export const PING_PONG_INTERVAL = 60000;
+export const PING_PONG_INTERVAL = 90000;
 
 // We queue up unsent writes, but only up to a point (to not have a huge memory issue).
 // Any write beyond this size result in an exception.
@@ -24,8 +24,8 @@ export const PING_PONG_INTERVAL = 60000;
 export const DEFAULT_MAX_QUEUE_SIZE = 1000;
 
 export let DEFAULT_COMMAND_TIMEOUT = 10_000;
-export let DEFAULT_KEEP_ALIVE = 90_000;
-export let DEFAULT_KEEP_ALIVE_TIMEOUT = 15_000;
+export let DEFAULT_KEEP_ALIVE = 25_000;
+export let DEFAULT_KEEP_ALIVE_TIMEOUT = 10_000;
 
 export function setDefaultSocketTimeouts({
   command = DEFAULT_COMMAND_TIMEOUT,

src/packages/frontend/admin/_style.sass

@@ -13,3 +13,6 @@
     td:first-child
       font-family: monospace
       font-weight: bold
+
+.admin-llm-test-running-row
+  background-color: #f0f0f0 !important