conat tests -- refactoring and improving

Author: williamstein

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

@@ -562,95 +562,4 @@ describe("test automatic node discovery", () => {
   });
 });
 
-describe("test automatic node discovery (and forgetting)", () => {
-  const nodes: { client; server }[] = [];
-  const clusterName = "auto";
-  const create = async (id) => {
-    nodes.push(
-      await createClusterNode({
-        id,
-        clusterName,
-        autoscanInterval: 50,
-        longAutoscanInterval: 6000,
-        forgetClusterNodeInterval: 500, // make automatic forgetting short so we can test.
-      }),
-    );
-  };
-
-  it("create three distinct servers with cluster support enabled", async () => {
-    await create("node0");
-    await create("node1");
-  });
-
-  it("connect 0 -> 1 and see other link get automatically added", async () => {
-    expect(nodes[0].server.clusterAddresses(clusterName).length).toBe(1);
-    await nodes[0].server.join(nodes[1].server.address());
-    expect(nodes[0].server.clusterAddresses(clusterName).length).toBe(2);
-    expect(nodes[1].server.clusterAddresses(clusterName).length).toBe(1);
-    await wait({
-      until: () => {
-        return nodes[1].server.clusterAddresses(clusterName).length == 2;
-      },
-    });
-  });
-
-  it("make a new node and a connection 2 -> 1 and observe cluster gets completed automatically", async () => {
-    await create("node2");
-    await nodes[2].server.join(nodes[1].server.address());
-    // node0 and node1 don't instantly know node2
-    expect(nodes[0].server.clusterAddresses(clusterName).length).toBe(2);
-    expect(nodes[1].server.clusterAddresses(clusterName).length).toBe(2);
-    expect(nodes[2].server.clusterAddresses(clusterName).length).toBe(2);
-    // but soon they will all know each other
-    await wait({
-      until: () => {
-        return (
-          nodes[0].server.clusterAddresses(clusterName).length == 3 &&
-          nodes[1].server.clusterAddresses(clusterName).length == 3 &&
-          nodes[2].server.clusterAddresses(clusterName).length == 3
-        );
-      },
-    });
-  });
-
-  const count = 5;
-  it(`add ${count} more nodes`, async () => {
-    for (let i = 3; i < 3 + count; i++) {
-      await create(`node${i}`);
-      await nodes[i].server.join(nodes[i - 1].server.address());
-    }
-    const total = nodes.length;
-    await wait({
-      until: () => {
-        for (let i = 0; i < total; i++) {
-          if (nodes[i].server.clusterAddresses(clusterName).length != total) {
-            return false;
-          }
-        }
-        return true;
-      },
-    });
-  });
-
-  it("close nodes[1], run scan, and observe that nodes[1] is forgotten", async () => {
-    const numNodes = () => {
-      return Object.keys(
-        nodes[0].server.clusterLinks[nodes[0].server.clusterName],
-      ).length;
-    };
-    const n = numNodes();
-    nodes[1].server.close();
-    expect(nodes[1].server.isHealthy()).toBe(false);
-    // not instantly gone
-    expect(numNodes()).toBe(n);
-    await nodes[0].server.scan();
-    // still not gone
-    expect(numNodes()).toBe(n);
-    // wait a second and scan, and it must be gone (because we set the interval very short)
-    await delay(1000);
-    await nodes[0].server.scan();
-    expect(numNodes()).toBe(n - 1);
-  });
-});
-
 afterAll(after);

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

@@ -1,4 +1,9 @@
-import { before, after, wait } from "@cocalc/backend/conat/test/setup";
+import {
+  before,
+  after,
+  wait,
+  waitForConsistentState,
+} from "@cocalc/backend/conat/test/setup";
 import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
 import { type ConatServer } from "@cocalc/conat/core/server";
 import { createClusterNode } from "./util";
