work in progress fixing a very subtle stream bug

Author: williamstein

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

@@ -15,6 +15,7 @@ import {
   persistServer as persistServer0,
   wait,
   setDefaultTimeouts,
+  setDefaultSocketTimeouts,
   waitForConsistentState,
 } from "../setup";
 import { uuid } from "@cocalc/util/misc";
@@ -25,9 +26,14 @@ 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,
+  });
 });
 
-jest.setTimeout(15000);
+jest.setTimeout(10000);
 describe("test using multiple persist servers in a cluster", () => {
   let client0, server1, client1;
   it("add another node", async () => {
@@ -142,25 +148,24 @@ describe("test using multiple persist servers in a cluster", () => {
   });
 
   it("remove one persist server", async () => {
+    openStreams1.map((x) => (x.stream.log = true));
+    console.log("CLOSING PERSIST SERVER");
     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",
-        sync: true,
-      });
-      expect(await s.getAll()).toEqual([project_id]);
-      s.close();
-    }
-    expect(Object.keys(persistServer1.sockets).length).toEqual(0);
-  });
+  //   it.skip("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]);
+  //       s.close();
+  //     }
+  //     expect(Object.keys(persistServer1.sockets).length).toEqual(0);
+  //   });
 
   // this can definitely take a long time (e.g., ~10s), as it involves automatic failover.
   it("Checks automatic failover works:  the streams connected to both servers we created above must keep working, despite at least one of them having its persist server get closed.", async () => {
@@ -169,9 +174,29 @@ 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 });
+      console.log(i, stream1.stream.storage);
+      await wait({
+        until: async () => {
+          console.log(
+            i,
+            stream1.stream.client.id,
+            stream1.stream.id,
+            stream1.getAll(),
+            stream1.messages,
+            (await stream1.stream.persistClient.getAll({ start_seq: 0 }))
+              .length,
+            stream1.stream.messages,
+            stream1.stream.raw,
+          );
+          return stream1.length >= 2;
+        },
+        timeout: 10000,
+        start: 1000,
+      });
       expect(stream1.length).toBe(2);
     }
   });

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

@@ -15,6 +15,7 @@ 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";
 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,8 @@ import { EventEmitter } from "events";
 import { getLogger } from "@cocalc/conat/client";
 import { until } from "@cocalc/util/async-utils";
 
+const RECONNECT_DELAY = 1000;
+
 interface GetAllOpts {
   start_seq?: number;
   end_seq?: number;
@@ -46,6 +48,7 @@ class PersistStreamClient extends EventEmitter {
   private reconnecting = false;
   private gettingMissed = false;
   private changesWhenGettingMissed: ChangefeedEvent[] = [];
+  id = Math.random();
 
   constructor(
     private client: Client,
@@ -61,6 +64,14 @@ class PersistStreamClient extends EventEmitter {
   }
 
   private init = () => {
+    if (this.reconnecting) {
+      console.log(
+        this.id,
+        "persist client reconnecting",
+        this.client.id,
+        this.storage.path,
+      );
+    }
     if (this.client.state == "closed") {
       this.close();
       return;
@@ -91,15 +102,17 @@ class PersistStreamClient extends EventEmitter {
       // console.log("persist client was disconnected", this.storage.path);
       this.reconnecting = true;
       this.socket.removeAllListeners();
-      setTimeout(this.init, 1000);
+      setTimeout(this.init, RECONNECT_DELAY);
     });
     this.socket.once("closed", () => {
       this.reconnecting = true;
       this.socket.removeAllListeners();
-      setTimeout(this.init, 1000);
+      setTimeout(this.init, RECONNECT_DELAY);
     });
 
     this.socket.on("data", (updates, headers) => {
+      if (this.storage.path.endsWith("foo"))
+        console.log(this.id, "data", updates, headers);
       if (updates == null && headers != null) {
         // has to be an error
         this.emit(
@@ -131,6 +144,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,

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";