fix: alive timeout check & start/destroy queue race condition

Author: SerhiyGreench

src/ClusterManager.ts

@@ -29,6 +29,7 @@ export interface ICluster {
     remove: (server: IServerInput) => void;
     find: <T extends IMessageQueueConnection>(
         server: IServerInput,
+        strict?: boolean,
     ) => T | undefined;
 }
 
@@ -49,6 +50,12 @@ export abstract class ClusterManager {
         return initializedCluster;
     }
 
+    public async anyCluster(
+        fn: (cluster: InitializedCluster) => Promise<void> | void,
+    ): Promise<void> {
+        await Promise.all(this.clusters.map(cluster => fn(cluster)));
+    }
+
     public async remove(
         cluster: string | InitializedCluster,
         destroy: boolean = true,

src/ClusteredRedisQueue.ts

@@ -23,18 +23,18 @@
  */
 import { EventEmitter } from 'events';
 import {
-    DEFAULT_IMQ_OPTIONS,
     buildOptions,
+    copyEventEmitter,
+    DEFAULT_IMQ_OPTIONS,
+    EventMap,
     ILogger,
     IMessageQueue,
     IMessageQueueConnection,
     IMQMode,
     IMQOptions,
+    IServerInput,
     JsonObject,
     RedisQueue,
-    EventMap,
-    IServerInput,
-    copyEventEmitter,
 } from '.';
 import { InitializedCluster } from './ClusterManager';
 
@@ -496,7 +496,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
      * @returns {void}
      */
     protected removeServer(server: IServerInput): void {
-        const remove = this.findServer(server);
+        const remove = this.findServer(server, true);
 
         if (!remove) {
             return;
@@ -534,14 +534,12 @@ export class ClusteredRedisQueue implements IMessageQueue,
         const imq = new RedisQueue(this.name, opts);
 
         if (initializeQueue) {
-            this.initializeQueue(imq)
-                .then(() => {
-                    this.clusterEmitter.emit('initialized', {
-                        server: newServer,
-                        imq,
-                    });
-                })
-                .catch();
+            this.initializeQueue(imq).then(() => {
+                this.clusterEmitter.emit('initialized', {
+                    server: newServer,
+                    imq,
+                });
+            });
         }
 
         newServer.imq = imq;
@@ -571,18 +569,23 @@ export class ClusteredRedisQueue implements IMessageQueue,
         }
     }
 
-    private findServer(server: IServerInput): ClusterServer | undefined {
+    private findServer(
+        server: IServerInput,
+        strict: boolean = false,
+    ): ClusterServer | undefined {
         return this.servers.find(
             existing => ClusteredRedisQueue.matchServers(
                 existing,
                 server,
+                strict,
             ),
         );
     }
 
     private static matchServers(
         source: IServerInput,
         target: IServerInput,
+        strict: boolean = false,
     ): boolean {
         const sameAddress = target.host === source.host
             && target.port === source.port;
@@ -593,6 +596,11 @@ export class ClusteredRedisQueue implements IMessageQueue,
 
         const sameId = target.id === source.id;
 
+        if (strict) {
+            return sameId && sameAddress;
+        }
+
         return sameId || sameAddress;
     }
+
 }

src/RedisQueue.ts

@@ -180,7 +180,7 @@ export class RedisQueue extends EventEmitter<EventMap>
     private destroyed: boolean = false;
 
     /**
-     * True if the current instance owns watcher connection, false otherwise
+     * True if the current instance owns a watcher connection, false otherwise
      *
      * @type {boolean}
      */
@@ -369,6 +369,8 @@ export class RedisQueue extends EventEmitter<EventMap>
             return this;
         }
 
+        this.destroyed = false;
+
         const connPromises = [];
 
         // istanbul ignore next
@@ -408,7 +410,6 @@ export class RedisQueue extends EventEmitter<EventMap>
 
         await this.initWatcher();
         this.initialized = true;
-        this.destroyed = false;
 
         return this;
     }
