make waitForInterest cancellable and use this when waiting for interest across a supercluster

Author: williamstein

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

@@ -10,7 +10,7 @@ import {
   after,
   initConatServer,
   once,
-  //delay,
+  delay,
   wait,
 } from "@cocalc/backend/conat/test/setup";
 import { server as createPersistServer } from "@cocalc/backend/conat/persist";
@@ -176,7 +176,7 @@ describe("create a supercluster enabled socketio server and test that the stream
   });
 });
 
-describe("create a supercluster with two distinct servers and send a message from one client to another via a link, and also use request/reply", () => {
+describe.only("create a supercluster with two distinct servers and send a message from one client to another via a link, and also use request/reply", () => {
   let server1, server2, client1, client2;
   it("create two distinct servers with supercluster support enabled", async () => {
     ({ server: server1, client: client1 } = await createCluster({
@@ -198,6 +198,19 @@ describe("create a supercluster with two distinct servers and send a message fro
     });
   });
 
+  it("tests that server-side waitForInterest can be aborted", async () => {
+    const controller = new AbortController();
+    const w = server2.waitForInterest(
+      "no-interest",
+      90000,
+      client2.conn.id,
+      controller.signal,
+    );
+    await delay(15);
+    controller.abort();
+    expect(await w).toBe(false);
+  });
+
   const N =
     "114381625757888867669235779976146612010218296721242362562561842935706935245733897830597123563958705058989075147599290026879543541";
 

src/packages/conat/core/server.ts

@@ -971,13 +971,15 @@ export class ConatServer {
     subject: string,
     timeout: number,
     socketId: string,
+    signal?: AbortSignal,
   ): Promise<boolean> => {
     const links = Object.values(this.superclusterLinks);
     if (links.length == 0) {
       return await this.waitForInterestInThisCluster(
         subject,
         timeout,
         socketId,
+        signal,
       );
     } else {
       const v: any[] = [];
@@ -993,15 +995,23 @@ export class ConatServer {
           }
           return false;
         };
+        const controller = new AbortController();
+        const signal2 = controller.signal;
         v.push(
           nothrow(
-            this.waitForInterestInThisCluster(subject, timeout, socketId),
+            this.waitForInterestInThisCluster(
+              subject,
+              timeout,
+              socketId,
+              signal2,
+            ),
           ),
         );
         for (const link of links) {
-          v.push(nothrow(link.waitForInterest(subject, timeout)));
+          v.push(nothrow(link.waitForInterest(subject, timeout, signal2)));
         }
         if (!timeout) {
+          // with timeout=0 they all immediately answer (so no need to worry about abort/pormise)
           const w = await Promise.all(v);
           for (const x of w) {
             if (x) {
@@ -1010,8 +1020,13 @@ export class ConatServer {
           }
           return false;
         }
+
+        signal?.addEventListener("abort", () => {
+          controller.abort();
+        });
         const w = await Promise.race(v);
-        // [ ] todo: cancel all the others.
+        // cancel all the others.
+        controller.abort();
         return w;
       } finally {
         done = true;
@@ -1023,6 +1038,7 @@ export class ConatServer {
     subject: string,
     timeout: number,
     socketId: string,
+    signal?: AbortSignal,
   ) => {
     const matches = this.interest.matches(subject);
     if (matches.length > 0 || !timeout) {
@@ -1035,12 +1051,20 @@ export class ConatServer {
       timeout = MAX_INTEREST_TIMEOUT;
     }
     const start = Date.now();
-    while (this.state != "closed" && this.sockets[socketId]) {
+    while (
+      this.state != "closed" &&
+      this.sockets[socketId] &&
+      !signal?.aborted
+    ) {
       if (Date.now() - start >= timeout) {
         throw Error("timeout");
       }
       await once(this.interest, "change");
-      if ((this.state as any) == "closed" || !this.sockets[socketId]) {
+      if (
+        (this.state as any) == "closed" ||
+        !this.sockets[socketId] ||
+        signal?.aborted
+      ) {
         return false;
       }
       const matches = this.interest.matches(subject);

src/packages/conat/core/supercluster.ts

@@ -112,7 +112,11 @@ export class SuperclusterLink {
     return randomChoice(targets);
   };
 
-  waitForInterest = async (subject: string, timeout: number) => {
+  waitForInterest = async (
+    subject: string,
+    timeout: number,
+    signal?: AbortSignal,
+  ) => {
     const matches = this.interest.matches(subject);
 
     if (matches.length > 0 || !timeout) {
@@ -122,12 +126,12 @@ export class SuperclusterLink {
       return matches.length > 0;
     }
     const start = Date.now();
-    while (this.state != "closed") {
+    while (this.state != "closed" && !signal?.aborted) {
       if (Date.now() - start >= timeout) {
         throw Error("timeout");
       }
       await once(this.interest, "change");
-      if ((this.state as any) == "closed") {
+      if ((this.state as any) == "closed" || signal?.aborted) {
         return false;
       }
       // todo: implement this.interest.hasMatch that just checks if there is at least one match