conat: got all tests passing again

Author: williamstein

src/packages/backend/conat/test/core/core-stream-break.test.ts

@@ -17,6 +17,7 @@ import {
   wait,
   delay,
   persistServer as setupPersistServer,
+  setDefaultTimeouts,
 } from "@cocalc/backend/conat/test/setup";
 
 beforeAll(before);
@@ -27,6 +28,9 @@ describe("stop persist server, create a client, create an ephemeral core-stream,
   let pclient;
   let persistServer;
 
+  // use much shorter timeout for this test of restarting persist server, which isn't a NORMAL thing.
+  setDefaultTimeouts({ request: 750, publish: 750 });
+
   it("close the persist server that was setup as part of before above", async () => {
     await setupPersistServer.end();
   });
@@ -59,7 +63,7 @@ describe("stop persist server, create a client, create an ephemeral core-stream,
 
     await expect(async () => {
       await stream.publish("y", { timeout: 100 });
-    }).rejects.toThrowError("no subscribers");
+    }).rejects.toThrowError();
 
     try {
       await stream.publish("y", { timeout: 100 });
@@ -68,9 +72,18 @@ describe("stop persist server, create a client, create an ephemeral core-stream,
     }
   });
 
-  it("starts persist server and can publish again", async () => {
+  it("starts persist server and can eventually publish again", async () => {
     persistServer = initPersistServer({ client: pclient });
-    await stream.publish("y");
+    await wait({
+      until: async () => {
+        try {
+          await stream.publish("y");
+          return true;
+        } catch {
+          return false;
+        }
+      },
+    });
   });
 
   it("creates a dstream, publishes, sees it hasn't saved, starts persist server and sees save works again", async () => {

src/packages/backend/conat/test/core/core-stream.test.ts

@@ -270,9 +270,8 @@ describe("test key:value delete", () => {
   });
 
   it("also delete the empty key one", async () => {
-    const n = stream.length;
     await stream2.deleteKv("");
-    await wait({ until: () => stream.length > n });
+    await wait({ until: async () => (await stream2.getKv("")) == undefined });
     expect(await stream2.getKv("")).toEqual(undefined);
     await wait({ until: () => stream.getKv("") === undefined });
   });

src/packages/backend/conat/test/setup.ts

@@ -15,6 +15,7 @@ import { tmpdir } from "node:os";
 import { join } from "path";
 export { wait } from "@cocalc/backend/conat/test/util";
 export { delay } from "awaiting";
+export { setDefaultTimeouts } from "@cocalc/conat/core/client";
 
 const logger = getLogger("conat:test:setup");
 

src/packages/conat/core/client.ts

@@ -263,10 +263,22 @@ export type ClientOptions = Options & {
 
 const INBOX_PREFIX = "_INBOX";
 const REPLY_HEADER = "CN-Reply";
-export const DEFAULT_REQUEST_TIMEOUT = 10000;
-export const DEFAULT_PUBLISH_TIMEOUT = 7500;
 const MAX_HEADER_SIZE = 100000;
 
+export let DEFAULT_REQUEST_TIMEOUT = 7500;
+export let DEFAULT_PUBLISH_TIMEOUT = 7500;
+
+export function setDefaultTimeouts({
+  request = DEFAULT_REQUEST_TIMEOUT,
+  publish = DEFAULT_PUBLISH_TIMEOUT,
+}: {
+  request?: number;
+  publish?: number;
+}) {
+  DEFAULT_REQUEST_TIMEOUT = request;
+  DEFAULT_PUBLISH_TIMEOUT = publish;
+}
+
 export enum DataEncoding {
   MsgPack = 0,
   JsonCodec = 1,

src/packages/conat/persist/client.ts

@@ -229,7 +229,10 @@ export class PersistStreamClient extends EventEmitter {
       timeout,
       maxWait,
     });
-    for await (const { data } of sub) {
+    for await (const { data, headers } of sub) {
+      if (headers?.error) {
+        throw new ConatError(`${headers.error}`, { code: headers.code });
+      }
       if (data == null || this.socket.state == "closed") {
         // done
         return;

src/packages/conat/persist/server.ts

@@ -40,7 +40,7 @@ client = await require('@cocalc/backend/conat').conat(); s = await require('@coc
 
 */
 
-import { type Client } from "@cocalc/conat/core/client";
+import { type Client, ConatError } from "@cocalc/conat/core/client";
 import {
   type ConatSocketServer,
   type ServerSocket,
@@ -94,6 +94,7 @@ export function server({
       subject: socket.subject,
     });
     let error = "";
+    let errorCode: any = undefined;
     let changefeed = false;
     let storage: undefined | StorageOptions = undefined;
     let stream: undefined | PersistentStream = undefined;
@@ -111,7 +112,8 @@ export function server({
             startChangefeed({ socket, stream, messagesThresh });
           }
         } catch (err) {
-          error = `${err};`;
+          error = `${err}`;
+          errorCode = err.code;
         }
       }
     });
@@ -127,7 +129,7 @@ export function server({
 
       try {
         if (error) {
-          throw Error(error);
+          throw new ConatError(error, { code: errorCode });
         }
         if (storage === undefined || stream === undefined) {
           // this happens, e.g., when you restart both the persist server and the conat
@@ -239,7 +241,7 @@ export function server({
 async function getAll({ stream, mesg, request, messagesThresh }) {
   let seq = 0;
   const respond = (error?, messages?: StoredMessage[]) => {
-    mesg.respondSync(messages, { headers: { error, seq } });
+    mesg.respondSync(messages, { headers: { error, seq, code: error?.code } });
     seq += 1;
   };
 

src/packages/conat/sync/core-stream.ts

@@ -201,7 +201,20 @@ export class CoreStream<T = any> extends EventEmitter {
       start_seq: this._start_seq,
       noEmit: true,
     });
-    this.configOptions = await this.config(this.configOptions);
+    let d = 750;
+    while (this.client != null) {
+      try {
+        this.configOptions = await this.config(this.configOptions);
+        break;
+      } catch (err) {
+        if (err.code == 403) {
+          // fatal permission error
+          throw err;
+        }
+        d = Math.min(15000, d * 1.3);
+        await delay(d);
+      }
+    }
     // NOTE: if we miss a message between getAllFromLeader and when we start listening,
     // then the sequence number will have a gap, and we'll immediately reconnect, starting
     // at the right point. So no data can possibly be lost.
@@ -245,27 +258,38 @@ export class CoreStream<T = any> extends EventEmitter {
     if (this.storage == null) {
       throw Error("bug -- storage must be set");
     }
-    if (this.client == null) {
-      return;
-    }
-    // console.log("get persistent stream", { start_seq });
-    const sub = await this.persistClient.getAll({
-      start_seq,
-    });
-    while (true) {
-      const { value, done } = await sub.next();
-      // console.log("got ", { value, done });
-      if (done) {
-        return;
-      }
-      const messages = value as StoredMessage[];
-      const seq = this.processPersistentMessages(messages, noEmit);
-      if (seq != null) {
-        // we update start_seq in case we need to try again
-        start_seq = seq! + 1;
+    let d = 750;
+    while (this.client != null) {
+      try {
+        // console.log("get persistent stream", { start_seq });
+        const sub = await this.persistClient.getAll({
+          start_seq,
+        });
+        // console.log("got sub");
+        while (true) {
+          const { value, done } = await sub.next();
+          if (done) {
+            return;
+          }
+          const messages = value as StoredMessage[];
+          const seq = this.processPersistentMessages(messages, noEmit);
+          if (seq != null) {
+            // we update start_seq in case we need to try again
+            start_seq = seq! + 1;
+          }
+        }
+      } catch (err) {
+        //         console.log(
+        //           `WARNING: getAllFromPersist - failed -- ${err}, code=${err.code}`,
+        //         );
+        if (err.code == 403) {
+          // fatal permission error
+          throw err;
+        }
+        d = Math.min(15000, d * 1.3);
+        await delay(d);
       }
     }
-    // console.log("finished getAll");
   };
 
   private processPersistentMessages = (