conat cluster: timeout adding new node to cluster

- before it would just keep trying to connect to a nonexisting node
  forever.  That didn't break anything, but was annoying in the logs.

Author: williamstein

src/packages/backend/conat/test/cluster/add-link-timeout.test.ts

@@ -0,0 +1,20 @@
+import { before, after } from "@cocalc/backend/conat/test/setup";
+import { createClusterNode } from "./util";
+import getPort from "@cocalc/backend/get-port";
+
+beforeAll(before);
+
+describe("adding a node times out if we can't connect to it, rather than trying forever", () => {
+  it("tries to add a link to a node that doesn't exist", async () => {
+    const { server } = await createClusterNode({
+      clusterName: "cluster0",
+      id: "1",
+    });
+    const port = await getPort();
+    expect(async () => {
+      await server.join(`localhost:${port}`, { timeout: 500 });
+    }).rejects.toThrow("timeout");
+  });
+});
+
+afterAll(after);

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

@@ -41,7 +41,7 @@ describe("using various sync data structures with a cluster", () => {
   });
 
   it("wait until both servers in the cluster have the same state", async () => {
-    await waitForConsistentState([server, server2], 5000);
+    await waitForConsistentState([server, server2], 15000);
   });
 
   let dstream2;

src/packages/backend/package.json

@@ -17,7 +17,7 @@
     "build": "pnpm exec tsc --build",
     "tsc": "pnpm exec tsc --watch  --pretty --preserveWatchOutput",
     "test": "pnpm exec jest --forceExit --maxWorkers=25%",
-    "test-conat": " pnpm exec jest --forceExit --maxWorkers=50% conat",
+    "test-conat": " pnpm exec jest --forceExit conat",
     "stress-test": " pnpm exec jest --forceExit",
     "testp": "pnpm exec jest --forceExit",
     "depcheck": "pnpx depcheck --ignores events",

src/packages/conat/core/client.ts

@@ -546,27 +546,30 @@ export class Client extends EventEmitter {
     setTimeout(() => this.conn.io.disconnect(), 1);
   };
 
