make listening on > be silent and document the subtle issue

Author: williamstein

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

@@ -5,7 +5,6 @@ import {
   waitForConsistentState,
   wait,
   addNodeToDefaultCluster,
-  delay,
 } from "@cocalc/backend/conat/test/setup";
 import { STICKY_QUEUE_GROUP } from "@cocalc/conat/core/client";
 import { randomId } from "@cocalc/conat/names";
@@ -152,6 +151,7 @@ describe("ensure sticky state sync and use is working properly", () => {
 
   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."),
@@ -165,6 +165,17 @@ describe("ensure sticky state sync and use is working properly", () => {
     deliveryTest,
   );
 
+  it("listen on > and note that it doesn't impact the count", async () => {
+    const sub = await clients[0].subscribe(">");
+    for (let i = 0; i < servers.length; i++) {
+      const { count } = await servers[i]
+        .client()
+        .publish("subject.0.foo", "hi");
+      expect(count).toBe(1);
+    }
+    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) =>

src/packages/backend/conat/test/persist/multiple-servers.test.ts

@@ -4,7 +4,7 @@ Unit tests of multiple persist servers at once:
 - making numerous distinct clients and seeing that they are distributed across persist servers
 - stopping a persist server and seeing that failover happens without data loss
 
-pnpm test `pwd`/multiple-servers.test.ts 
+pnpm test `pwd`/multiple-servers.test.ts
 
 */
 

src/packages/backend/conat/test/socket/basic.test.ts

@@ -1,6 +1,6 @@
 /*
 
-pnpm test `pwd`/basic.test.ts 
+pnpm test `pwd`/basic.test.ts
 
 */
 

src/packages/backend/conat/test/socket/restarts.test.ts

@@ -18,8 +18,6 @@ beforeAll(async () => {
   setDefaultTimeouts({ request: 500, publish: 500 });
 });
 
-//jest.setTimeout(25000);
-
 describe("create a client and server and socket, verify it works, restart conat server, then confirm that socket still works", () => {
   const SUBJECT = "reconnect.one";
 

src/packages/conat/core/client.ts

@@ -22,6 +22,22 @@ much better in so many ways:
      - JsonCodec: uses JSON.stringify and TextEncoder.  This does not work
        with Buffer or Date and is less compact, but can be very fast.
 
+- One BIG DIFFERENCE from Nats is that when a message is sent the sender
+  can optionally find out how many clients received it.  They can also wait
+  until there is interest and send the message again.  This is automated so
+  it's very easy to use, and it makes writing distributed services without
+  race conditions making them broken temporarily much easier.  HOWEVER,
+  there is one caveat -- if as an admin you create a "tap", i.e., you
+  subscribe to all messages matching some pattern just to see what's going
+  on, then currently that counts in the delivery count and interest, and that
+  would then cause these race conditions to happen again. E.g., a user
+  signs in, subscribes to their INBOX, sends a request, and gets a response
+  to that inbox, but does this all quickly, and in a cluster, the server doesn't
+  see the inbox yet, so it fails.  As a workaround, subscriptions to the
+  subject pattern '>' are invisible, so you can always tap into '>' for debugging
+  purposes.  TODO: implement a general way of making an invisible subscriber that
+  doesn't count.
+
 
 THE CORE API
 
@@ -604,13 +620,14 @@ export class Client extends EventEmitter {
       );
     }
     timeout = Math.min(timeout, MAX_INTEREST_TIMEOUT);
-    const response = await this.conn
-      .timeout(timeout ? timeout : 10000)
-      .emitWithAck("wait-for-interest", { subject, timeout });
-    if (response.error) {
-      throw new ConatError(response.error, { code: response.code });
+    try {
+      const response = await this.conn
+        .timeout(timeout ? timeout : 10000)
+        .emitWithAck("wait-for-interest", { subject, timeout });
+      return response;
+    } catch (err) {
+      throw toConatError(err);
     }
-    return response;
   };
 
   recvStats = (bytes: number) => {
@@ -954,7 +971,7 @@ export class Client extends EventEmitter {
             });
           }
         } catch (err) {
-          throw new ConatError(`${err}`, { code: 408 });
+          throw toConatError(err);
         }
         if (response?.error) {
           throw new ConatError(response.error, { code: response.code });
@@ -1300,9 +1317,7 @@ export class Client extends EventEmitter {
                 return response;
               }
             } catch (err) {
-              throw new ConatError(`timeout - ${subject} - ${err}`, {
-                code: 408,
-              });
+              throw toConatError(err);
             }
           } else {
             return await this.conn.emitWithAck("publish", v);
@@ -1910,3 +1925,15 @@ function isEmpty(obj: object): boolean {
   }
   return true;
 }
+
+function toConatError(socketIoError) {
+  // only errors are "disconnected" and a timeout
+  const e = `${socketIoError}`;
+  if (e.includes("disconnected")) {
+    return e;
+  } else {
+    return new ConatError(`timeout - ${e}`, {
+      code: 408,
+    });
+  }
+}

src/packages/conat/core/server.ts

@@ -743,7 +743,9 @@ export class ConatServer extends EventEmitter {
           });
           if (target !== undefined) {
             this.io.to(target).emit(pattern, { subject, data });
-            count += 1;
+            if (!isSilentPattern(pattern)) {
+              count++;
+            }
           }
         }
       }
@@ -773,7 +775,9 @@ export class ConatServer extends EventEmitter {
           if (id == this.id) {
             // another client of this server
             this.io.to(target).emit(pattern, { subject, data });
-            count += 1;
+            if (!isSilentPattern(pattern)) {
+              count++;
+            }
           } else {
             // client connected to a different server -- we note this, and
             // will send everything for each server at once, instead of
@@ -803,7 +807,11 @@ export class ConatServer extends EventEmitter {
       // use explicit index since length of data depends on
       // whether or not there are headers!
       data1[7] = outsideTargets[id];
-      count += 1;
+      for (const { pattern } of data1[7]) {
+        if (!isSilentPattern(pattern)) {
+          count++;
+        }
+      }
       link?.client.conn.emit("publish", data1);
     }
 
@@ -1797,6 +1805,17 @@ function getServerAddress(options: Options) {
   return `http${options.ssl || port == 443 ? "s" : ""}://${options.clusterIpAddress ?? "localhost"}:${port}${path}`;
 }
 
+/*
+Without this, if an admin subscribed to '>' then suddenly all the algorithms
+for responding to messages, sockets, etc., based on "waiting for interest"
+would start failing.  The following is a hack to allow subscribing to '>'.
+Really we need something more general for other admin "wire taps", but
+this will have to do for now.
+*/
+function isSilentPattern(pattern: string): boolean {
+  return pattern == ">";
+}
+
 /*
 const watching = new Set(["xyz"]);
 let last = Date.now();