implement computing hash of interest and sticky, so we can be sure conat cluster state doesn't "drift"

Author: williamstein

src/packages/conat/core/cluster.ts

@@ -10,6 +10,7 @@ import type { DStream } from "@cocalc/conat/sync/dstream";
 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 logger = getLogger("conat:core:cluster");
 
@@ -41,11 +42,14 @@ export async function clusterLink(
   return link;
 }
 
+export type Sticky = { [pattern: string]: { [subject: string]: string } };
+export type Interest = Patterns<{ [queue: string]: Set<string> }>;
+
 export { type ClusterLink };
 
 class ClusterLink {
-  public interest: Patterns<{ [queue: string]: Set<string> }> = new Patterns();
-  private sticky: { [pattern: string]: { [subject: string]: string } } = {};
+  public interest: Interest = new Patterns();
+  private sticky: Sticky = {};
   private streams: ClusterStreams;
   private state: "init" | "ready" | "closed" = "init";
   private clientStateChanged = Date.now(); // when client status last changed
@@ -322,3 +326,37 @@ export async function trimClusterStreams(
 
   return { seqsInterest: seqs, seqsSticky: seqs2 };
 }
+
+function hashSet(X: Set<string>): number {
+  let h = 0;
+  for (const a of X) {
+    h += hash_string(a); // integers, and not too many, so should commute
+  }
+  return h;
+}
+
+function hashInterestValue(X: { [queue: string]: Set<string> }): number {
+  let h = 0;
+  for (const queue in X) {
+    h += hashSet(X[queue]); // integers, and not too many, so should commute
+  }
+  return h;
+}
+
+export function hashInterest(
+  interest: Patterns<{ [queue: string]: Set<string> }>,
+): number {
+  return interest.hash(hashInterestValue);
+}
+
+export function hashSticky(sticky: Sticky): number {
+  let h = 0;
+  for (const pattern in sticky) {
+    h += hash_string(pattern);
+    const x = sticky[pattern];
+    for (const subject in x) {
+      h += hash_string(x[subject]);
+    }
+  }
+  return h;
+}

src/packages/conat/core/patterns.ts

@@ -1,6 +1,7 @@
 import { isEqual } from "lodash";
 import { getLogger } from "@cocalc/conat/client";
 import { EventEmitter } from "events";
+import { hash_string } from "@cocalc/util/misc";
 
 type Index = { [pattern: string]: Index | string };
 
@@ -21,6 +22,14 @@ export class Patterns<T> extends EventEmitter {
     this.index = {};
   };
 
+  hash = (hashT: (x: T) => number): number => {
+    let h = 0;
+    for (const pattern in this.patterns) {
+      h += hash_string(pattern) + hashT(this.patterns[pattern]);
+    }
+    return h;
+  };
+
   serialize = (fromT?: (x: T) => any) => {
     let patterns: { [pattern: string]: any };
     if (fromT != null) {

src/packages/conat/core/server.ts

@@ -61,6 +61,8 @@ import {
   type ClusterStreams,
   trimClusterStreams,
   createClusterPersistServer,
+  Sticky,
+  Interest,
 } from "./cluster";
 import { type ConatSocketServer } from "@cocalc/conat/socket";
 import { throttle } from "lodash";
@@ -172,10 +174,10 @@ export class ConatServer extends EventEmitter {
   public state: State = "init";
 
   private subscriptions: { [socketId: string]: Set<string> } = {};
-  private interest: Patterns<{ [queue: string]: Set<string> }> = new Patterns();
+  private 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: { [pattern: string]: { [subject: string]: string } } = {};
+  private sticky: Sticky = {};
 
   private clusterStreams?: ClusterStreams;
   private clusterLinks: {
@@ -1585,10 +1587,8 @@ function getAddress(socket) {
 
 export function updateInterest(
   update: InterestUpdate,
-  interest: Patterns<{ [queue: string]: Set<string> }>,
-  sticky: {
-    [pattern: string]: { [subject: string]: string };
-  },
+  interest: Interest,
+  sticky: Sticky,
 ) {
   const { op, subject, queue, room } = update;
   const groups = interest.get(subject);
@@ -1626,12 +1626,7 @@ export function updateInterest(
 }
 
 // returns true if this update actually causes a change to sticky
-export function updateSticky(
-  update: StickyUpdate,
-  sticky: {
-    [pattern: string]: { [subject: string]: string };
-  },
-): boolean {
+export function updateSticky(update: StickyUpdate, sticky: Sticky): boolean {
   const { pattern, subject, target } = update;
   if (sticky[pattern] === undefined) {
     sticky[pattern] = {};