-  waitUntilSignedIn = reuseInFlight(async () => {
-    // not "signed in" if --
-    //   - not connected, or
-    //   - no info at all (which gets sent on sign in)
-    //   - or the user is {error:....}, which is what happens when sign in fails
-    //     e.g., do to an expired cookie
-    if (
-      this.info == null ||
-      this.state != "connected" ||
-      this.info?.user?.error
-    ) {
-      await once(this, "info");
-    }
-    if (
-      this.info == null ||
-      this.state != "connected" ||
-      this.info?.user?.error
-    ) {
-      throw Error("failed to sign in");
-    }
-  });
+  // this has NO timeout by default
+  waitUntilSignedIn = reuseInFlight(
+    async ({ timeout }: { timeout?: number } = {}) => {
+      // not "signed in" if --
+      //   - not connected, or
+      //   - no info at all (which gets sent on sign in)
+      //   - or the user is {error:....}, which is what happens when sign in fails
+      //     e.g., do to an expired cookie
+      if (
+        this.info == null ||
+        this.state != "connected" ||
+        this.info?.user?.error
+      ) {
+        await once(this, "info", timeout);
+      }
+      if (
+        this.info == null ||
+        this.state != "connected" ||
+        this.info?.user?.error
+      ) {
+        throw Error("failed to sign in");
+      }
+    },
+  );
 
   private statsLoop = async () => {
     await until(

src/packages/conat/core/cluster.ts

@@ -11,18 +11,30 @@ import { once } from "@cocalc/util/async-utils";
 import { server as createPersistServer } from "@cocalc/conat/persist/server";
 import { getLogger } from "@cocalc/conat/client";
 import { hash_string } from "@cocalc/util/misc";
+const CREATE_LINK_TIMEOUT = 45_000;
 
 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 });
   if (client.info == null) {
-    await client.waitUntilSignedIn();
-    if (client.info == null) throw Error("bug -- failed to sign in");
+    try {
+      await client.waitUntilSignedIn({
+        timeout: timeout ?? CREATE_LINK_TIMEOUT,
+      });
+    } catch (err) {
+      client.close();
+      throw err;
+    }
+    if (client.info == null) {
+      // this is impossible
+      throw Error("BUG -- failed to sign in");
+    }
   }
   const { id, clusterName } = client.info;
   if (!id) {

src/packages/conat/core/server.ts

@@ -1103,36 +1103,46 @@ export class ConatServer extends EventEmitter {
   // - the address obviously must be reachable over the network
   // - the systemAccountPassword of this node and the one with the given
   //   address must be the same.
-  join = reuseInFlight(async (address: string): Promise<ClusterLink> => {
-    if (!this.options.systemAccountPassword) {
-      throw Error("systemAccountPassword must be set");
-    }
-    logger.debug("join: connecting to ", address);
-    const link0 = this.clusterLinksByAddress[address];
-    if (link0 != null) {
-      logger.debug("join: already connected to ", address);
-      return link0;
-    }
-    try {
-      const link = await clusterLink(
-        address,
-        this.options.systemAccountPassword,
-        this.updateSticky,
-      );
-      const { clusterName, id } = link;
-      if (this.clusterLinks[clusterName] == null) {
-        this.clusterLinks[clusterName] = {};
+  join = reuseInFlight(
+    async (
+      address: string,
+      { timeout }: { timeout?: number } = {},
+    ): Promise<ClusterLink> => {
+      if (!this.options.systemAccountPassword) {
+        throw Error("systemAccountPassword must be set");
       }
-      this.clusterLinks[clusterName][id] = link;
-      this.clusterLinksByAddress[address] = link;
-      this.scanSoon();
-      logger.debug("join: successfully created new connection to ", address);
-      return link;
-    } catch (err) {
-      logger.debug("join: FAILED creating a new connection to ", address, err);
-      throw err;
-    }
-  });
+      logger.debug("join: connecting to ", address);
+      const link0 = this.clusterLinksByAddress[address];
+      if (link0 != null) {
+        logger.debug("join: already connected to ", address);
+        return link0;
+      }
+      try {
+        const link = await clusterLink(
+          address,
+          this.options.systemAccountPassword,
+          this.updateSticky,
+          timeout,
+        );
+        const { clusterName, id } = link;
+        if (this.clusterLinks[clusterName] == null) {
+          this.clusterLinks[clusterName] = {};
+        }
+        this.clusterLinks[clusterName][id] = link;
+        this.clusterLinksByAddress[address] = link;
+        this.scanSoon();
+        logger.debug("join: successfully created new connection to ", address);
+        return link;
+      } catch (err) {
+        logger.debug(
+          "join: FAILED creating a new connection to ",
+          address,
+          err,
+        );
+        throw err;
+      }
+    },
+  );
 
   unjoin = ({
     id,

src/packages/util/async-utils.ts

@@ -181,9 +181,13 @@ export class TimeoutError extends Error {
 export async function once(
   obj: EventEmitter,
   event: string,
-  timeout_ms: number = 0,
+  timeout_ms: number | undefined = 0,
 ): Promise<any> {
   if (obj == null) throw Error("once -- obj is undefined");
+  if (timeout_ms == null) {
+    // clients might explicitly pass in undefined, but below we expect 0 to mean "no timeout"
+    timeout_ms = 0;
+  }
   if (typeof obj.once != "function")
     throw Error("once -- obj.once must be a function");
 
@@ -209,7 +213,9 @@ export async function once(
     function onTimeout() {
       cleanup();
       reject(
-        new TimeoutError(`once: timeout of ${timeout_ms}ms waiting for "${event}"`),
+        new TimeoutError(
+          `once: timeout of ${timeout_ms}ms waiting for "${event}"`,
+        ),
       );
     }