Merge remote-tracking branch 'origin/master' into cocalc-api-20250927

Author: haraldschilly

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

@@ -3,7 +3,6 @@ import {
   after,
   defaultCluster as servers,
   waitForConsistentState,
-  wait,
   addNodeToDefaultCluster,
 } from "@cocalc/backend/conat/test/setup";
 import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
@@ -46,9 +45,6 @@ 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);
     }
   });
 
@@ -65,51 +61,12 @@ 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];
 
@@ -154,17 +111,6 @@ 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,
@@ -180,14 +126,6 @@ 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,7 +206,6 @@ 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) {
@@ -220,9 +219,8 @@ 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", "sticky"]) {
+              for (const type of ["interest"]) {
                 console.log(
                   `server stream ${type}: `,
                   hashServer[type],
@@ -254,8 +252,6 @@ 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 ./open-files.test.ts
+pnpm test `pwd`/open-files.test.ts
 
 */
 

src/packages/conat/core/cluster.ts

@@ -1,11 +1,6 @@
 import { type Client, connect } from "./client";
 import { Patterns } from "./patterns";
-import {
-  updateInterest,
-  updateSticky,
-  type InterestUpdate,
-  type StickyUpdate,
-} from "@cocalc/conat/core/server";
+import { updateInterest, type InterestUpdate } 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";
@@ -47,14 +42,12 @@ 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
@@ -85,10 +78,7 @@ class ClusterLink {
       clusterName: this.clusterName,
     });
     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);
+      updateInterest(update, this.interest);
     }
     // I have a slight concern about this because updates might not
     // arrive in order during automatic failover.  That said, maybe
@@ -97,7 +87,6 @@ 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";
   };
 
@@ -106,11 +95,7 @@ class ClusterLink {
   };
 
   handleInterestUpdate = (update: InterestUpdate) => {
-    updateInterest(update, this.interest, this.sticky);
-  };
-
-  handleStickyUpdate = (update: StickyUpdate) => {
-    updateSticky(update, this.sticky);
+    updateInterest(update, this.interest);
   };
 
   private handleClientStateChanged = () => {
@@ -134,7 +119,6 @@ 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;
     }
@@ -178,10 +162,9 @@ class ClusterLink {
     return false;
   };
 
-  hash = (): { interest: number; sticky: number } => {
+  hash = (): { interest: number } => {
     return {
       interest: hashInterest(this.interest),
-      sticky: hashSticky(this.sticky),
     };
   };
 }
@@ -195,7 +178,6 @@ function clusterStreamNames({
 }) {
   return {
     interest: `cluster/${clusterName}/${id}/interest`,
-    sticky: `cluster/${clusterName}/${id}/sticky`,
   };
 }
 
@@ -225,7 +207,6 @@ export async function createClusterPersistServer({
 
 export interface ClusterStreams {
   interest: DStream<InterestUpdate>;
-  sticky: DStream<StickyUpdate>;
 }
 
 export async function clusterStreams({
@@ -252,27 +233,21 @@ 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, sticky };
+  return { interest };
 }
 
 // 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[]; seqsSticky: number[] }> {
-  const { interest, sticky } = streams;
+): Promise<{ seqsInterest: number[] }> {
+  const { interest } = 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
@@ -300,45 +275,7 @@ export async function trimClusterStreams(
     logger.debug("trimClusterStream: successfully trimmed interest", { 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 };
+  return { seqsInterest: seqs };
 }
 
 function hashSet(X: Set<string>): number {
@@ -362,15 +299,3 @@ 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;
-}