Merge remote-tracking branch 'origin/master' into conat-router-peer-discovery

Author: haraldschilly

src/packages/backend/conat/test/cluster/cluster-basics.test.ts

@@ -90,7 +90,6 @@ describe("create a cluster enabled socketio server and test that the streams upd
     link = await clusterLink(
       server.address(),
       server.options.systemAccountPassword,
-      () => {},
     );
     await wait({
       until: () => {
@@ -156,7 +155,6 @@ describe("create a cluster enabled socketio server and test that the streams upd
     const link2 = await clusterLink(
       server.address(),
       server.options.systemAccountPassword,
-      () => {},
     );
     await wait({
       until: () => {

src/packages/backend/conat/test/cluster/cluster-sticky-state.test.ts

@@ -0,0 +1,188 @@
+import {
+  before,
+  after,
+  defaultCluster as servers,
+  waitForConsistentState,
+  wait,
+  addNodeToDefaultCluster,
+} from "@cocalc/backend/conat/test/setup";
+import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
+import { randomId } from "@cocalc/conat/names";
+
+beforeAll(before);
+
+describe("ensure sticky state sync and use is working properly", () => {
+  let clients: any;
+  it("2-node cluster", async () => {
+    await addNodeToDefaultCluster();
+    expect(servers.length).toBe(2);
+    await waitForConsistentState(servers);
+    clients = servers.map((x) => x.client());
+  });
+
+  const count = 25;
+  const subs0: any[] = [];
+  const subs1: any[] = [];
+  it(`create ${count} distinct sticky subscriptions and send one message to each to create sticky routing state on servers[0]`, async () => {
+    clients.push(servers[0].client());
+    clients.push(servers[1].client());
+    for (let i = 0; i < count; i++) {
+      subs0.push(
+        await clients[1].subscribe(`subject.${i}.*`, {
+          queue: STICKY_QUEUE_GROUP,
+        }),
+      );
+      // wait so above subscription is known to *both* servers:
+      // @ts-ignore
+      await servers[0].waitForInterest(
+        `subject.${i}.0`,
+        5000,
+        clients[0].conn.id,
+      );
+      subs1.push(
+        await clients[0].subscribe(`subject.${i}.*`, {
+          queue: STICKY_QUEUE_GROUP,
+        }),
+      );
+      // publishing causes a choice to be made and saved on servers[0]
+      await clients[0].publish(`subject.${i}.foo`, "hello");
+      expect(servers[0].sticky[`subject.${i}.*`]).not.toBe(undefined);
+      // but no choice on servers[1]
+      expect(servers[1].sticky[`subject.${i}.*`]).toBe(undefined);
+    }
+  });
+
+  let chosen;
+  it("see which subscription got chosen for subject.0.* -- this is useful later", async () => {
+    const p0 = async () => {
+      await subs0[0].next();
+      return 0;
+    };
+    const p1 = async () => {
+      await subs1[0].next();
+      return 1;
+    };
+    chosen = await Promise.race([p0(), p1()]);
+  });
+
+  it(`sticky on servers[0] should have ${count} entries starting in "subject".`, async () => {
+    const v = Object.keys(servers[0].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(count);
+  });
+
+  it(`sticky on servers[1] should have no entries starting in "subject".`, async () => {
+    const v = Object.keys(servers[1].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(0);
+  });
+
+  it(`servers[1]'s link to servers[0] should *eventually* have ${count} entries starting in "subject."`, async () => {
+    // @ts-ignore
+    const link = servers[1].clusterLinksByAddress[servers[0].address()];
+    let v;
+    await wait({
+      until: () => {
+        v = Object.keys(link.sticky).filter((s) => s.startsWith("subject."));
+        return v.length == count;
+      },
+    });
+    expect(v.length).toBe(count);
+  });
+
+  it("send message from clients[1] to each subject", async () => {
+    for (let i = 0; i < count; i++) {
+      await clients[1].publish(`subject.${i}.foo`);
+    }
+  });
+
+  it(`sticky on servers[1] should STILL have no entries starting in "subject", since no choices had to be made`, async () => {
+    const v = Object.keys(servers[1].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(0);
+  });
+
+  async function deliveryTest() {
+    const sub = chosen == 0 ? subs0[0] : subs1[0];
+
+    // clear up the subscription (we sent it stuff above)
+    const sentinel = randomId();
+    await clients[0].publish("subject.0.foo", sentinel);
+    while (true) {
+      const { value } = await sub.next();
+      if (value.data == sentinel) {
+        break;
+      }
+    }
+    for (const server of servers) {
+      // we randomize the last segment to verify that it is NOT used
+      // as input to the sticky routing choice.
+      const { count } = await server
+        .client()
+        .publish(`subject.0.${randomId()}`, "delivery-test");
+      expect(count).toBe(1);
+    }
+    const ids = new Set<string>();
+    for (let i = 0; i < servers.length; i++) {
+      // on of the subs will receive it and one will hang forever (which is fine)
+      const { value } = await sub.next();
+      expect(value.data).toBe("delivery-test");
+      ids.add(value.client.id);
+    }
+    // all messages must go to the SAME subscriber, since sticky
+    expect(ids.size).toBe(1);
+  }
+
+  it("publish from every node to subject.0.foo", deliveryTest);
+
+  const count2 = 5;
+  it(`add ${count2} more nodes to the cluster should be reaonably fast and not blow up in a feedback loop`, async () => {
+    for (let i = 0; i < count2; i++) {
+      await addNodeToDefaultCluster();
+    }
+  });
+
+  it("wait until cluster is consistent", async () => {
+    await waitForConsistentState(servers);
+  });
+
+  it("double check the links have the sticky state", () => {
+    for (const server of servers.slice(1)) {
+      // @ts-ignore
+      const link = server.clusterLinksByAddress[servers[0].address()];
+      const v = Object.keys(link.sticky).filter((s) =>
+        s.startsWith("subject."),
+      );
+      expect(v.length).toBe(count);
+    }
+  });
+
+  it(
+    "in bigger, cluster, publish from every node to subject.0.foo",
+    deliveryTest,
+  );
+
+  it("listen on > and note that it doesn't impact the count", async () => {
+    const sub = await clients[0].subscribe(">");
+    for (let i = 0; i < servers.length; i++) {
+      const { count } = await servers[i]
+        .client()
+        .publish("subject.0.foo", "hi");
+      expect(count).toBe(1);
+    }
+    sub.close();
+  });
+
+  it("unjoining servers[0] from servers[1] should transfer the sticky state to servers[1]", async () => {
+    await servers[1].unjoin({ address: servers[0].address() });
+    const v = Object.keys(servers[1].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(count);
+  });
+});
+
+afterAll(after);

src/packages/backend/conat/test/persist/multiple-servers.test.ts

@@ -4,7 +4,7 @@ Unit tests of multiple persist servers at once:
 - making numerous distinct clients and seeing that they are distributed across persist servers
 - stopping a persist server and seeing that failover happens without data loss
 
-pnpm test `pwd`/multiple-servers.test.ts 
+pnpm test `pwd`/multiple-servers.test.ts
 
 */
 

src/packages/backend/conat/test/socket/basic.test.ts

@@ -1,6 +1,6 @@
 /*
 
-pnpm test `pwd`/basic.test.ts 
+pnpm test `pwd`/basic.test.ts
 
 */
 

src/packages/backend/conat/test/socket/restarts.test.ts

@@ -18,8 +18,6 @@ beforeAll(async () => {
   setDefaultTimeouts({ request: 500, publish: 500 });
 });
 
-//jest.setTimeout(25000);
-
 describe("create a client and server and socket, verify it works, restart conat server, then confirm that socket still works", () => {
   const SUBJECT = "reconnect.one";
 

src/packages/backend/package.json

@@ -16,9 +16,8 @@
     "clean": "rm -rf dist node_modules",
     "build": "pnpm exec tsc --build",
     "tsc": "pnpm exec tsc --watch  --pretty --preserveWatchOutput",
-    "test": "pnpm exec jest --forceExit --maxWorkers=25%",
+    "test": "pnpm exec jest --forceExit",
     "test-conat": " pnpm exec jest --forceExit conat",
-    "stress-test": " pnpm exec jest --forceExit",
     "testp": "pnpm exec jest --forceExit",
     "depcheck": "pnpx depcheck --ignores events",
     "prepublishOnly": "pnpm test",

src/packages/conat/core/client.ts

@@ -22,6 +22,22 @@ much better in so many ways:
      - JsonCodec: uses JSON.stringify and TextEncoder.  This does not work
        with Buffer or Date and is less compact, but can be very fast.
 
+- One BIG DIFFERENCE from Nats is that when a message is sent the sender
+  can optionally find out how many clients received it.  They can also wait
+  until there is interest and send the message again.  This is automated so
+  it's very easy to use, and it makes writing distributed services without
+  race conditions making them broken temporarily much easier.  HOWEVER,
+  there is one caveat -- if as an admin you create a "tap", i.e., you
+  subscribe to all messages matching some pattern just to see what's going
+  on, then currently that counts in the delivery count and interest, and that
+  would then cause these race conditions to happen again. E.g., a user
+  signs in, subscribes to their INBOX, sends a request, and gets a response
+  to that inbox, but does this all quickly, and in a cluster, the server doesn't
+  see the inbox yet, so it fails.  As a workaround, subscriptions to the
+  subject pattern '>' are invisible, so you can always tap into '>' for debugging
+  purposes.  TODO: implement a general way of making an invisible subscriber that
+  doesn't count.
+
 
 THE CORE API
 
@@ -604,13 +620,14 @@ export class Client extends EventEmitter {
       );
     }
     timeout = Math.min(timeout, MAX_INTEREST_TIMEOUT);
-    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 });
+    try {
+      const response = await this.conn
+        .timeout(timeout ? timeout : 10000)
+        .emitWithAck("wait-for-interest", { subject, timeout });
+      return response;
+    } catch (err) {
+      throw toConatError(err);
     }
-    return response;
   };
 
   recvStats = (bytes: number) => {
@@ -954,7 +971,7 @@ export class Client extends EventEmitter {
             });
           }
         } catch (err) {
-          throw new ConatError(`${err}`, { code: 408 });
+          throw toConatError(err);
         }
         if (response?.error) {
           throw new ConatError(response.error, { code: response.code });
@@ -1300,9 +1317,7 @@ export class Client extends EventEmitter {
                 return response;
               }
             } catch (err) {
-              throw new ConatError(`timeout - ${subject} - ${err}`, {
-                code: 408,
-              });
+              throw toConatError(err);
             }
           } else {
             return await this.conn.emitWithAck("publish", v);
@@ -1910,3 +1925,15 @@ function isEmpty(obj: object): boolean {
   }
   return true;
 }
+
+function toConatError(socketIoError) {
+  // only errors are "disconnected" and a timeout
+  const e = `${socketIoError}`;
+  if (e.includes("disconnected")) {
+    return e;
+  } else {
+    return new ConatError(`timeout - ${e}`, {
+      code: 408,
+    });
+  }
+}

src/packages/conat/core/cluster.ts

@@ -18,7 +18,6 @@ const logger = getLogger("conat:core:cluster");
 export async function clusterLink(
   address: string,
   systemAccountPassword: string,
-  updateStickyLocal: (sticky: StickyUpdate) => void,
   timeout = CREATE_LINK_TIMEOUT,
 ) {
   const client = connect({ address, systemAccountPassword });
@@ -43,13 +42,7 @@ export async function clusterLink(
   if (!clusterName) {
     throw Error("clusterName must be specified");
   }
-  const link = new ClusterLink(
-    client,
-    id,
-    clusterName,
-    address,
-    updateStickyLocal,
-  );
+  const link = new ClusterLink(client, id, clusterName, address);
   await link.init();
   return link;
 }
@@ -61,7 +54,7 @@ export { type ClusterLink };
 
 class ClusterLink {
   public interest: Interest = new Patterns();
-  private sticky: Sticky = {};
+  public sticky: Sticky = {};
   private streams: ClusterStreams;
   private state: "init" | "ready" | "closed" = "init";
   private clientStateChanged = Date.now(); // when client status last changed
@@ -71,7 +64,6 @@ class ClusterLink {
     public readonly id: string,
     public readonly clusterName: string,
     public readonly address: string,
-    private readonly updateStickyLocal: (sticky: StickyUpdate) => void,
   ) {
     if (!client) {
       throw Error("client must be specified");
@@ -119,7 +111,6 @@ class ClusterLink {
 
   handleStickyUpdate = (update: StickyUpdate) => {
     updateSticky(update, this.sticky);
-    this.updateStickyLocal(update);
   };
 
   private handleClientStateChanged = () => {