@@ -586,7 +587,7 @@ export class RedisQueue extends EventEmitter<EventMap>
 
     // noinspection JSUnusedLocalSymbols
     /**
-     * Watcher setter, sets the watcher connection property for this
+     * Watcher setter sets the watcher connection property for this
      * queue instance
      *
      * @param {IRedisClient} conn
@@ -864,13 +865,17 @@ export class RedisQueue extends EventEmitter<EventMap>
     }
 
     /**
-     * Returns number of established watcher connections on redis
+     * Returns the number of established watcher connections on redis
      *
      * @access private
      * @returns {Promise<number>}
      */
     // istanbul ignore next
     private async watcherCount(): Promise<number> {
+        if (!this.writer) {
+            return 0;
+        }
+
         const rx = new RegExp(
             `\\bname=${this.options.prefix}:[\\S]+?:watcher:`,
         );
@@ -883,7 +888,7 @@ export class RedisQueue extends EventEmitter<EventMap>
     }
 
     /**
-     * Processes delayed message by its given redis key
+     * Processes delayed a message by its given redis key
      *
      * @access private
      * @param {string} key

src/UDPClusterManager.ts

@@ -55,15 +55,6 @@ export const DEFAULT_UDP_CLUSTER_MANAGER_OPTIONS = {
 };
 
 export interface UDPClusterManagerOptions {
-    /**
-     * Represents the cluster operations that are responsible for managing
-     * clusters. This includes operations such as adding, removing, or checking
-     * if a cluster server exists.
-     *
-     * @type {ICluster}
-     */
-    cluster?: ICluster;
-
     /**
      * Message queue broadcast port
      *
@@ -224,7 +215,7 @@ export class UDPClusterManager extends ClusterManager {
         if (!server && message.type === MessageType.Up) {
             cluster.add(message);
 
-            const added = cluster.find<ClusterServer>(message);
+            const added = cluster.find<ClusterServer>(message, true);
 
             if (added) {
                 UDPClusterManager.serverAliveWait(
@@ -271,13 +262,13 @@ export class UDPClusterManager extends ClusterManager {
                  return;
             }
 
-            for (const cluster of context.clusters) {
+            context.anyCluster(cluster => {
                 UDPClusterManager.processMessageOnCluster(
                     cluster,
                     message,
                     context.options.aliveTimeoutCorrection,
                 );
-            }
+            }).then();
         };
     }
 
@@ -307,7 +298,11 @@ export class UDPClusterManager extends ClusterManager {
         aliveTimeoutCorrection?: number,
         message?: Message,
     ): void {
-        clearTimeout(server.timer);
+        if (server.timer) {
+            clearTimeout(server.timer);
+            server.timer = undefined;
+        }
+
         server.timestamp = Date.now();
 
         if (message) {
@@ -317,23 +312,38 @@ export class UDPClusterManager extends ClusterManager {
         const correction = aliveTimeoutCorrection || 0;
         const timeout = (server.timeout || 0) + correction;
 
-        server.timer = setTimeout(() => {
-            const existing = cluster.find<ClusterServer>(server);
+        if (timeout <= 0) {
+            return;
+        }
+
+        const timerId = setTimeout(() => {
+            const existing = cluster.find<ClusterServer>(server, true);
 
-            if (!existing) {
+            if (!existing || existing.timer !== timerId) {
                 return;
             }
 
             const now = Date.now();
-            const delta = now - (existing.timestamp || now);
+
+            if (!existing.timestamp) {
+                clearTimeout(existing.timer);
+                existing.timer = undefined;
+                cluster.remove(existing);
+
+                return;
+            }
+
+            const delta = now - existing.timestamp;
             const currentTimeout = (existing.timeout || 0) + correction;
 
             if (delta >= currentTimeout) {
-                clearTimeout(server.timer);
-
-                cluster.remove(server);
+                clearTimeout(existing.timer);
+                existing.timer = undefined;
+                cluster.remove(existing);
             }
         }, timeout);
+
+        server.timer = timerId;
     }
 
     /**