@@ -54,6 +59,12 @@ describe("create cluster of two nodes, and verify that *sticky* subs properly wo
   });
 
   it("send messages and note they all go to the same target -- next the hard case across the cluster", async () => {
+    // NOTE: if we do this test without waiting for consistent state, it will definitely
+    // fail sometimes, since server2 literally doesn't know enough about the servers yet, 
+    // so has to make a different choice.  Services of course must account for the fact that
+    // for the first moments of their existence, sticky routing can't work.
+    await waitForConsistentState([server1, server2]);
+    
     await client2.waitForInterest(subject);
     for (let i = 0; i < count; i++) {
       await client2.publish(subject, "hi");

src/packages/backend/conat/test/cluster/node-discovery.test.ts

@@ -0,0 +1,107 @@
+import {
+  before,
+  after,
+  delay,
+  wait,
+  waitForConsistentState,
+} from "@cocalc/backend/conat/test/setup";
+import { createClusterNode } from "./util";
+
+beforeAll(before);
+
+describe("test automatic node discovery (and forgetting)", () => {
+  const nodes: { client; server }[] = [];
+  const clusterName = "auto";
+  const create = async (id) => {
+    nodes.push(
+      await createClusterNode({
+        id,
+        clusterName,
+        autoscanInterval: 50,
+        longAutoscanInterval: 6000,
+        forgetClusterNodeInterval: 500, // make automatic forgetting short so we can test.
+      }),
+    );
+  };
+
+  it("create three distinct servers with cluster support enabled", async () => {
+    await create("node0");
+    await create("node1");
+  });
+
+  it("connect 0 -> 1 and see other link get automatically added", async () => {
+    expect(nodes[0].server.clusterAddresses(clusterName).length).toBe(1);
+    await nodes[0].server.join(nodes[1].server.address());
+    expect(nodes[0].server.clusterAddresses(clusterName).length).toBe(2);
+    expect(nodes[1].server.clusterAddresses(clusterName).length).toBe(1);
+    await wait({
+      until: () => {
+        return nodes[1].server.clusterAddresses(clusterName).length == 2;
+      },
+    });
+  });
+
+  it("make a new node and a connection 2 -> 1 and observe cluster gets completed automatically", async () => {
+    await create("node2");
+    await nodes[2].server.join(nodes[1].server.address());
+    // node0 and node1 don't instantly know node2
+    expect(nodes[0].server.clusterAddresses(clusterName).length).toBe(2);
+    expect(nodes[1].server.clusterAddresses(clusterName).length).toBe(2);
+    expect(nodes[2].server.clusterAddresses(clusterName).length).toBe(2);
+    // but soon they will all know each other
+    await wait({
+      until: () => {
+        return (
+          nodes[0].server.clusterAddresses(clusterName).length == 3 &&
+          nodes[1].server.clusterAddresses(clusterName).length == 3 &&
+          nodes[2].server.clusterAddresses(clusterName).length == 3
+        );
+      },
+    });
+  });
+
+  const count = 5;
+  it(`add ${count} more nodes`, async () => {
+    for (let i = 3; i < 3 + count; i++) {
+      await create(`node${i}`);
+      await nodes[i].server.join(nodes[i - 1].server.address());
+    }
+    const total = nodes.length;
+    await wait({
+      until: () => {
+        for (let i = 0; i < total; i++) {
+          if (nodes[i].server.clusterAddresses(clusterName).length != total) {
+            return false;
+          }
+        }
+        return true;
+      },
+    });
+  });
+
+  it("check state is consistent", async () => {
+    await waitForConsistentState(nodes.map((x) => x.server));
+  });
+
+  it("close nodes[1], run scan, and observe that nodes[1] is forgotten", async () => {
+    const numNodes = () => {
+      return Object.keys(
+        nodes[0].server.clusterLinks[nodes[0].server.clusterName],
+      ).length;
+    };
+    const n = numNodes();
+    nodes[1].server.close();
+    expect(nodes[1].server.isHealthy()).toBe(false);
+    // not instantly gone
+    expect(numNodes()).toBe(n);
+    await nodes[0].server.scan();
+    // still not gone
+    expect(numNodes()).toBe(n);
+    // wait a second and scan, and it must be gone (because we set the interval very short)
+    await delay(1000);
+    await nodes[0].server.scan();
+    expect(numNodes()).toBe(n - 1);
+  });
+});
+
+afterAll(after);

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

@@ -40,7 +40,7 @@ describe("test using multiple persist servers in a cluster", () => {
   });
 
   it("wait until both servers in the cluster have the same state", async () => {
-    await waitForConsistentState([server, server1], 5000);
+    await waitForConsistentState([server, server1]);
   });
 
   const openStreamsConnectedToBothServers0: any[] = [];

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

@@ -24,14 +24,14 @@ import {
 import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
 import { waitForSubscription } from "./util";
 
-// should be several thousand, so 250 seems reasonable as a cutoff to indicate
-// things are horribly wrong
-const REQUIRED_SINGLE_SERVER_RECV_MESSAGES_PER_SECOND = 250;
-// should be tens of thousands
-const REQUIRED_SINGLE_SERVER_SEND_MESSAGES_PER_SECOND = 500;
+// These are all very low since results are very low when running
+// tests massively in parallel.
 
-const REQUIRED_CLUSTER_SERVER_RECV_MESSAGES_PER_SECOND = 200;
-const REQUIRED_CLUSTER_SERVER_SEND_MESSAGES_PER_SECOND = 400;
+const REQUIRED_SINGLE_SERVER_RECV_MESSAGES_PER_SECOND = 150;
+const REQUIRED_SINGLE_SERVER_SEND_MESSAGES_PER_SECOND = 200;
+
+const REQUIRED_CLUSTER_SERVER_RECV_MESSAGES_PER_SECOND = 100;
+const REQUIRED_CLUSTER_SERVER_SEND_MESSAGES_PER_SECOND = 200;
 
 const VERBOSE = false;
 const log = VERBOSE ? console.log : (..._args) => {};

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

@@ -145,7 +145,7 @@ export function connect(opts?): Client {
 // other thinks it is.
 export async function waitForConsistentState(
   servers: ConatServer[],
-  timeout = 10000,
+  timeout = 20000,
 ): Promise<void> {
   if (servers.length <= 1) {
     return;

src/packages/backend/conat/test/sync/astream.test.ts

@@ -93,7 +93,7 @@ describe("test basics with an astream", () => {
   });
 });
 
-const stress1 = 1e4;
+const stress1 = 1e3;
 describe(`stress test -- write, then read back, ${stress1} messages`, () => {
   let client, s;
   const name = "stress-test";