cluster: starting work on sticky stream

Author: williamstein

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

@@ -15,9 +15,9 @@ import {
 } from "@cocalc/backend/conat/test/setup";
 import {
   superclusterLink,
-  superclusterStream,
+  superclusterStreams,
   superclusterService,
-  trimSuperclusterStream,
+  trimSuperclusterStreams,
 } from "@cocalc/conat/core/supercluster";
 import { isEqual } from "lodash";
 
@@ -42,16 +42,16 @@ describe("create a supercluster enabled socketio server and test that the stream
     ({ server, client } = await createCluster());
   });
 
-  let stream;
+  let streams;
   it("get the interest stream via our client. There MUST be at least two persist subjects in there, since they were needed to even create the interest stream.", async () => {
-    stream = await superclusterStream({
+    streams = await superclusterStreams({
       client,
       clusterName: server.options.clusterName,
     });
     const service = superclusterService(server.options.clusterName);
     await wait({
       until: () => {
-        const v = stream.getAll();
+        const v = streams.interest.getAll();
         expect(service).toContain(server.options.clusterName);
         const persistUpdates = v.filter((update) =>
           update.subject.startsWith(service),
@@ -68,25 +68,25 @@ describe("create a supercluster enabled socketio server and test that the stream
   it("subscribe and see update appear in the stream; close sub and see delete appear", async () => {
     const sub = await client.subscribe("foo");
     while (true) {
-      const v = stream.getAll().filter((x) => x.subject == "foo");
+      const v = streams.interest.getAll().filter((x) => x.subject == "foo");
       if (v.length == 1) {
         expect(v[0]).toEqual(
           expect.objectContaining({ op: "add", subject: "foo" }),
         );
         break;
       }
-      await once(stream, "change");
+      await once(streams.interest, "change");
     }
     sub.close();
     while (true) {
-      const v = stream.getAll().filter((x) => x.subject == "foo");
+      const v = streams.interest.getAll().filter((x) => x.subject == "foo");
       if (v.length == 2) {
         expect(v[1]).toEqual(
           expect.objectContaining({ op: "delete", subject: "foo" }),
         );
         break;
       }
-      await once(stream, "change");
+      await once(streams.interest, "change");
     }
   });
 
@@ -341,15 +341,15 @@ describe("test trimming the interest stream", () => {
   let server, client;
   it("create a supercluster server", async () => {
     ({ server, client } = await createCluster());
-    await wait({ until: () => server.superclusterStream != null });
+    await wait({ until: () => server.superclusterStreams != null });
   });
 
   let sub;
   it("subscribes and verifies that trimming does nothing", async () => {
     sub = await client.sub("389");
-    const seqs = await trimSuperclusterStream(
-      server.superclusterStream,
-      server.interest,
+    const seqs = await trimSuperclusterStreams(
+      server.superclusterStreams,
+      server,
       0,
     );
     expect(seqs).toEqual([]);
@@ -358,9 +358,9 @@ describe("test trimming the interest stream", () => {
   it("unsubscribes and verifies that trimming with a 5s maxAge does nothing", async () => {
     sub.close();
     await delay(100);
-    const seqs = await trimSuperclusterStream(
-      server.superclusterStream,
-      server.interest,
+    const seqs = await trimSuperclusterStreams(
+      server.superclusterStreams,
+      server,
       5000,
     );
     expect(seqs).toEqual([]);
@@ -372,17 +372,17 @@ describe("test trimming the interest stream", () => {
     let seqs;
     await wait({
       until: async () => {
-        seqs = await trimSuperclusterStream(
-          server.superclusterStream,
-          server.interest,
+        seqs = await trimSuperclusterStreams(
+          server.superclusterStreams,
+          server,
           0,
         );
         return seqs.length >= 2;
       },
     });
     expect(seqs.length).toBe(2);
     await delay(1);
-    for (const update of server.superclusterStream.getAll()) {
+    for (const update of server.superclusterStreams.interest.getAll()) {
       if (update.subject == "389" && update.op == "add") {
         throw Error("adding 389 should have been removed");
       }

src/packages/conat/core/server.ts

@@ -83,12 +83,12 @@ import ConsistentHash from "consistent-hash";
 import { is_array } from "@cocalc/util/misc";
 import { UsageMonitor } from "@cocalc/conat/monitor/usage";
 import { once, until } from "@cocalc/util/async-utils";
-import type { DStream } from "@cocalc/conat/sync/dstream";
 import {
   superclusterLink,
   type SuperclusterLink,
-  superclusterStream,
-  trimSuperclusterStream,
+  superclusterStreams,
+  type ClusterStreams,
+  trimSuperclusterStreams,
   createSuperclusterPersistServer,
 } from "./supercluster";
 import { type ConatSocketServer } from "@cocalc/conat/socket";
@@ -202,7 +202,7 @@ export class ConatServer {
   // which is the socket.io room to send the messages to.
   private sticky: { [pattern: string]: { [subject: string]: string } } = {};
 
-  private superclusterStream?: DStream<InterestUpdate>;
+  private superclusterStreams?: ClusterStreams;
   private superclusterLinks: { [clusterName: string]: SuperclusterLink } = {};
   private superclusterPersistServer?: ConatSocketServer;
   private clusterName?: string;
@@ -348,8 +348,12 @@ export class ConatServer {
     }
     this.state = "closed";
 
-    this.superclusterStream?.close();
-    delete this.superclusterStream;
+    if (this.superclusterStreams != null) {
+      for (const name in this.superclusterStreams) {
+        this.superclusterStreams[name].close();
+      }
+      delete this.superclusterStreams;
+    }
     for (const name in this.superclusterLinks) {
       this.superclusterLinks[name].close();
       delete this.superclusterLinks[name];
@@ -475,8 +479,8 @@ export class ConatServer {
   private queuedSuperclusterUpdates: InterestUpdate[] = [];
   private updateSuperclusterStream = (update: InterestUpdate) => {
     if (!this.clusterName) return;
-    if (this.superclusterStream !== undefined) {
-      this.superclusterStream.publish(update);
+    if (this.superclusterStreams !== undefined) {
+      this.superclusterStreams.interest.publish(update);
       this.trimSuperclusterStream();
     } else {
       this.queuedSuperclusterUpdates.push(update);
@@ -490,12 +494,13 @@ export class ConatServer {
   private trimSuperclusterStream = throttle(
     async () => {
       if (
-        this.superclusterStream !== undefined &&
-        this.interest !== undefined
+        this.superclusterStreams !== undefined &&
+        this.interest !== undefined &&
+        this.sticky !== undefined
       ) {
-        await trimSuperclusterStream(
-          this.superclusterStream,
-          this.interest,
+        await trimSuperclusterStreams(
+          this.superclusterStreams,
+          { interest: this.interest, sticky: this.sticky },
           5 * 60000,
         );
       }
@@ -942,20 +947,15 @@ export class ConatServer {
       client,
       clusterName: this.clusterName,
     });
-    this.log("creating interest stream");
-    const stream = await superclusterStream({
+    this.log("creating cluster streams");
+    this.superclusterStreams = await superclusterStreams({
       client,
       clusterName: this.clusterName,
     });
-    this.log("initializing interest stream");
-    if (stream.length > 0) {
-      await stream.delete({ all: true });
-    }
-    this.superclusterStream = stream;
     // add in everything so far in interest (TODO)
     if (this.queuedSuperclusterUpdates.length > 0) {
       for (const update0 of this.queuedSuperclusterUpdates) {
-        this.superclusterStream.publish(update0);
+        this.superclusterStreams.interest.publish(update0);
       }
       this.queuedSuperclusterUpdates.length = 0;
     }

src/packages/conat/core/supercluster.ts

@@ -9,6 +9,7 @@ import {
   randomChoice,
   updateInterest,
   type InterestUpdate,
+  type StickyUpdate,
 } from "@cocalc/conat/core/server";
 import type { DStream } from "@cocalc/conat/sync/dstream";
 import { once } from "@cocalc/util/async-utils";
@@ -32,7 +33,7 @@ export async function superclusterLink({
 export class SuperclusterLink {
   private interest: Patterns<{ [queue: string]: Set<string> }> = new Patterns();
   private sticky: { [pattern: string]: { [subject: string]: string } } = {};
-  private stream: DStream<InterestUpdate>;
+  private streams: ClusterStreams;
   private state: "init" | "ready" | "closed" = "init";
 
   constructor(
@@ -48,25 +49,33 @@ export class SuperclusterLink {
   }
 
   init = async () => {
-    this.stream = await superclusterStream({
+    this.streams = await superclusterStreams({
       client: this.client,
       clusterName: this.clusterName,
     });
-    for (const update of this.stream.getAll()) {
+    for (const update of this.streams.interest.getAll()) {
       updateInterest(update, this.interest, this.sticky);
     }
-    this.stream.on("change", this.handleUpdate);
+    this.streams.interest.on("change", this.handleInterestUpdate);
     this.state = "ready";
   };
 
-  handleUpdate = (update) => {
+  handleInterestUpdate = (update) => {
     updateInterest(update, this.interest, this.sticky);
   };
 
   close = () => {
+    if (this.state == "closed") {
+      return;
+    }
     this.state = "closed";
-    this.stream?.removeListener("change", this.handleUpdate);
-    this.stream?.close();
+    if (this.streams != null) {
+      this.streams.interest.removeListener("change", this.handleInterestUpdate);
+      this.streams.interest.close();
+      this.streams.sticky.close();
+      // @ts-ignore
+      delete this.streams;
+    }
   };
 
   publish = ({ subject, data }) => {
@@ -164,56 +173,74 @@ export async function createSuperclusterPersistServer({
   return await createPersistServer({ client, service });
 }
 
-export async function superclusterStream({
+export interface ClusterStreams {
+  interest: DStream<InterestUpdate>;
+  sticky: DStream<StickyUpdate>;
+}
+
+export async function superclusterStreams({
   client,
   clusterName,
 }: {
   client: Client;
   clusterName: string;
-}): Promise<DStream<InterestUpdate>> {
+}): Promise<ClusterStreams> {
   logger.debug("superclusterStream: ", { clusterName });
   if (!clusterName) {
     throw Error("clusterName must be set");
   }
-  const stream = await client.sync.dstream<InterestUpdate>({
-    name: superclusterStreamNames(clusterName).interest,
+  const names = superclusterStreamNames(clusterName);
+  const opts = {
     service: superclusterService(clusterName),
     noCache: true,
+    ephemeral: true,
+  };
+  const interest = await client.sync.dstream<InterestUpdate>({
+    name: names.interest,
+    ...opts,
+  });
+  const sticky = await client.sync.dstream<StickyUpdate>({
+    name: names.sticky,
+    ...opts,
   });
-  logger.debug("superclusterStream: GOT IT", { clusterName });
-  return stream;
+  logger.debug("superclusterStreams: got them", { clusterName });
+  return { interest, sticky };
 }
 
 // Periodically delete not-necessary updates from the interest stream
-export async function trimSuperclusterStream(
-  stream: DStream<InterestUpdate>,
-  interest: Patterns<{ [queue: string]: Set<string> }>,
+export async function trimSuperclusterStreams(
+  streams: ClusterStreams,
+  data: {
+    interest: Patterns<{ [queue: string]: Set<string> }>;
+    sticky: { [pattern: string]: { [subject: string]: string } };
+  },
   // don't delete anything that isn't at lest minAge ms old.
   minAge: number,
 ): Promise<number[]> {
-  // we simply iterate over the stream checking for subjects
+  const { interest } = streams;
+  // we iterate over the interest stream checking for subjects
   // with no current interest at all; in such cases it is safe
   // to purge them entirely from the stream.
   const seqs: number[] = [];
   const now = Date.now();
-  for (let n = 0; n < stream.length; n++) {
-    const time = stream.time(n);
+  for (let n = 0; n < interest.length; n++) {
+    const time = interest.time(n);
     if (time == null || now - time.valueOf() <= minAge) {
       break;
     }
-    const update = stream.get(n) as InterestUpdate;
-    if (!interest.hasPattern(update.subject)) {
-      const seq = stream.seq(n);
+    const update = interest.get(n) as InterestUpdate;
+    if (!data.interest.hasPattern(update.subject)) {
+      const seq = interest.seq(n);
       if (seq != null) {
         seqs.push(seq);
       }
     }
   }
   if (seqs.length > 0) {
-    // [ ] todo -- add to stream.delete a version where it takes an array of sequence numbers
+    // [ ] todo -- add to interest.delete a version where it takes an array of sequence numbers
     logger.debug("trimSuperclusterStream: trimming", { seqs });
     for (const seq of seqs) {
-      await stream.delete({ seq });
+      await interest.delete({ seq });
     }
   }
   return seqs;