Merge pull request #8605 from sagemathinc/revert-sticky-routing

Revert sticky routing series (3e0a443..45bcc84).

Author: williamstein

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

@@ -3,6 +3,7 @@ import {
   after,
   defaultCluster as servers,
   waitForConsistentState,
+  wait,
   addNodeToDefaultCluster,
 } from "@cocalc/backend/conat/test/setup";
 import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
@@ -45,6 +46,9 @@ describe("ensure sticky state sync and use is working properly", () => {
       );
       // publishing causes a choice to be made and saved on servers[0]
       await clients[0].publish(`subject.${i}.foo`, "hello");
+      expect(servers[0].sticky[`subject.${i}.*`]).not.toBe(undefined);
+      // but no choice on servers[1]
+      expect(servers[1].sticky[`subject.${i}.*`]).toBe(undefined);
     }
   });
 
@@ -61,12 +65,51 @@ describe("ensure sticky state sync and use is working properly", () => {
     chosen = await Promise.race([p0(), p1()]);
   });
 
+  it(`sticky on servers[0] should have ${count} entries starting in "subject".`, async () => {
+    const v = Object.keys(servers[0].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(count);
+  });
+
+  it(`sticky on servers[1] should have no entries starting in "subject".`, async () => {
+    const v = Object.keys(servers[1].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(0);
+  });
+
+  it(`servers[1]'s link to servers[0] should *eventually* have ${count} entries starting in "subject."`, async () => {
+    // @ts-ignore
+    const link = servers[1].clusterLinksByAddress[servers[0].address()];
+    let v;
+    await wait({
+      until: () => {
+        v = Object.keys(link.sticky).filter((s) => s.startsWith("subject."));
+        return v.length == count;
+      },
+    });
+    expect(v.length).toBe(count);
+  });
+
   it("send message from clients[1] to each subject", async () => {
     for (let i = 0; i < count; i++) {
       await clients[1].publish(`subject.${i}.foo`);
     }
   });
 
+  // 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];
 
@@ -111,6 +154,17 @@ describe("ensure sticky state sync and use is working properly", () => {
     await waitForConsistentState(servers);
   });
 
+  it("double check the links have the sticky state", () => {
+    for (const server of servers.slice(1)) {
+      // @ts-ignore
+      const link = server.clusterLinksByAddress[servers[0].address()];
+      const v = Object.keys(link.sticky).filter((s) =>
+        s.startsWith("subject."),
+      );
+      expect(v.length).toBe(count);
+    }
+  });
+
   it(
     "in bigger, cluster, publish from every node to subject.0.foo",
     deliveryTest,
@@ -126,6 +180,14 @@ describe("ensure sticky state sync and use is working properly", () => {
     }
     sub.close();
   });
+
+  it("unjoining servers[0] from servers[1] should transfer the sticky state to servers[1]", async () => {
+    await servers[1].unjoin({ address: servers[0].address() });
+    const v = Object.keys(servers[1].sticky).filter((s) =>
+      s.startsWith("subject."),
+    );
+    expect(v.length).toBe(count);
+  });
 });
 
 afterAll(after);

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

