hopefully fix issue #8424 -- hub conat api repeatedly throws an unhandled exception

- this code actually rewrites all of the callback use of socketio client
  to use their async api, which just makes the code easier to read. I
  don't think that necessarily fixed anything, though I did restructure
  the code so it is much clearer, which was enough to convince me I just
  missed an await / catch somewhere.

- indeed, all mesg.respond must be await and catch'd, and there were
  some that were not!  So that's almost certainly the issue.  I add a
  noThrow option, for when there's just no need to catch, and audited
  the code and found several offenders.

Author: williamstein

src/packages/conat/core/client.ts

@@ -209,7 +209,6 @@ import * as msgpack from "@msgpack/msgpack";
 import { randomId } from "@cocalc/conat/names";
 import type { JSONValue } from "@cocalc/util/types";
 import { EventEmitter } from "events";
-import { callback } from "awaiting";
 import {
   isValidSubject,
   isValidSubjectWithoutWildcards,
@@ -531,10 +530,7 @@ export class Client extends EventEmitter {
   }
 
   cluster = async () => {
-    return await callback(
-      this.conn.timeout(10000).emit.bind(this.conn),
-      "cluster",
-    );
+    return await this.conn.timeout(10000).emitWithAck("cluster");
   };
 
   disconnect = () => {
@@ -608,20 +604,14 @@ export class Client extends EventEmitter {
       );
     }
     timeout = Math.min(timeout, MAX_INTEREST_TIMEOUT);
-    const f = (cb) => {
-      this.conn
-        .timeout(timeout ? timeout : 10000)
-        .emit("wait-for-interest", { subject, timeout }, (err, response) => {
-          if (err) {
-            cb(err);
-          } else if (response.error) {
-            cb(new ConatError(response.error, { code: response.code }));
-          } else {
-            cb(undefined, response);
-          }
-        });
-    };
-    return await callback(f);
+    const response = await this.conn.timeout(timeout ? timeout : 10000).emitWithAck(
+      "wait-for-interest",
+      { subject, timeout },
+    );
+    if (response.error) {
+      throw new ConatError(response.error, { code: response.code });
+    }
+    return response;
   };
 
   recvStats = (bytes: number) => {
@@ -827,11 +817,9 @@ export class Client extends EventEmitter {
     let stable = true;
     if (missing.length > 0) {
       stable = false;
-      const resp = await callback(
-        this.conn.timeout(timeout).emit.bind(this.conn),
-        "subscribe",
-        missing,
-      );
+      const resp = await this.conn
+        .timeout(timeout)
+        .emitWithAck("subscribe", missing);
       // some subscription could fail due to permissions changes, e.g., user got
       // removed from a project.
       for (let i = 0; i < missing.length; i++) {
@@ -851,11 +839,7 @@ export class Client extends EventEmitter {
       }
     }
     if (extra.length > 0) {
-      await callback(
-        this.conn.timeout(timeout).emit.bind(this.conn),
-        "unsubscribe",
-        extra,
-      );
+      await this.conn.timeout(timeout).emitWithAck("unsubscribe", extra);
       stable = false;
     }
     return stable;
@@ -866,11 +850,9 @@ export class Client extends EventEmitter {
   private getSubscriptions = async (
     timeout = DEFAULT_REQUEST_TIMEOUT,
   ): Promise<Set<string>> => {
-    const subs = await callback(
-      this.conn.timeout(timeout).emit.bind(this.conn),
-      "subscriptions",
-      null,
-    );
+    const subs = await this.conn
+      .timeout(timeout)
+      .emitWithAck("subscriptions", null);
     return new Set(subs);
   };
 
@@ -958,30 +940,29 @@ export class Client extends EventEmitter {
     this.stats.subs++;
     let promise;
     if (confirm) {
-      const f = (cb) => {
-        const handle = (response) => {
-          if (response?.error) {
-            cb(new ConatError(response.error, { code: response.code }));
+      const f = async () => {
+        let response;
+        try {
+          if (timeout) {
+            response = await this.conn
+              .timeout(timeout)
+              .emitWithAck("subscribe", { subject, queue });
           } else {
-            cb(response?.error, response);
-          }
-        };
-        if (timeout) {
-          this.conn
-            .timeout(timeout)
-            .emit("subscribe", { subject, queue }, (err, response) => {
-              if (err) {
-                handle({ error: `${err}`, code: 408 });
-              } else {
-                handle(response);
-              }
+            // this should never be used -- see above
+            response = await this.conn.emitWithAck("subscribe", {
+              subject,
+              queue,
             });
-        } else {
-          // this should never be used -- see above
-          this.conn.emit("subscribe", { subject, queue }, handle);
+          }
+        } catch (err) {
+          throw new ConatError(`${err}`, { code: 408 });
         }
+        if (response?.error) {
+          throw new ConatError(response.error, { code: response.code });
+        }
+        return response;
       };
-      promise = callback(f);
+      promise = f();
     } else {
       this.conn.emit("subscribe", { subject, queue });
       promise = undefined;
@@ -1182,7 +1163,8 @@ export class Client extends EventEmitter {
       // already closed
       return { bytes: 0 };
     }
-    return this._publish(subject, mesg, opts);
+    // must NOT confirm
+    return this._publish(subject, mesg, { ...opts, confirm: false });
   };
 
   publish = async (
@@ -1254,6 +1236,7 @@ export class Client extends EventEmitter {
       encoding = DEFAULT_ENCODING,
       confirm,
       timeout = DEFAULT_PUBLISH_TIMEOUT,
+      noThrow,
     }: PublishOptions & { confirm?: boolean } = {},
   ) => {
     if (this.isClosed()) {
@@ -1301,49 +1284,35 @@ export class Client extends EventEmitter {
         v.push(headers);
       }
       if (confirm) {
-        let done = false;
-        const f = (cb) => {
-          const handle = (response) => {
-            if (this.state == "closed" && response?.error) {
-              if (!process.env.COCALC_TEST_MODE) {
-                console.warn(
-                  "conat client: ignoring outstanding error message since client closed",
-                );
-              }
-              cb(undefined, response);
-              return;
-            }
-            // console.log("_publish", { done, subject, mesg, headers, confirm });
-            if (response?.error) {
-              cb(new ConatError(response.error, { code: response.code }));
-            } else {
-              cb(response?.error, response);
-            }
-          };
+        const f = async () => {
           if (timeout) {
-            const timer = setTimeout(() => {
-              done = true;
-              cb(new ConatError("timeout", { code: 408 }));
-            }, timeout);
-
-            this.conn.timeout(timeout).emit("publish", v, (err, response) => {
-              if (done) {
-                return;
-              }
-              clearTimeout(timer);
-              if (err) {
-                handle({ error: `${err}`, code: 408 });
+            try {
+              const response = await this.conn
+                .timeout(timeout)
+                .emitWithAck("publish", v);
+              if (response?.error) {
+                throw new ConatError(response.error, { code: response.code });
               } else {
-                handle(response);
+                return response;
               }
-            });
+            } catch (err) {
+              throw new ConatError(`timeout - ${subject} - ${err}`, {
+                code: 408,
+              });
+            }
           } else {
-            this.conn.emit("publish", v, handle);
+            return await this.conn.emitWithAck("publish", v);
           }
         };
         const promise = (async () => {
-          const response = await callback(f);
-          count = Math.max(count, response.count ?? 0);
+          try {
+            const response = await f();
+            count = Math.max(count, response.count ?? 0);
+          } catch (err) {
+            if (!noThrow) {
+              throw err;
+            }
+          }
         })();
         promises.push(promise);
       } else {
@@ -1616,6 +1585,13 @@ interface PublishOptions {
   // on success.  Note that waitForInterest always has a timeout, defaulting
   // to DEFAULT_WAIT_FOR_INTEREST_TIMEOUT if above timeout not given.
   waitForInterest?: boolean;
+
+  // noThrow -- if set and publishing would throw an exception, it is
+  // instead silently dropped and undefined is returned instead.
+  // Use this where you might want to use publishSync, but still want
+  // to ensure there is interest; however, it's not important to know
+  // if there was an error sending.
+  noThrow?: boolean;
 }
 
 interface RequestManyOptions extends PublishOptions {

src/packages/conat/files/sync.ts

@@ -0,0 +1,70 @@
+/*
+File Sync Services
+
+There's a bunch of these running with access to all project files.
+This is somewhat similar to the persist servers in architecture,
+except this is to support sync editing of files.
+
+These listen on this sticky subject:
+
+    subject = sync.project-{project_id}.edit
+
+User sends a message that contains the path to a file they want to edit.
+The fact user can even send the message means they have read/write
+privileges to the subject.
+
+
+the path to a file in a project
+
+OUTPUT: starts (or keeps running) the filesystem aware side of an editing session
+
+*/
+
+// import { type Client } from "@cocalc/conat/core/client";
+
+interface SyncDoc {
+  close: () => void;
+}
+
+export type SyncDocCreator = (opts: {
+  project_id: string;
+  path: string;
+  doctype?: any;
+}) => SyncDoc;
+
+/*
+interface Options {
+  client: Client;
+
+  // projects = absolute path in filesystem to user projects, so join(projects, project_id)
+  // is the path to project_id's files.
+  projects: string;
+
+  createSyncDoc: SyncDocCreator;
+}
+
+export function init(opts: Options) {
+  return new SyncServer(opts.client, opts.projects, opts.createSyncDocs);
+}
+
+interface Api {
+  open: (opts: { path: string; doctype?: any }) => Promise<void>;
+}
+
+class SyncServer {
+  constructor(client: Client, projects: string, createSyncDoc: SyncDocCreator) {
+    this.client.service = await client1.service<Api>("arith.*", {
+      add: async (a, b) => a + b,
+      mul: async (a, b) => a * b,
+      // Here we do NOT use an arrow => function and this is
+      // bound to the calling mesg, which lets us get the subject.
+      // Because user identity and permissions are done via wildcard
+      // subjects, having access to the calling message is critical
+      async open({ path, doctype }) {
+        const mesg: Message = this as any;
+        console.log(mesg.subject);
+      },
+    });
+  }
+}
+*/

src/packages/conat/socket/client.ts

@@ -176,7 +176,7 @@ export class ConatSocketClient extends ConatSocketBase {
           return;
         } else if (cmd == "ping") {
           // logger.silly("responding to ping from server", this.id);
-          mesg.respond(null);
+          mesg.respondSync(null);
         } else if (mesg.isRequest()) {
           // logger.silly("client got request");
           this.emit("request", mesg);

src/packages/conat/socket/server.ts

@@ -191,7 +191,8 @@ export class ConatSocketServer extends ConatSocketBase {
       mesg.respondSync("closed");
     } else if (cmd == "connect") {
       // very important that connected is successfully delivered, so do not use respondSync.
-      mesg.respond("connected");
+      // Using respond waits for interest.
+      mesg.respond("connected", { noThrow: true });
     } else {
       mesg.respondSync({ error: `unknown command - '${cmd}'` });
     }

src/packages/project/conat/api/index.ts

@@ -23,9 +23,9 @@ How to do development (so in a dev project doing cc-in-cc dev).
     export DEBUG=cocalc:*
     export DEBUG_CONSOLE=yes
 
-If API_KEY is a project-wide API key, then you can change 
-COCALC_PROJECT_ID however you want and don't have to worry 
-about whether the project is running or the project secret 
+If API_KEY is a project-wide API key, then you can change
+COCALC_PROJECT_ID however you want and don't have to worry
+about whether the project is running or the project secret
 key changing when the project is restarted.
 
 2. Then do this:
@@ -103,30 +103,30 @@ async function handleMessage(api, subject, mesg) {
     const { service } = request.args[0] ?? {};
     if (service == "open-files") {
       terminateOpenFiles();
-      mesg.respond({ status: "terminated", service });
+      await mesg.respond({ status: "terminated", service });
       return;
     } else if (service == "listings") {
       closeListings();
-      mesg.respond({ status: "terminated", service });
+      await mesg.respond({ status: "terminated", service });
       return;
     } else if (service == "files:read") {
       await closeFilesRead();
-      mesg.respond({ status: "terminated", service });
+      await mesg.respond({ status: "terminated", service });
       return;
     } else if (service == "files:write") {
       await closeFilesWrite();
-      mesg.respond({ status: "terminated", service });
+      await mesg.respond({ status: "terminated", service });
       return;
     } else if (service == "api") {
       // special hook so admin can terminate handling. This is useful for development.
       terminate = true;
       console.warn("TERMINATING listening on ", subject);
       logger.debug("TERMINATING listening on ", subject);
-      mesg.respond({ status: "terminated", service });
+      await mesg.respond({ status: "terminated", service });
       api.stop();
       return;
     } else {
-      mesg.respond({ error: `Unknown service ${service}` });
+      await mesg.respond({ error: `Unknown service ${service}` });
     }
   } else {
     handleApiRequest(request, mesg);
@@ -147,7 +147,7 @@ async function handleApiRequest(request, mesg) {
       resp = { error: `${err}` };
     }
   }
-  mesg.respond(resp);
+  await mesg.respond(resp);
 }
 
 import * as system from "./system";