conat: migrating time test; nailing down order of subscription startup

Author: williamstein

src/packages/backend/conat/test/core/connect.test.ts

@@ -1,3 +1,7 @@
+/*
+pnpm test ./connect.test.ts
+*/
+
 import { getPort } from "@cocalc/backend/conat/test/util";
 import { initConatServer } from "@cocalc/backend/conat/test/setup";
 import { connect } from "@cocalc/backend/conat/conat";
@@ -50,7 +54,7 @@ describe("create server *after* client and ensure connects properly", () => {
   });
 });
 
-describe("create server after creating a subscription and publishing a message, and observe that it works and nothing is lost", () => {
+describe("create server after sync creating a subscription and publishing a message, and observe that messages are dropped", () => {
   let cn;
   it("starts a client, despite there being no server yet", async () => {
     cn = connect(`http://localhost:${port}`, { path });
@@ -71,17 +75,53 @@ describe("create server after creating a subscription and publishing a message,
     await wait({ until: () => cn.conn.connected });
   });
 
-  it("see both messages we sent before connecting arrive", async () => {
-    const { value: mesg1 } = await sub.next();
-    expect(mesg1.data).toBe("hello");
-    const { value: mesg2 } = await sub.next();
-    expect(mesg2.data).toBe("conat");
-  });
-
-  it("publish another message", async () => {
+  it("see that both messages we sent before connecting were dropped", async () => {
     const { bytes, count } = await cn.publish("xyz", "more");
     expect(count).toBe(1);
     expect(bytes).toBe(5);
+    const { value: mesg1 } = await sub.next();
+    // we just got a message but it's AFTER the two above.
+    expect(mesg1.data).toBe("more");
+  });
+
+  it("clean up", () => {
+    server.close();
+    cn.close();
+    sub.close();
+  });
+});
+
+describe("create server after async creating a subscription and async publishing a message, and observe that it DOES works", () => {
+  let cn;
+  it("starts a client, despite there being no server yet", async () => {
+    cn = connect(`http://localhost:${port}`, { path });
+    expect(cn.conn.connected).toBe(false);
+  });
+
+  let sub;
+  let recv: any[] = [];
+  it("create a sync subscription before the server exists", () => {
+    const f = async () => {
+      sub = await cn.subscribe("xyz");
+      await cn.publish("xyz", "hello");
+      const { value: mesg } = await sub.next();
+      recv.push(mesg.data);
+      await cn.publish("xyz", "conat");
+      const { value: mesg2 } = await sub.next();
+      recv.push(mesg2.data);
+    };
+    f();
+  });
+
+  let server;
+  it("start the server", async () => {
+    server = await initConatServer({ port, path });
+    await wait({ until: () => cn.conn.connected });
+  });
+
+  it("see that both messages we sent before connecting arrive", async () => {
+    await wait({ until: () => recv.length == 2 });
+    expect(recv).toEqual(["hello", "conat"]);
   });
 
   it("clean up", () => {

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

@@ -7,6 +7,8 @@ import {
 } from "@cocalc/conat/core/server";
 import { Server } from "socket.io";
 import getLogger from "@cocalc/backend/logger";
+import { setNatsClient } from "@cocalc/conat/client";
+import { sha1 } from "@cocalc/backend/sha1";
 
 const logger = getLogger("conat:test:setup");
 
@@ -35,6 +37,12 @@ export async function before() {
   port = await getPort();
   address = `http://localhost:${port}`;
   server = await initConatServer({ port, path });
+  setNatsClient({
+    getNatsEnv: async () => {
+      return { cn: connect(), sha1 } as any;
+    },
+    getLogger,
+  });
 }
 
 const clients: Client[] = [];

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

@@ -4,10 +4,11 @@ DEVELOPMENT:
 pnpm test ./time.test.ts
 */
 
-// this sets client
-import "@cocalc/backend/conat";
-
+import { timeClient, createTimeService } from "@cocalc/conat/service/time";
 import time, { getSkew } from "@cocalc/conat/time";
+import { before, after } from "@cocalc/backend/conat/test/setup";
+
+beforeAll(before);
 
 describe("get time from nats", () => {
   it("tries to get the time before the skew, so it is not initialized yet", () => {
@@ -29,14 +30,13 @@ describe("get time from nats", () => {
   });
 });
 
-import { timeClient, createTimeService } from "@cocalc/conat/service/time";
-
 describe("start the time server and client and test that it works", () => {
   it("starts the time server and queries it", async () => {
-    await import("@cocalc/backend/conat");
     createTimeService();
     const client = timeClient();
     const t = await client.time();
     expect(Math.abs(Date.now() - t)).toBeLessThan(200);
   });
 });
+
+afterAll(after);

src/packages/conat/client.ts

@@ -62,7 +62,6 @@ export class ClientWithState extends EventEmitter {
         return this.env;
       }
       this.env = await client.getNatsEnv();
-      this.monitorConnectionState(this.env.nc);
       return this.env;
     });
     this.account_id = client.account_id;
@@ -85,7 +84,7 @@ export class ClientWithState extends EventEmitter {
   };
 
   close = () => {
-    this.env?.nc.close();
+    this.env?.nc?.close();
     this.setConnectionState("closed");
     this.removeAllListeners();
     delete this.env;
@@ -99,22 +98,6 @@ export class ClientWithState extends EventEmitter {
     this.emit(state);
     this.emit("state", state);
   };
-
-  private monitorConnectionState = async (nc) => {
-    this.setConnectionState("connected");
-
-    for await (const { type } of nc.status()) {
-      if (this.state == "closed") {
-        return;
-      }
-      if (type.includes("ping") || type == "update" || type == "reconnect") {
-        // connection is working well
-        this.setConnectionState("connected");
-      } else if (type == "reconnecting") {
-        this.setConnectionState("connecting");
-      }
-    }
-  };
 }
 
 // do NOT do this until some explicit use of nats is initiated, since we shouldn't

src/packages/conat/core/client.ts

@@ -420,6 +420,11 @@ export class Client {
       throw Error(`already subscribed to '${subject}'`);
     }
     this.queueGroups[subject] = queue;
+    const sub = new SubscriptionEmitter({
+      client: this,
+      subject,
+      closeWhenOffCalled,
+    });
     let promise;
     if (confirm) {
       const f = (cb) => {
@@ -436,11 +441,6 @@ export class Client {
       this.conn.emit("subscribe", { subject, queue });
       promise = undefined;
     }
-    const sub = new SubscriptionEmitter({
-      client: this,
-      subject,
-      closeWhenOffCalled,
-    });
     sub.once("close", () => {
       if (this.queueGroups?.[subject] == null) {
         return;

src/packages/conat/core/server.ts

@@ -289,7 +289,9 @@ export class ConatServer {
       });
     }
     const room = socketSubjectRoom({ socket, subject });
-    socket.join(room);
+    // critical to await socket.join so we don't advertise that there is 
+    // a subscriber before the socket is actually getting messages.
+    await socket.join(room); 
     await this.updateInterest({ op: "add", subject, room, queue });
   };
 

src/packages/conat/service/service.ts

@@ -71,9 +71,9 @@ export async function callNatsService(opts: ServiceCall): Promise<any> {
   try {
     return await doRequest();
   } catch (err) {
-    //console.log(`request to '${subject}' failed -- ${err}`);
+    // console.log(`request to '${subject}' failed -- ${err}`);
     // it failed.
-    if (err.name == "NatsError" && !opts.noRetry) {
+    if (!opts.noRetry) {
       // it's a nats problem
       const p = opts.path ? `${trunc_middle(opts.path, 64)}:` : "";
       if (err.code == 503) {