@@ -206,6 +206,7 @@ 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) {
@@ -219,8 +220,9 @@ export async function waitForConsistentState(
             }
             const hashLink = link.hash();
             const x = link.interest.serialize().patterns;
+            const y = link.sticky;
             const showInfo = () => {
-              for (const type of ["interest"]) {
+              for (const type of ["interest", "sticky"]) {
                 console.log(
                   `server stream ${type}: `,
                   hashServer[type],
@@ -252,6 +254,8 @@ export async function waitForConsistentState(
                 j,
                 serverInterest: a,
                 linkInterest: x,
+                serverSticky: b,
+                linkSticky: y,
               });
             };
             if (!isEqual(hashServer, hashLink)) {

src/packages/backend/conat/test/sync/open-files.test.ts

@@ -8,7 +8,7 @@ to open so they can fulfill their backend responsibilities:
 
 DEVELOPMENT:
 
-pnpm test `pwd`/open-files.test.ts
+pnpm test ./open-files.test.ts
 
 */
 

src/packages/conat/core/cluster.ts

@@ -1,6 +1,11 @@
 import { type Client, connect } from "./client";
 import { Patterns } from "./patterns";
-import { updateInterest, type InterestUpdate } from "@cocalc/conat/core/server";
+import {
+  updateInterest,
+  updateSticky,
+  type InterestUpdate,
+  type StickyUpdate,
+} from "@cocalc/conat/core/server";
 import type { DStream } from "@cocalc/conat/sync/dstream";
 import { once } from "@cocalc/util/async-utils";
 import { server as createPersistServer } from "@cocalc/conat/persist/server";
@@ -42,12 +47,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: Interest = new Patterns();
+  public sticky: Sticky = {};
   private streams: ClusterStreams;
   private state: "init" | "ready" | "closed" = "init";
   private clientStateChanged = Date.now(); // when client status last changed
@@ -78,7 +85,10 @@ class ClusterLink {
       clusterName: this.clusterName,
     });
     for (const update of this.streams.interest.getAll()) {
-      updateInterest(update, this.interest);
+      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
@@ -87,6 +97,7 @@ class ClusterLink {
     // it is about, and when that server goes down none of this state
     // matters anymore.
     this.streams.interest.on("change", this.handleInterestUpdate);
+    this.streams.sticky.on("change", this.handleStickyUpdate);
     this.state = "ready";
   };
 
@@ -95,7 +106,11 @@ class ClusterLink {
   };
 
   handleInterestUpdate = (update: InterestUpdate) => {
-    updateInterest(update, this.interest);
+    updateInterest(update, this.interest, this.sticky);
+  };
+
+  handleStickyUpdate = (update: StickyUpdate) => {
+    updateSticky(update, this.sticky);
   };
 
   private handleClientStateChanged = () => {
@@ -119,6 +134,7 @@ class ClusterLink {
     if (this.streams != null) {
       this.streams.interest.removeListener("change", this.handleInterestUpdate);
       this.streams.interest.close();
+      this.streams.sticky.close();
       // @ts-ignore
       delete this.streams;
     }
@@ -162,9 +178,10 @@ class ClusterLink {
     return false;
   };
 
-  hash = (): { interest: number } => {
+  hash = (): { interest: number; sticky: number } => {
     return {
       interest: hashInterest(this.interest),
+      sticky: hashSticky(this.sticky),
     };
   };
 }
@@ -178,6 +195,7 @@ function clusterStreamNames({
 }) {
   return {
     interest: `cluster/${clusterName}/${id}/interest`,
+    sticky: `cluster/${clusterName}/${id}/sticky`,
   };
 }
 
@@ -207,6 +225,7 @@ export async function createClusterPersistServer({
 
 export interface ClusterStreams {
   interest: DStream<InterestUpdate>;
+  sticky: DStream<StickyUpdate>;
 }
 
 export async function clusterStreams({
@@ -233,21 +252,27 @@ export async function clusterStreams({
     name: names.interest,
     ...opts,
   });
+  const sticky = await client.sync.dstream<StickyUpdate>({
+    noInventory: true,
+    name: names.sticky,
+    ...opts,
+  });
   logger.debug("clusterStreams: got them", { clusterName });
-  return { interest };
+  return { interest, sticky };
 }
 
 // Periodically delete not-necessary updates from the interest stream
 export async function trimClusterStreams(
   streams: ClusterStreams,
   data: {
     interest: Patterns<{ [queue: string]: Set<string> }>;
+    sticky: { [pattern: string]: { [subject: string]: string } };
     links: { interest: Patterns<{ [queue: string]: Set<string> }> }[];
   },
   // don't delete anything that isn't at lest minAge ms old.
   minAge: number,
-): Promise<{ seqsInterest: number[] }> {
-  const { interest } = streams;
+): Promise<{ seqsInterest: number[]; seqsSticky: number[] }> {
+  const { interest, sticky } = streams;
   // First deal with interst
   // we iterate over the interest stream checking for subjects
   // with no current interest at all; in such cases it is safe
@@ -275,7 +300,45 @@ export async function trimClusterStreams(
     logger.debug("trimClusterStream: successfully trimmed interest", { seqs });
   }
 
-  return { seqsInterest: seqs };
+  // Next deal with sticky -- trim ones where the pattern is no longer of interest.
+  // There could be other reasons to trim but it gets much trickier. This one is more
+  // obvious, except we have to check for any interest in the whole cluster, not
+  // just this node.
+  const seqs2: number[] = [];
+  function noInterest(pattern: string) {
+    if (data.interest.hasPattern(pattern)) {
+      return false;
+    }
+    for (const link of data.links) {
+      if (link.interest.hasPattern(pattern)) {
+        return false;
+      }
+    }
+    // nobody cares
+    return true;
+  }
+  for (let n = 0; n < sticky.length; n++) {
+    const time = sticky.time(n);
+    if (time == null) continue;
+    if (now - time.valueOf() <= minAge) {
+      break;
+    }
+    const update = sticky.get(n) as StickyUpdate;
+    if (noInterest(update.pattern)) {
+      const seq = sticky.seq(n);
+      if (seq != null) {
+        seqs2.push(seq);
+      }
+    }
+  }
+  if (seqs2.length > 0) {
+    // [ ] todo -- add to interest.delete a version where it takes an array of sequence numbers
+    logger.debug("trimClusterStream: trimming sticky", { seqs2 });
+    await sticky.delete({ seqs: seqs2 });
+    logger.debug("trimClusterStream: successfully trimmed sticky", { seqs2 });
+  }
+
+  return { seqsInterest: seqs, seqsSticky: seqs2 };
 }
 
 function hashSet(X: Set<string>): number {
@@ -299,3 +362,15 @@ export function hashInterest(
 ): 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;
+}