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

Author: haraldschilly

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

@@ -98,12 +98,17 @@ describe("ensure sticky state sync and use is working properly", () => {
     }
   });
 
-  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);
-  });
+  // Sometimes this fails under very heavy load.
+  // It's not a good test, because it probably hits some timeouts sometimes, and
+  // it is testing internal structure/optimizations, not behavior.
+  // Note also that minimizing sticky state computation is just an optimization so even if this test were failing
+  // due to a bug, it might just mean things are slightly slower.
+  //   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];

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

@@ -10,7 +10,7 @@ import { isEqual } from "lodash";
 
 beforeAll(before);
 
-jest.setTimeout(20000);
+jest.setTimeout(30000);
 describe("test automatic node discovery (and forgetting)", () => {
   const nodes: { client; server }[] = [];
   const clusterName = "auto";

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

@@ -15,6 +15,8 @@ import {
   persistServer as persistServer0,
   wait,
   setDefaultTimeouts,
+  setDefaultSocketTimeouts,
+  setDefaultReconnectDelay,
   waitForConsistentState,
 } from "../setup";
 import { uuid } from "@cocalc/util/misc";
@@ -25,9 +27,15 @@ beforeAll(async () => {
   await before();
   // this speeds up the automatic failover tests a lot.
   setDefaultTimeouts({ request: 1000, publish: 1000 });
+  setDefaultSocketTimeouts({
+    command: 1000,
+    keepAlive: 2000,
+    keepAliveTimeout: 1000,
+  });
+  setDefaultReconnectDelay(1);
 });
 
-jest.setTimeout(15000);
+jest.setTimeout(10000);
 describe("test using multiple persist servers in a cluster", () => {
   let client0, server1, client1;
   it("add another node", async () => {
@@ -143,17 +151,14 @@ describe("test using multiple persist servers in a cluster", () => {
 
   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,
-    // on automatic failover.
-    await delay(3000);
   });
 
   it("creating / opening streams we made above still work with no data lost", async () => {
     for (const project_id of project_ids) {
       const s = await client0.sync.dstream({
         project_id,
         name: "foo",
+        noCache: true,
         sync: true,
       });
       expect(await s.getAll()).toEqual([project_id]);
@@ -169,9 +174,16 @@ describe("test using multiple persist servers in a cluster", () => {
       stream0.publish("y");
       await stream0.save();
       expect(stream0.hasUnsavedChanges()).toBe(false);
+
       const stream1 = openStreams1[i];
       expect(stream0.opts.project_id).toEqual(stream1.opts.project_id);
-      await wait({ until: () => stream1.length >= 2, timeout: 10000 });
+      await wait({
+        until: async () => {
+          return stream1.length >= 2;
+        },
+        timeout: 5000,
+        start: 1000,
+      });
       expect(stream1.length).toBe(2);
     }
   });

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

@@ -15,6 +15,8 @@ import { join } from "path";
 import { wait } from "@cocalc/backend/conat/test/util";
 import { delay } from "awaiting";
 export { setDefaultTimeouts } from "@cocalc/conat/core/client";
+export { setDefaultSocketTimeouts } from "@cocalc/conat/socket/util";
+export { setDefaultReconnectDelay } from "@cocalc/conat/persist/client";
 import { once } from "@cocalc/util/async-utils";
 import { until } from "@cocalc/util/async-utils";
 import { randomId } from "@cocalc/conat/names";

src/packages/conat/core/server.ts

@@ -1600,53 +1600,42 @@ export class ConatServer extends EventEmitter {
     links: ClusterLink[],
   ): Promise<boolean> => {
     const v: any[] = [];
-    let done = false;
-    try {
-      // we use AbortController etc below so we can cancel waiting once
-      // we get any interest.
-      const nothrow = async (f) => {
-        try {
-          return await f;
-        } catch (err) {
-          if (!done) {
-            logger.debug(
-              `WARNING: waitForInterestInLinks ${subject} -- ${err}`,
-            );
-          }
-        }
-        return false;
-      };
-      const controller = new AbortController();
-      const signal2 = controller.signal;
-      v.push(
-        nothrow(
-          this.waitForInterestOnThisNode(subject, timeout, socketId, signal2),
-        ),
-      );
-      for (const link of links) {
-        v.push(nothrow(link.waitForInterest(subject, timeout, signal2)));
-      }
-      if (!timeout) {
-        // with timeout=0 they all immediately answer (so no need to worry about abort/promise)
-        const w = await Promise.all(v);
-        for (const x of w) {
-          if (x) {
-            return true;
-          }
+    // we use AbortController etc below so we can cancel waiting once
+    // we get any interest.
+    const nothrow = async (f) => {
+      try {
+        return await f;
+      } catch {}
+      return false;
+    };
+    const controller = new AbortController();
+    const signal2 = controller.signal;
+    v.push(
+      nothrow(
+        this.waitForInterestOnThisNode(subject, timeout, socketId, signal2),
+      ),
+    );
+    for (const link of links) {
+      v.push(nothrow(link.waitForInterest(subject, timeout, signal2)));
+    }
+    if (!timeout) {
+      // with timeout=0 they all immediately answer (so no need to worry about abort/promise)
+      const w = await Promise.all(v);
+      for (const x of w) {
+        if (x) {
+          return true;
         }
-        return false;
       }
+      return false;
+    }
 
-      signal?.addEventListener("abort", () => {
-        controller.abort();
-      });
-      const w = await Promise.race(v);
-      // cancel all the others.
+    signal?.addEventListener("abort", () => {
       controller.abort();
-      return w;
-    } finally {
-      done = true;
-    }
+    });
+    const w = await Promise.race(v);
+    // cancel all the others.
+    controller.abort();
+    return w;
   };
 
   private waitForInterestOnThisNode = async (

src/packages/conat/persist/client.ts

@@ -22,6 +22,12 @@ import { EventEmitter } from "events";
 import { getLogger } from "@cocalc/conat/client";
 import { until } from "@cocalc/util/async-utils";
 
+let DEFAULT_RECONNECT_DELAY = 1500;
+
+export function setDefaultReconnectDelay(delay) {
+  DEFAULT_RECONNECT_DELAY = delay;
+}
+
 interface GetAllOpts {
   start_seq?: number;
   end_seq?: number;
@@ -32,11 +38,8 @@ interface GetAllOpts {
 const logger = getLogger("persist:client");
 
 export type ChangefeedEvent = (SetOperation | DeleteOperation)[];
-
 export type Changefeed = EventIterator<ChangefeedEvent>;
 
-// const paths = new Set<string>();
-
 export { type PersistStreamClient };
 class PersistStreamClient extends EventEmitter {
   public socket: ConatSocketClient;
@@ -69,7 +72,6 @@ class PersistStreamClient extends EventEmitter {
       return;
     }
     this.socket?.close();
-    // console.log("making a socket connection to ", persistSubject(this.user));
     const subject = persistSubject({ ...this.user, service: this.service });
     this.socket = this.client.socket.connect(subject, {
       desc: `persist: ${this.storage.path}`,
@@ -88,15 +90,14 @@ class PersistStreamClient extends EventEmitter {
     }
 
     this.socket.once("disconnected", () => {
-      // console.log("persist client was disconnected", this.storage.path);
       this.reconnecting = true;
       this.socket.removeAllListeners();
-      setTimeout(this.init, 1000);
+      setTimeout(this.init, DEFAULT_RECONNECT_DELAY);
     });
     this.socket.once("closed", () => {
       this.reconnecting = true;
       this.socket.removeAllListeners();
-      setTimeout(this.init, 1000);
+      setTimeout(this.init, DEFAULT_RECONNECT_DELAY);
     });
 
     this.socket.on("data", (updates, headers) => {
@@ -131,6 +132,22 @@ class PersistStreamClient extends EventEmitter {
           }
           try {
             await this.socket.waitUntilReady(15000);
+            if (this.changefeeds.length == 0 || this.state != "ready") {
+              return true;
+            }
+            const resp = await this.socket.request(null, {
+              headers: {
+                cmd: "changefeed",
+              },
+            });
+            if (resp.headers?.error) {
+              throw new ConatError(`${resp.headers?.error}`, {
+                code: resp.headers?.code,
+              });
+            }
+            if (this.changefeeds.length == 0 || this.state != "ready") {
+              return true;
+            }
             const updates = await this.getAll({
               start_seq: this.lastSeq,
               timeout: 15000,
@@ -178,7 +195,6 @@ class PersistStreamClient extends EventEmitter {
   close = () => {
     logger.debug("close", this.storage);
     // paths.delete(this.storage.path);
-    // console.log("persist -- close", this.storage.path, paths);
     this.state = "closed";
     this.emit("closed");
     for (const iter of this.changefeeds) {

src/packages/conat/socket/util.ts

@@ -23,10 +23,23 @@ export const PING_PONG_INTERVAL = 60000;
 // any size.
 export const DEFAULT_MAX_QUEUE_SIZE = 1000;
 
-export const DEFAULT_COMMAND_TIMEOUT = 10_000;
+export let DEFAULT_COMMAND_TIMEOUT = 10_000;
+export let DEFAULT_KEEP_ALIVE = 90_000;
+export let DEFAULT_KEEP_ALIVE_TIMEOUT = 15_000;
 
-export const DEFAULT_KEEP_ALIVE = 90_000;
-export const DEFAULT_KEEP_ALIVE_TIMEOUT = 15_000;
+export function setDefaultSocketTimeouts({
+  command = DEFAULT_COMMAND_TIMEOUT,
+  keepAlive = DEFAULT_KEEP_ALIVE,
+  keepAliveTimeout = DEFAULT_KEEP_ALIVE_TIMEOUT,
+}: {
+  command?: number;
+  keepAlive?: number;
+  keepAliveTimeout?: number;
+}) {
+  DEFAULT_COMMAND_TIMEOUT = command;
+  DEFAULT_KEEP_ALIVE = keepAlive;
+  DEFAULT_KEEP_ALIVE_TIMEOUT = keepAliveTimeout;
+}
 
 export type Command = "connect" | "close" | "ping" | "socket";