fix critical bug in sticky state prop

Author: williamstein

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

@@ -128,7 +128,7 @@ export async function before() {
   tempDir = await mkdtemp(join(tmpdir(), "conat-test"));
   syncFiles.local = join(tempDir, "local");
   syncFiles.archive = join(tempDir, "archive");
-  
+
   server = await createServer();
   client = connect();
   persistServer = createPersistServer({ client });
@@ -184,40 +184,64 @@ export async function waitForConsistentState(
         // now look at everybody else's view of servers[i].
         // @ts-ignore
         const a = servers[i].interest.serialize().patterns;
+        const b = servers[i].sticky;
+        const hashServer = servers[i].hash();
         for (let j = 0; j < servers.length; j++) {
           if (i != j) {
             // @ts-ignore
             const link = servers[j].clusterLinks[clusterName]?.[servers[i].id];
             if (link == null) {
               throw Error(`node ${j} is not connected to node ${i}`);
             }
+            const hashLink = link.hash();
             const x = link.interest.serialize().patterns;
+            const y = link.sticky;
             const showInfo = () => {
-              console.log(
-                "server stream getAll: ",
-                // @ts-ignore
-                servers[i].clusterStreams.interest.stream.client.id,
-                // @ts-ignore
-                servers[i].clusterStreams.interest.stream.storage.path,
-                // @ts-ignore
-                servers[i].clusterStreams.interest.seqs(),
-                // @ts-ignore
-                //servers[i].clusterStreams.interest.getAll(),
-              );
-              console.log(
-                "link stream getAll: ",
-                // @ts-ignore
-                link.streams.interest.stream.client.id,
-                // @ts-ignore
-                link.streams.interest.stream.storage.path,
-                // @ts-ignore
-                link.streams.interest.seqs(),
-                // @ts-ignore
-                //link.streams.interest.getAll(),
-              );
-              console.log("waitForConsistentState", { i, j, a, x });
+              for (const type of ["interest", "sticky"]) {
+                console.log(
+                  `server stream ${type}: `,
+                  hashServer[type],
+                  // @ts-ignore
+                  servers[i].clusterStreams[type].stream.client.id,
+                  // @ts-ignore
+                  servers[i].clusterStreams[type].stream.storage.path,
+                  // @ts-ignore
+                  servers[i].clusterStreams[type].seqs(),
+                  // @ts-ignore
+                  //servers[i].clusterStreams.interest.getAll(),
+                );
+
+                console.log(
+                  `link stream ${type}: `,
+                  hashLink[type],
+                  // @ts-ignore
+                  link.streams[type].stream.client.id,
+                  // @ts-ignore
+                  link.streams[type].stream.storage.path,
+                  // @ts-ignore
+                  link.streams[type].seqs(),
+                  // @ts-ignore
+                  //link.streams.interest.getAll(),
+                );
+              }
+              console.log("waitForConsistentState", {
+                i,
+                j,
+                serverInterest: a,
+                linkInterest: x,
+                serverSticky: b,
+                linkSticky: y,
+              });
             };
-            if (!isEqual(a, x)) {
+            if (!isEqual(hashServer, hashLink)) {
+              if (Date.now() - start > 3000) {
+                console.log("hashes are not equal");
+                // likely going to fail
+                showInfo();
+              }
+              return false;
+            }
+            if (!isEqual(a, x) /*|| !isEqual(b, y) */) {
               // @ts-ignore
               const seqs0 = servers[i].clusterStreams.interest.seqs();
               const seqs1 = link.streams.interest.seqs();

src/packages/conat/core/cluster.ts

@@ -83,6 +83,9 @@ class ClusterLink {
     for (const update of this.streams.interest.getAll()) {
       updateInterest(update, this.interest, this.sticky);
     }
+    for (const update of this.streams.sticky.getAll()) {
+      updateSticky(update, this.sticky);
+    }
     // I have a slight concern about this because updates might not
     // arrive in order during automatic failover.  That said, maybe
     // automatic failover doesn't matter with these streams, since
@@ -168,6 +171,13 @@ class ClusterLink {
 
     return false;
   };
+
+  hash = (): { interest: number; sticky: number } => {
+    return {
+      interest: hashInterest(this.interest),
+      sticky: hashSticky(this.sticky),
+    };
+  };
 }
 
 function clusterStreamNames({

src/packages/conat/core/server.ts

@@ -63,6 +63,8 @@ import {
   createClusterPersistServer,
   Sticky,
   Interest,
+  hashInterest,
+  hashSticky,
 } from "./cluster";
 import { type ConatSocketServer } from "@cocalc/conat/socket";
 import { throttle } from "lodash";
@@ -174,10 +176,10 @@ export class ConatServer extends EventEmitter {
   public state: State = "init";
 
   private subscriptions: { [socketId: string]: Set<string> } = {};
-  private interest: Interest = new Patterns();
+  public interest: Interest = new Patterns();
   // the target string is JSON.stringify({ id: string; subject: string }),
   // which is the socket.io room to send the messages to.
-  private sticky: Sticky = {};
+  public sticky: Sticky = {};
 
   private clusterStreams?: ClusterStreams;
   private clusterLinks: {
@@ -1532,6 +1534,13 @@ export class ConatServer extends EventEmitter {
     }
     return false;
   };
+
+  hash = (): { interest: number; sticky: number } => {
+    return {
+      interest: hashInterest(this.interest),
+      sticky: hashSticky(this.sticky),
+    };
+  };
 }
 
 function getSubjectFromRoom(room: string) {

src/packages/util/smc-version.js

@@ -1,2 +1,2 @@
 /* autogenerated by the update_version script */
-exports.version=1752021564;
+exports.version=1752088354;