conats: completely migrate the hub-api over to use it... and it just works, first try!?

- wow.

Author: williamstein

src/packages/frontend/nats/client.ts

@@ -55,7 +55,6 @@ import {
 } from "@cocalc/nats/client";
 import type { ConnectionInfo } from "./types";
 import { fromJS } from "immutable";
-import { requestMany } from "@cocalc/nats/service/many";
 import Cookies from "js-cookie";
 import { ACCOUNT_ID_COOKIE } from "@cocalc/frontend/client/client";
 import { isConnected, waitUntilConnected } from "@cocalc/nats/util";
@@ -271,31 +270,18 @@ export class NatsClient extends EventEmitter {
     name,
     args = [],
     timeout = DEFAULT_TIMEOUT,
-    requestMany: requestMany0 = false,
   }: {
     service?: string;
     name: string;
     args: any[];
     timeout?: number;
-    // requestMany -- if true do a requestMany request, which is more complicated/slower, but
-    // supports arbitrarily large responses irregardless of the nats server max message size.
-    requestMany?: boolean;
   }) => {
-    const { nc } = await this.getEnv();
+    const { cn } = await this.getEnv();
     const subject = `hub.account.${this.client.account_id}.${service}`;
     try {
-      const data = this.jc.encode({
-        name,
-        args,
-      });
-      let resp;
-      await waitUntilConnected();
-      if (requestMany0) {
-        resp = await requestMany({ nc, subject, data, maxWait: timeout });
-      } else {
-        resp = await nc.request(subject, data, { timeout });
-      }
-      return this.jc.decode(resp.data);
+      const data = { name, args };
+      const resp = await cn.request(subject, data, { timeout });
+      return resp.data;
     } catch (err) {
       err.message = `${err.message} - callHub: subject='${subject}', name='${name}', `;
       throw err;

src/packages/nats/hub-api/db.ts

@@ -33,8 +33,6 @@ export interface DB {
   getLegacyTimeTravelPatches: (opts: {
     account_id?: string;
     uuid: string;
-    // you should set this to true to enable potentially very large response support
-    requestMany?: boolean;
     // also, make this bigger:
     timeout?: number;
   }) => Promise<string>;

src/packages/nats/hub-api/index.ts

@@ -1,11 +1,3 @@
-/*
-NOTE: If you need to send *very large responses* to a message or increase timeouts,
-see getLegacyTimeTravelPatches in db.ts.  You just have to allow the keys requestMany
-and timeout to the *first* argument of the function (which must be an object).
-The framework will then automatically allow large responses when the user sets
-requestMany:true.
-*/
-
 import { isValidUUID } from "@cocalc/util/misc";
 import { type Purchases, purchases } from "./purchases";
 import { type System, system } from "./system";
@@ -47,7 +39,6 @@ export function initHubApi(callHubApi): HubApi {
         const resp = await callHubApi({
           name: `${group}.${functionName}`,
           args,
-          requestMany: args[0]?.requestMany,
           timeout: args[0]?.timeout,
         });
         return handleErrorMessage(resp);

src/packages/nats/server/client.ts

@@ -551,12 +551,12 @@ export class Message {
     this.subject = subject;
   }
 
-  respond = (data: any) => {
+  respond = async (data: any) => {
     const subject = this.headers?.[REPLY_HEADER];
     if (!subject) {
       throw Error("message is not a request");
     }
-    this.client.publish(subject, data);
+    await this.client.publish(subject, data);
   };
 }
 

src/packages/nats/server/server.ts

@@ -14,6 +14,14 @@ For clustering:
    s0 = await require('@cocalc/server/nats/socketio').initConatServer({valkey:'redis://localhost:6379', port:3000})
    
    s1 = await require('@cocalc/server/nats/socketio').initConatServer({valkey:'redis://localhost:6379', port:3001})
+   
+Corresponding clients:
+
+   c0 = require('@cocalc/nats/server/client').connect('http://localhost:3000')
+
+   c1 = require('@cocalc/nats/server/client').connect('http://localhost:3001')
+
+---
     
 Or from cocalc/src
 
@@ -41,7 +49,14 @@ const MAX_PAYLOAD = 1 * MB;
 
 const MAX_DISCONNECTION_DURATION = 2 * 60 * 1000;
 
-const DEBUG = true;
+const DEBUG = false;
+
+interface InterestUpdate {
+  op: "add" | "delete";
+  subject: string;
+  queue?: string;
+  room: string;
+}
 
 export function init(opts) {
   return new ConatServer(opts);
@@ -167,15 +182,15 @@ export class ConatServer {
     let lastId = "0";
     let d = 50;
     while (true) {
-      console.log("waiting for interest update");
+      // console.log("waiting for interest update");
       const results = await this.valkey.sub.xread(
         "block" as any,
         0,
         "STREAMS",
         "interest",
         lastId,
       );
-      console.log("got ", results);
+      // console.log("got ", results);
       if (results == null) {
         d = Math.min(1000, d * 1.2);
         await delay(d);
@@ -189,7 +204,7 @@ export class ConatServer {
         this._updateInterest(update);
       }
       lastId = messages[messages.length - 1][0];
-      console.log({ lastId });
+      // console.log({ lastId });
     }
   };
 
@@ -344,29 +359,22 @@ export class ConatServer {
     socket.on("disconnecting", async () => {
       const rooms = Array.from(socket.rooms) as string[];
       const d = this.options.maxDisconnectionDuration ?? 0;
-      console.log(`will unsubscribe in ${d}ms unless client reconnects`);
+      // console.log(`will unsubscribe in ${d}ms unless client reconnects`);
       await delay(d);
       if (!this.io.of("/").adapter.sids.has(id)) {
-        console.log("client not back");
+        //  console.log("client not back");
         for (const room of rooms) {
           const subject = getSubjectFromRoom(room);
           this.unsubscribe({ socket, subject });
         }
         delete this.subscriptions[id];
       } else {
-        console.log("client is back!");
+        // console.log("client is back!");
       }
     });
   };
 }
 
-interface InterestUpdate {
-  op: "add" | "delete";
-  subject: string;
-  queue?: string;
-  room: string;
-}
-
 function getSubjectFromRoom(room: string) {
   if (room.startsWith("{")) {
     return JSON.parse(room).subject;

src/packages/server/nats/api/index.ts

@@ -12,7 +12,7 @@ your dev hub after doing the above.
 
 2. Run this script at the terminal:
 
-    echo "require('@cocalc/server/nats').default()" | COCALC_MODE='single-user' DEBUG_CONSOLE=yes DEBUG=cocalc:* node
+    echo "require('@cocalc/server/nats/api').initAPI()" | COCALC_MODE='single-user' DEBUG_CONSOLE=yes DEBUG=cocalc:* node
 
 
 3. Optional: start more servers -- requests get randomly routed to exactly one of them:
@@ -40,26 +40,18 @@ To view all requests (and replies) in realtime:
 And remember to use the nats command, do "pnpm nats-cli" from cocalc/src.
 */
 
-import { JSONCodec } from "nats";
 import getLogger from "@cocalc/backend/logger";
 import { type HubApi, getUserId, transformArgs } from "@cocalc/nats/hub-api";
-import { getConnection } from "@cocalc/backend/nats";
+import { getEnv } from "@cocalc/backend/nats";
 import userIsInGroup from "@cocalc/server/accounts/is-in-group";
 import { terminate as terminateDatabase } from "@cocalc/database/nats/changefeeds";
 import { terminate as terminateChangefeedServer } from "@cocalc/nats/changefeed/server";
-import { Svcm } from "@nats-io/services";
 import { terminate as terminateAuth } from "@cocalc/server/nats/auth";
 import { terminate as terminateTieredStorage } from "@cocalc/server/nats/tiered-storage/api";
-import { respondMany } from "@cocalc/nats/service/many";
 import { delay } from "awaiting";
-import { waitUntilConnected } from "@cocalc/nats/util";
-
-const MONITOR_INTERVAL = 30000;
 
 const logger = getLogger("server:nats:api");
 
-const jc = JSONCodec();
-
 export function initAPI() {
   mainLoop();
 }
@@ -88,93 +80,68 @@ async function mainLoop() {
   }
 }
 
-async function serviceMonitor({ nc, api, subject }) {
-  while (!terminate) {
-    logger.debug(`serviceMonitor: waiting ${MONITOR_INTERVAL}ms`);
-    await delay(MONITOR_INTERVAL);
-    try {
-      await waitUntilConnected();
-      await nc.request(subject, jc.encode({ name: "ping" }), {
-        timeout: 7500,
-      });
-      logger.debug("serviceMonitor: ping succeeded");
-    } catch (err) {
-      logger.debug(
-        `serviceMonitor: ping failed, so restarting service -- ${err}`,
-      );
-      api.stop();
-      return;
-    }
-  }
-}
-
 async function serve() {
   const subject = "hub.*.*.api";
   logger.debug(`initAPI -- subject='${subject}', options=`, {
     queue: "0",
   });
-  const nc = await getConnection();
-  // @ts-ignore
-  const svcm = new Svcm(nc);
-
-  await waitUntilConnected();
-  const service = await svcm.add({
-    name: "hub-server",
-    version: "0.1.0",
-    description: "CoCalc Hub Server",
-  });
-
-  const api = service.addEndpoint("api", { subject });
-  serviceMonitor({ api, subject, nc });
-  await listen({ api, subject });
+  const { cn } = await getEnv();
+  const api = await cn.subscribe(subject);
+  for await (const mesg of api) {
+    (async () => {
+      try {
+        await handleMessage({ api, subject, mesg });
+      } catch (err) {
+        logger.debug(`WARNING: unexpected error  - ${err}`);
+      }
+    })();
+  }
 }
 
-async function listen({ api, subject }) {
-  for await (const mesg of api) {
-    const request = jc.decode(mesg.data) ?? ({} as any);
-    if (request.name == "system.terminate") {
+async function handleMessage({ api, subject, mesg }) {
+  const request = mesg.data ?? ({} as any);
+  if (request.name == "system.terminate") {
+    // special hook so admin can terminate handling. This is useful for development.
+    const { account_id } = getUserId(mesg.subject);
+    if (!(!!account_id && (await userIsInGroup(account_id, "admin")))) {
+      mesg.respond({ error: "only admin can terminate" });
+      return;
+    }
+    // TODO: could be part of handleApiRequest below, but done differently because
+    // one case halts this loop
+    const { service } = request.args[0] ?? {};
+    logger.debug(`Terminate service '${service}'`);
+    if (service == "db") {
+      terminateDatabase();
+      mesg.respond({ status: "terminated", service });
+      return;
+    } else if (service == "auth") {
+      terminateAuth();
+      mesg.respond({ status: "terminated", service });
+      return;
+    } else if (service == "tiered-storage") {
+      terminateTieredStorage();
+      mesg.respond({ status: "terminated", service });
+      return;
+    } else if (service == "changefeeds") {
+      terminateChangefeedServer();
+      mesg.respond({ status: "terminated", service });
+      return;
+    } else if (service == "api") {
       // special hook so admin can terminate handling. This is useful for development.
-      const { account_id } = getUserId(mesg.subject);
-      if (!(!!account_id && (await userIsInGroup(account_id, "admin")))) {
-        mesg.respond(jc.encode({ error: "only admin can terminate" }));
-        continue;
-      }
-      // TODO: could be part of handleApiRequest below, but done differently because
-      // one case halts this loop
-      const { service } = request.args[0] ?? {};
-      logger.debug(`Terminate service '${service}'`);
-      if (service == "db") {
-        terminateDatabase();
-        mesg.respond(jc.encode({ status: "terminated", service }));
-        continue;
-      } else if (service == "auth") {
-        terminateAuth();
-        mesg.respond(jc.encode({ status: "terminated", service }));
-        continue;
-      } else if (service == "tiered-storage") {
-        terminateTieredStorage();
-        mesg.respond(jc.encode({ status: "terminated", service }));
-        continue;
-      } else if (service == "changefeeds") {
-        terminateChangefeedServer();
-        mesg.respond(jc.encode({ status: "terminated", service }));
-        continue;
-      } else if (service == "api") {
-        // special hook so admin can terminate handling. This is useful for development.
-        console.warn("TERMINATING listening on ", subject);
-        logger.debug("TERMINATING listening on ", subject);
-        terminate = true;
-        mesg.respond(jc.encode({ status: "terminated", service }));
-        api.stop();
-        return;
-      } else {
-        mesg.respond(jc.encode({ error: `Unknown service ${service}` }));
-      }
+      console.warn("TERMINATING listening on ", subject);
+      logger.debug("TERMINATING listening on ", subject);
+      terminate = true;
+      mesg.respond({ status: "terminated", service });
+      api.stop();
+      return;
     } else {
-      // we explicitly do NOT await this, since we want this hub server to handle
-      // potentially many messages at once, not one at a time!
-      handleApiRequest({ request, mesg });
+      mesg.respond({ error: `Unknown service ${service}` });
     }
+  } else {
+    // we explicitly do NOT await this, since we want this hub server to handle
+    // potentially many messages at once, not one at a time!
+    handleApiRequest({ request, mesg });
   }
 }
 
@@ -193,7 +160,7 @@ async function handleApiRequest({ request, mesg }) {
     resp = { error: `${err}` };
   }
   try {
-    await respondMany({ mesg, data: jc.encode(resp) });
+    await mesg.respond(resp);
   } catch (err) {
     // there's nothing we can do here, e.g., maybe NATS just died.
     logger.debug(