conat core: changing sticky state to not be redundant

Author: williamstein

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,86 @@
+import {
+  before,
+  after,
+  defaultCluster as servers,
+  waitForConsistentState,
+  wait,
+  addNodeToDefaultCluster,
+} from "@cocalc/backend/conat/test/setup";
+import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
+
+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 = 5;
+  it(`create ${count} distinct sticky subscriptions and send one message to each to creat sticky routing state on servers[0]`, async () => {
+    clients.push(servers[0].client());
+    clients.push(servers[1].client());
+    for (let i = 0; i < count; i++) {
+      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,
+      );
+      await clients[0].subscribe(`subject.${i}.*`, {
+        queue: STICKY_QUEUE_GROUP,
+      });
+      // cause choice to be made and saved on servers[0]
+      await clients[0].publish(`subject.${i}.foo`);
+    }
+  });
+
+  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.skip(`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.skip(`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);
+  });
+});
+
+afterAll(after);

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/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 = () => {

src/packages/conat/core/server.ts

@@ -137,6 +137,8 @@ export interface Options {
   // if true, use https when creating an internal client.
   ssl?: boolean;
 
+  // WARNING: **superclusters are NOT fully iplemented yet.**
+  //
   // if clusterName is set, enable clustering. Each node
   // in the cluster must have a different name. systemAccountPassword
   // must also be set.  This only has an impact when the id is '0'.
@@ -456,8 +458,54 @@ export class ConatServer extends EventEmitter {
     }
   };
 
-  private getStickyTarget = ({ pattern, subject }) => {
-    return this.sticky[pattern]?.[subject];
+  private getStickyTarget = ({
+    pattern,
+    subject,
+    targets: targets0,
+  }: {
+    pattern: string;
+    subject: string;
+    targets: Set<string>; // the current valid choices as defined by subscribers known via interest graph
+  }) => {
+    if (!this.cluster || this.clusterName == null) {
+      return this.sticky[pattern]?.[subject];
+    }
+    const targets = new Set<string>();
+    const target = this.sticky[pattern]?.[subject];
+    if (target !== undefined && targets0.has(target)) {
+      targets.add(target);
+    }
+    // next check sticky state of other nodes in the cluster
+    const cluster = this.clusterLinks[this.clusterName];
+    for (const id in cluster) {
+      const target = cluster[id].sticky[pattern]?.[subject];
+      if (target !== undefined && targets0.has(target)) {
+        targets.add(target);
+      }
+    }
+    if (targets.size == 0) {
+      return undefined;
+    }
+    if (targets.size == 1) {
+      for (const target of targets) {
+        return target;
+      }
+    }
+
+    // problem: there are distinct mutually incompatible
+    // choices of targets.  This can only happen if at least
+    // two choices were made when the cluster was in an
+    // inconsistent state.
+    // We just take the first in alphabetical order.
+    // Since the sticky maps being used to make
+    // this list of targets is eventually consistent
+    // across the cluster, the same choice of target from
+    // those targets will be made by all nodes.
+    // The main problem with doing this is its slightly more
+    // effort.  The main advantage is that no communication
+    // or coordination between nodes is needed to "fix or agree
+    // on something", and that's a huge advantage!!
+    return Array.from(targets).sort()[0];
   };
 
   ///////////////////////////////////////
@@ -678,7 +726,7 @@ export class ConatServer extends EventEmitter {
     }
 
     //
-    // TODO: Supercluster routing.
+    // TODO: Supercluster routing.  NOT IMPLEMENTED YET
     //
     //     // if no matches in local cluster, try the supercluster (if there is one)
     //     if (count == 0) {
@@ -1122,7 +1170,6 @@ export class ConatServer extends EventEmitter {
         const link = await clusterLink(
           address,
           this.options.systemAccountPassword,
-          this.updateSticky,
           timeout,
         );
         const { clusterName, id } = link;

src/packages/conat/core/sticky.ts

@@ -1,4 +1,5 @@
 import ConsistentHash from "consistent-hash";
+import { hash_string } from "@cocalc/util/misc";
 
 export function consistentHashingChoice(
   v: Set<string>,
@@ -18,7 +19,10 @@ export function consistentHashingChoice(
   for (const x of w) {
     hr.add(x);
   }
-  return hr.get(resource);
+  // we hash the resource so that the values are randomly distributed even
+  // if the resources look very similar (e.g., subject.1, subject.2, etc.)
+  // I thought that "consistent-hash" hashed the resource, but it doesn't really.
+  return hr.get(hash_string(resource));
 }
 
 export function stickyChoice({
@@ -35,11 +39,12 @@ export function stickyChoice({
   getStickyTarget: (opts: {
     pattern: string;
     subject: string;
+    targets: Set<string>;
   }) => string | undefined;
 }) {
   const v = subject.split(".");
   subject = v.slice(0, v.length - 1).join(".");
-  const currentTarget = getStickyTarget({ pattern, subject });
+  const currentTarget = getStickyTarget({ pattern, subject, targets });
   if (currentTarget === undefined || !targets.has(currentTarget)) {
     // we use consistent hashing instead of random to make the choice, because if
     // choice is being made by two different socketio servers at the same time,