feat: Sharded pub/sub support via dedicated subscribers (#1956)

* Added sharded pub/sub support by implementing a cluster subscriber group
* Rewrote the resubscribe logic for sharded PubSub by making the sharded subscriber group aware of the channels.
* Fixed potentially leaking connections when calling disconnect on the cluster object
* Added and extended the integration test cases in regards to sharded pubsub.
* Added a Github action that allows running tests manually
* Provided documentation about how to use sharded pub/sub

---------

Co-authored-by: Tihomir Krasimirov Mateev <[email protected]>

Author: dmaier-redislabs

.github/workflows/release.yml

@@ -8,7 +8,7 @@ concurrency:
 
 jobs:
   test:
-    uses: ./.github/workflows/test.yml
+    uses: ./.github/workflows/test_with_cov.yml
   release:
     runs-on: ubuntu-latest
     needs: test

.github/workflows/test.yml

@@ -11,7 +11,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        node: [12.x, 14.x, 16.x, 18.x]
+        node: [12.x, 14.x, 16.x, 18.x, 20.x]
     steps:
       - name: Git checkout
         uses: actions/checkout@v2
@@ -33,27 +33,3 @@ jobs:
       - run: npm run build
       - run: npm run test:tsd
       - run: npm run test:cov || npm run test:cov || npm run test:cov
-      - name: Coveralls
-        if: matrix.node == '18.x'
-        uses: coverallsapp/github-action@master
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          flag-name: node-${{matrix.node}}
-          parallel: true
-
-  # test-cluster:
-  #   runs-on: ubuntu-latest
-  #   steps:
-  #     - uses: actions/checkout@v2
-  #     - name: Build and test cluster
-  #       run: bash test/cluster/docker/main.sh
-
-  code-coverage:
-    needs: test
-    runs-on: ubuntu-latest
-    steps:
-      - name: Coveralls
-        uses: coverallsapp/github-action@master
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          parallel-finished: true

.github/workflows/test_with_cov.yml

@@ -7,7 +7,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        node: [12.x, 14.x, 16.x, 18.x]
+        node: [12.x, 14.x, 16.x, 18.x, 20.x]
     steps:
       - name: Git checkout
         uses: actions/checkout@v2

README.md

@@ -46,7 +46,7 @@ used in the world's biggest online commerce company [Alibaba](http://www.alibaba
 | Version        | Branch | Node.js Version | Redis Version   |
 | -------------- | ------ | --------------- | --------------- |
 | 5.x.x (latest) | main   | >= 12           | 2.6.12 ~ latest |
-| 4.x.x          | v4     | >= 6            | 2.6.12 ~ 7      |
+| 4.x.x          | v4     | >= 8            | 2.6.12 ~ 7      |
 
 Refer to [CHANGELOG.md](CHANGELOG.md) for features and bug fixes introduced in v5.
 
@@ -1196,6 +1196,38 @@ sub.subscribe("news", () => {
 });
 ```
 
+### Sharded Pub/Sub
+
+For sharded Pub/Sub, use the `spublish` and `ssubscribe` commands instead of the traditional `publish` and `subscribe`. With the old commands, the Redis cluster handles message propagation behind the scenes, allowing you to publish or subscribe to any node without considering sharding. However, this approach has scalability limitations that are addressed with sharded Pub/Sub. Here’s what you need to know:
+
+1. Instead of a single subscriber connection, there is now one subscriber connection per shard. Because of the potential overhead, you can enable or disable the use of the cluster subscriber group with the `shardedSubscribers` option. By default, this option is set to `false`, meaning sharded subscriptions are disabled. You should enable this option when establishing your cluster connection before using `ssubscribe`.
+2. All channel names that you pass to a single `ssubscribe` need to map to the same hash slot. You can call `ssubscribe` multiple times on the same cluster client instance to subscribe to channels across slots. The cluster's subscriber group takes care of forwarding the `ssubscribe` command to the shard that is responsible for the channels.
+
+The following basic example shows you how to use sharded Pub/Sub:
+
+```javascript
+const cluster: Cluster = new Cluster([{host: host, port: port}], {shardedSubscribers: true});
+
+//Register the callback
+cluster.on("smessage", (channel, message) => {
+    console.log(message);
+});
+
+        
+//Subscribe to the channels on the same slot
+cluster.ssubscribe("channel{my}:1", "channel{my}:2").then( ( count: number ) => {
+    console.log(count);
+}).catch( (err) => {
+    console.log(err);
+});
+
+//Publish a message
+cluster.spublish("channel{my}:1", "This is a test message to my first channel.").then((value: number) => {
+    console.log("Published a message to channel{my}:1");
+});
+```
+
+
 ### Events
 
 | Event        | Description                                                                                                                                                                                                |

lib/cluster/ClusterOptions.ts

@@ -121,6 +121,17 @@ export interface ClusterOptions extends CommanderOptions {
    */
   slotsRefreshInterval?: number;
 
+
+  /**
+   * Use sharded subscribers instead of a single subscriber.
+   *
+   * If sharded subscribers are used, then one additional subscriber connection per master node
+   * is established. If you don't plan to use SPUBLISH/SSUBSCRIBE, then this should be disabled.
+   *
+   * @default false
+   */
+  shardedSubscribers?: boolean;
+
   /**
    * Passed to the constructor of `Redis`
    *
@@ -216,4 +227,5 @@ export const DEFAULT_CLUSTER_OPTIONS: ClusterOptions = {
   dnsLookup: lookup,
   enableAutoPipelining: false,
   autoPipeliningIgnoredCommands: [],
+  shardedSubscribers: false,
 };

lib/cluster/ClusterSubscriber.ts

@@ -8,12 +8,18 @@ const debug = Debug("cluster:subscriber");
 
 export default class ClusterSubscriber {
   private started = false;
+
+  //There is only one connection for the entire pool
   private subscriber: any = null;
   private lastActiveSubscriber: any;
 
+  //The slot range for which this subscriber is responsible
+  private slotRange: number[] = []
+
   constructor(
     private connectionPool: ConnectionPool,
-    private emitter: EventEmitter
+    private emitter: EventEmitter,
+    private isSharded : boolean = false
   ) {
     // If the current node we're using as the subscriber disappears
     // from the node pool for some reason, we will select a new one
@@ -47,6 +53,22 @@ export default class ClusterSubscriber {
     return this.subscriber;
   }
 
+  /**
+   * Associate this subscriber to a specific slot range.
+   *
+   * Returns the range or an empty array if the slot range couldn't be associated.
+   *
+   * BTW: This is more for debugging and testing purposes.
+   *
+   * @param range
+   */
+  associateSlotRange(range: number[]): number[] {
+    if (this.isSharded) {
+      this.slotRange = range;
+    }
+    return this.slotRange;
+  }
+
   start(): void {
     this.started = true;
     this.selectSubscriber();
@@ -59,9 +81,13 @@ export default class ClusterSubscriber {
       this.subscriber.disconnect();
       this.subscriber = null;
     }
-    debug("stopped");
   }
 
+  isStarted(): boolean {
+    return this.started;
+  }
+
+
   private onSubscriberEnd = () => {
     if (!this.started) {
       debug(
@@ -112,13 +138,17 @@ export default class ClusterSubscriber {
      * provided for the subscriber is correct, and if not, the current subscriber
      * will be disconnected and a new subscriber will be selected.
      */
+    let connectionPrefix = "subscriber";
+    if (this.isSharded)
+      connectionPrefix = "ssubscriber";
+
     this.subscriber = new Redis({
       port: options.port,
       host: options.host,
       username: options.username,
       password: options.password,
       enableReadyCheck: true,
-      connectionName: getConnectionName("subscriber", options.connectionName),
+      connectionName: getConnectionName(connectionPrefix, options.connectionName),
       lazyConnect: true,
       tls: options.tls,
       // Don't try to reconnect the subscriber connection. If the connection fails
@@ -179,17 +209,27 @@ export default class ClusterSubscriber {
     for (const event of [
       "message",
       "messageBuffer",
-      "smessage",
-      "smessageBuffer",
     ]) {
       this.subscriber.on(event, (arg1, arg2) => {
         this.emitter.emit(event, arg1, arg2);
       });
     }
+
     for (const event of ["pmessage", "pmessageBuffer"]) {
       this.subscriber.on(event, (arg1, arg2, arg3) => {
         this.emitter.emit(event, arg1, arg2, arg3);
       });
     }
+
+    if (this.isSharded == true) {
+      for (const event of [
+        "smessage",
+        "smessageBuffer",
+      ]) {
+        this.subscriber.on(event, (arg1, arg2) => {
+          this.emitter.emit(event, arg1, arg2);
+        });
+      }
+    }
   }
 }