fix: prevent race condition when subscribe command issued during connect

Author: PavelPashov

lib/DataHandler.ts

@@ -14,6 +14,10 @@ export interface Condition {
   select: number;
   auth?: string | [string, string];
   subscriber: false | SubscriptionSet;
+  /**
+   * Whether the connection has issued a subscribe command during `connect` or `ready`.
+   */
+  hasIssuedSubscribe: boolean;
 }
 
 export type FlushQueueOptions = {
@@ -89,6 +93,7 @@ export default class DataHandler {
   }
 
   private returnReply(reply: ReplyData) {
+    console.log('reply');
     if (this.handleMonitorReply(reply)) {
       return;
     }
@@ -97,6 +102,7 @@ export default class DataHandler {
     }
 
     const item = this.shiftCommand(reply);
+    console.log('reply command', item.command.name);
     if (!item) {
       return;
     }

lib/Redis.ts

@@ -195,6 +195,7 @@ class Redis extends Commander implements DataHandledable {
           ? [options.username, options.password]
           : options.password,
         subscriber: false,
+        hasIssuedSubscribe: false,
       };
 
       const _this = this;
@@ -429,6 +430,17 @@ class Redis extends Commander implements DataHandledable {
       command.reject(new Error(CONNECTION_CLOSED_ERROR_MSG));
       return command.promise;
     }
+
+    // Make sure know that a subscribe command is sent to the server
+    // In order to prevent race condition by sending another non-subscribe command
+    // before we've received the response of the previous subscribe command
+    if (
+      (this.status === "connect" || this.status === "ready") &&
+      Command.checkFlag("ENTER_SUBSCRIBER_MODE", command.name)
+    ) {
+      this.condition.hasIssuedSubscribe = true;
+    }
+
     if (
       this.condition?.subscriber &&
       !Command.checkFlag("VALID_IN_SUBSCRIBER_MODE", command.name)
@@ -527,7 +539,10 @@ class Redis extends Commander implements DataHandledable {
         this.manuallyClosing = true;
       }
 
-      if (this.options.socketTimeout !== undefined && this.socketTimeoutTimer === undefined) {
+      if (
+        this.options.socketTimeout !== undefined &&
+        this.socketTimeoutTimer === undefined
+      ) {
         this.setSocketTimeout();
       }
     }
@@ -546,7 +561,11 @@ class Redis extends Commander implements DataHandledable {
 
   private setSocketTimeout() {
     this.socketTimeoutTimer = setTimeout(() => {
-      this.stream.destroy(new Error(`Socket timeout. Expecting data, but didn't receive any in ${this.options.socketTimeout}ms.`));
+      this.stream.destroy(
+        new Error(
+          `Socket timeout. Expecting data, but didn't receive any in ${this.options.socketTimeout}ms.`
+        )
+      );
       this.socketTimeoutTimer = undefined;
     }, this.options.socketTimeout);
 

lib/redis/event_handler.ts

@@ -77,14 +77,14 @@ export function connectHandler(self) {
 
     const clientCommandPromises = [];
 
-    if (self.options.connectionName) {
+    if (self.options.connectionName && !self.condition.hasIssuedSubscribe) {
       debug("set the connection name [%s]", self.options.connectionName);
       clientCommandPromises.push(
         self.client("setname", self.options.connectionName).catch(noop)
       );
     }
 
-    if (!self.options.disableClientInfo) {
+    if (!self.options.disableClientInfo && !self.condition.hasIssuedSubscribe) {
       debug("set the client info");
       clientCommandPromises.push(
         getPackageMeta()
@@ -112,11 +112,16 @@ export function connectHandler(self) {
     Promise.all(clientCommandPromises)
       .catch(noop)
       .finally(() => {
-        if (!self.options.enableReadyCheck) {
+        // Ready check should not be performed after a subscribe command
+        // Because it might result in a race condition
+        // Additionally because we're using RESP2 another client should be used for normal commands
+        const shouldReadyCheck = self.options.enableReadyCheck && !self.condition.hasIssuedSubscribe;
+
+        if (!shouldReadyCheck) {
           exports.readyHandler(self)();
         }
 
-        if (self.options.enableReadyCheck) {
+        if (shouldReadyCheck) {
           self._readyCheck(function (err, info) {
             if (connectionEpoch !== self.connectionEpoch) {
               return;