improver cluster unit testing; more testing of consistent state

Author: williamstein

src/packages/backend/conat/test/cluster/pubsub.test.ts

@@ -9,6 +9,8 @@ import {
   addNodeToDefaultCluster,
   wait,
   delay,
+  waitForConsistentState,
+  defaultCluster,
 } from "@cocalc/backend/conat/test/setup";
 import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
 
@@ -79,7 +81,7 @@ describe("same basic test, but in the other direction", () => {
   });
 });
 
-describe("with three nodes and two subscribers with different queue group on distinct nodes", () => {
+describe("three nodes and two subscribers with different queue group on distinct nodes", () => {
   let client0, client1, client2, server2;
   it("get the clients", async () => {
     client0 = server.client();
@@ -116,7 +118,7 @@ describe("with three nodes and two subscribers with different queue group on dis
     expect(count).toBe(1);
   });
 
-  it("make a sticky sub from all nodes and see that all messages go to the same receiver, no matter which node we send from", async () => {
+  it("make sticky subs from each node, and see that all messages go to the same receiver, no matter which node we send from", async () => {
     const sub0 = await client0.subscribe("sticky", {
       queue: STICKY_QUEUE_GROUP,
     });
@@ -150,20 +152,38 @@ describe("with three nodes and two subscribers with different queue group on dis
     await client2.waitForInterest("sticky");
 
     for (let i = 0; i < 20; i++) {
-      const { count } = await client2.publish("sticky");
+      const { count } = await client2.publish("sticky", null);
       expect(count).toBe(1);
     }
     for (let i = 0; i < 20; i++) {
-      const { count } = await client1.publish("sticky");
+      const { count } = await client1.publish("sticky", null);
       expect(count).toBe(1);
     }
     for (let i = 0; i < 20; i++) {
-      const { count } = await client0.publish("sticky");
+      const { count } = await client0.publish("sticky", null);
       expect(count).toBe(1);
     }
+    await wait({ until: () => n0 + n1 + n2 == 60 });
     expect(n0 * n1).toBe(0);
     expect(n0 * n2).toBe(0);
     expect(n1 * n2).toBe(0);
+    expect(n0 + n1 + n2).toBe(60);
+
+    await waitForConsistentState(defaultCluster);
+
+    // add another node and test
+    const server3 = await addNodeToDefaultCluster();
+    const client3 = server3.client();
+    await waitForConsistentState(defaultCluster);
+    for (let i = 0; i < 20; i++) {
+      const { count } = await client3.publish("sticky", null);
+      expect(count).toBe(1);
+    }
+    await wait({ until: () => n0 + n1 + n2 == 80 });
+    expect(n0 * n1).toBe(0);
+    expect(n0 * n2).toBe(0);
+    expect(n1 * n2).toBe(0);
+    expect(n0 + n1 + n2).toBe(80);
   });
 });
 

src/packages/backend/conat/test/persist/cluster.test.ts

@@ -138,13 +138,13 @@ describe("test using multiple persist servers in a cluster", () => {
       expect(t.getAll()).toEqual(["x"]);
       openStreams1.push(t);
     }
-    expect(openStreams0.length).toBeGreaterThan(1);
+    expect(openStreams0.length).toBeGreaterThan(0);
   });
 
   it("remove one persist server", async () => {
     persistServer1.close();
     // [ ] TODO: removing this delay leads to very consistent failures
-    // involving data loss, but things should just be slower, never broken, 
+    // involving data loss, but things should just be slower, never broken,
     // on automatic failover.
     await delay(3000);
   });

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

@@ -36,6 +36,9 @@ export async function initConatServer(
   }
 
   const server = createConatServer(options);
+  if (server.clusterName == "default") {
+    defaultCluster.push(server);
+  }
   if (server.state != "ready") {
     await once(server, "ready");
   }
@@ -64,6 +67,8 @@ export async function createServer(opts?) {
 
 // add another node to the cluster -- this is still in the same process (not forked), which
 // is generally good since you can console.log from it, faster, etc.
+// this does connect the new node to all existing nodes.
+export const defaultCluster: ConatServer[] = [];
 export async function addNodeToDefaultCluster(): Promise<ConatServer> {
   const port = await getPort();
   const node = await initConatServer({
@@ -73,8 +78,10 @@ export async function addNodeToDefaultCluster(): Promise<ConatServer> {
     id: getNodeId(),
     systemAccountPassword: "secret",
   });
-  await server.join(node.address());
-  await node.join(server.address());
+  for (const s of defaultCluster) {
+    await s.join(node.address());
+    await node.join(s.address());
+  }
   return node;
 }
 
@@ -171,7 +178,9 @@ export async function waitForConsistentState(
   }
 
   if (ids.size != servers.length) {
-    throw Error("all servers must have distinct ids");
+    throw Error(
+      `all servers must have distinct ids -- ${JSON.stringify(servers.map((x) => x.id))}`,
+    );
   }
 
   const start = Date.now();

src/packages/conat/core/server.ts

@@ -187,7 +187,7 @@ export class ConatServer extends EventEmitter {
   } = {};
   private clusterLinksByAddress: { [address: string]: ClusterLink } = {};
   private clusterPersistServer?: ConatSocketServer;
-  private clusterName?: string;
+  public readonly clusterName?: string;
   private queuedClusterUpdates: Update[] = [];
 
   constructor(options: Options) {