nats file read/write servers -- keep them running

- I really need to create a new NATS client that wraps ALL subscriptions
  with this sort of logic and has better semantics...

Author: williamstein

src/packages/nats/files/read.ts

@@ -41,6 +41,7 @@ for await (const chunk of await a.readFile({project_id:'00847397-d6a8-4cb0-96a8-
 import { getEnv } from "@cocalc/nats/client";
 import { projectSubject } from "@cocalc/nats/names";
 import { Empty, headers, type Subscription } from "@nats-io/nats-core";
+import { runLoop } from "./util";
 
 let subs: { [name: string]: Subscription } = {};
 export async function close({ project_id, compute_server_id, name = "" }) {
@@ -76,13 +77,16 @@ export async function createServer({
     return;
   }
   const { nc } = await getEnv();
-  // console.log(subject);
-  const sub = nc.subscribe(subject);
-  subs[subject] = sub;
-  listen(sub, createReadStream);
+  runLoop({
+    listen,
+    subs,
+    subject,
+    nc,
+    opts: { createReadStream },
+  });
 }
 
-async function listen(sub, createReadStream) {
+async function listen({ sub, createReadStream }) {
   // NOTE: we just handle as many messages as we get in parallel, so this
   // could be a large number of simultaneous downloads. These are all by
   // authenticated users of the project, and the load is on the project,

src/packages/nats/files/util.ts

@@ -0,0 +1,22 @@
+import { delay } from "awaiting";
+import { waitUntilConnected } from "@cocalc/nats/util";
+import { getLogger } from "@cocalc/nats/client";
+
+const logger = getLogger("files:util");
+
+export async function runLoop({ subs, listen, opts, subject, nc }) {
+  while (true) {
+    const sub = nc.subscribe(subject);
+    subs[subject] = sub;
+    try {
+      await listen({ ...opts, sub });
+    } catch (err) {
+      logger.debug(`runLoop: error - ${err}`);
+    }
+    if (subs[subject] == null) return;
+    await delay(3000 + Math.random());
+    await waitUntilConnected();
+    if (subs[subject] == null) return;
+    logger.debug(`runLoop: will restart`);
+  }
+}

src/packages/nats/files/write.ts

@@ -73,6 +73,7 @@ import {
 import { projectSubject } from "@cocalc/nats/names";
 import { type Subscription } from "@nats-io/nats-core";
 import { type Readable } from "node:stream";
+import { runLoop } from "./util";
 
 function getWriteSubject({ project_id, compute_server_id }) {
   return projectSubject({
@@ -111,9 +112,13 @@ export async function createServer({
     return;
   }
   const { nc } = await getEnv();
-  sub = nc.subscribe(subject);
-  subs[subject] = sub;
-  listen({ sub, createWriteStream, project_id, compute_server_id });
+  runLoop({
+    listen,
+    subs,
+    subject,
+    nc,
+    opts: { createWriteStream, project_id, compute_server_id },
+  });
 }
 
 async function listen({