fix: fix potential ressource leaks

Signed-off-by: Jérôme Benoit <[email protected]>

Author: jerome-benoit

src/pools/abstract-pool.ts

@@ -793,52 +793,70 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     message: MessageValue<Data>,
   ): Promise<boolean> {
-    return await new Promise<boolean>((resolve, reject) => {
-      const taskFunctionOperationListener = (
-        message: MessageValue<Response>,
-      ): void => {
-        this.checkMessageWorkerId(message)
-        const workerId = this.getWorkerInfo(workerNodeKey)?.id
-        if (
-          message.taskFunctionOperationStatus != null &&
-          message.workerId === workerId
-        ) {
-          if (message.taskFunctionOperationStatus) {
-            resolve(true)
-          } else {
+    let taskFunctionOperationListener:
+      | ((message: MessageValue<Response>) => void)
+      | undefined
+    try {
+      return await new Promise<boolean>((resolve, reject) => {
+        if (this.workerNodes[workerNodeKey] == null) {
+          resolve(true)
+          return
+        }
+        taskFunctionOperationListener = (
+          message: MessageValue<Response>,
+        ): void => {
+          this.checkMessageWorkerId(message)
+          const workerId = this.getWorkerInfo(workerNodeKey)?.id
+          if (
+            message.taskFunctionOperationStatus != null &&
+            message.workerId === workerId
+          ) {
+            if (message.taskFunctionOperationStatus) {
+              resolve(true)
+              return
+            }
             reject(
               new Error(
                 `Task function operation '${message.taskFunctionOperation?.toString()}' failed on worker ${message.workerId?.toString()} with error: '${message.workerError?.error.message}'`,
               ),
             )
           }
-          this.deregisterWorkerMessageListener(
-            this.getWorkerNodeKeyByWorkerId(message.workerId),
-            taskFunctionOperationListener,
-          )
         }
+        this.registerWorkerMessageListener(
+          workerNodeKey,
+          taskFunctionOperationListener,
+        )
+        this.sendToWorker(workerNodeKey, message)
+      })
+    } finally {
+      if (taskFunctionOperationListener != null) {
+        this.deregisterWorkerMessageListener(
+          workerNodeKey,
+          taskFunctionOperationListener,
+        )
       }
-      this.registerWorkerMessageListener(
-        workerNodeKey,
-        taskFunctionOperationListener,
-      )
-      this.sendToWorker(workerNodeKey, message)
-    })
+    }
   }
 
   private async sendTaskFunctionOperationToWorkers(
     message: MessageValue<Data>,
   ): Promise<boolean> {
+    const targetWorkerNodeKeys = [...this.workerNodes.keys()]
     const taskFunctionOperationsListener = (
       message: MessageValue<Response>,
       resolve: (value: boolean | PromiseLike<boolean>) => void,
       reject: (reason?: unknown) => void,
       responsesReceived: MessageValue<Response>[],
     ): void => {
       this.checkMessageWorkerId(message)
-      if (message.taskFunctionOperationStatus != null) {
+      if (
+        message.taskFunctionOperationStatus != null &&
+        targetWorkerNodeKeys.includes(
+          this.getWorkerNodeKeyByWorkerId(message.workerId),
+        )
+      ) {
         responsesReceived.push(message)
-        if (responsesReceived.length >= this.workerNodes.length) {
+        if (responsesReceived.length >= targetWorkerNodeKeys.length) {
           if (
             responsesReceived.every(
               (msg) => msg.taskFunctionOperationStatus === true,
@@ -860,9 +878,13 @@ export abstract class AbstractPool<
         }
       }
     }
-    let listener: (message: MessageValue<Response>) => void
+    let listener: ((message: MessageValue<Response>) => void) | undefined
     try {
       return await new Promise<boolean>((resolve, reject) => {
+        if (targetWorkerNodeKeys.length === 0) {
+          resolve(true)
+          return
+        }
         const responsesReceived: MessageValue<Response>[] = []
         listener = (message: MessageValue<Response>) => {
           taskFunctionOperationsListener(
@@ -872,14 +894,16 @@ export abstract class AbstractPool<
             responsesReceived,
           )
         }
-        for (const workerNodeKey of this.workerNodes.keys()) {
+        for (const workerNodeKey of targetWorkerNodeKeys) {
           this.registerWorkerMessageListener(workerNodeKey, listener)
           this.sendToWorker(workerNodeKey, message)
         }
       })
     } finally {
-      for (const workerNodeKey of this.workerNodes.keys()) {
-        this.deregisterWorkerMessageListener(workerNodeKey, listener!)
+      if (listener != null) {
+        for (const workerNodeKey of targetWorkerNodeKeys) {
+          this.deregisterWorkerMessageListener(workerNodeKey, listener)
+        }
       }
     }
   }
@@ -1306,27 +1330,35 @@ export abstract class AbstractPool<
   }
 
   private async sendKillMessageToWorker(workerNodeKey: number): Promise<void> {
-    await new Promise<void>((resolve, reject) => {
-      if (this.workerNodes[workerNodeKey] == null) {
-        resolve()
-        return
-      }
-      const killMessageListener = (message: MessageValue<Response>): void => {
-        this.checkMessageWorkerId(message)
-        if (message.kill === 'success') {
+    let killMessageListener:
+      | ((message: MessageValue<Response>) => void)
+      | undefined
+    try {
+      await new Promise<void>((resolve, reject) => {
+        if (this.workerNodes[workerNodeKey] == null) {
           resolve()
-        } else if (message.kill === 'failure') {
-          reject(
-            new Error(
-              `Kill message handling failed on worker ${message.workerId?.toString()}`,
-            ),
-          )
+          return
         }
+        killMessageListener = (message: MessageValue<Response>): void => {
+          this.checkMessageWorkerId(message)
+          if (message.kill === 'success') {
+            resolve()
+          } else if (message.kill === 'failure') {
+            reject(
+              new Error(
+                `Kill message handling failed on worker ${message.workerId?.toString()}`,
+              ),
+            )
+          }
+        }
+        this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
+        this.sendToWorker(workerNodeKey, { kill: true })
+      })
+    } finally {
+      if (killMessageListener != null) {
+        this.deregisterWorkerMessageListener(workerNodeKey, killMessageListener)
       }
-      // FIXME: should be registered only once
-      this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
-      this.sendToWorker(workerNodeKey, { kill: true })
-    })
+    }
   }
 
   /**

src/pools/thread/fixed.ts

@@ -101,7 +101,7 @@ export class FixedThreadPool<
     workerNodeKey: number,
     listener: (message: MessageValue<Message>) => void,
   ): void {
-    this.workerNodes[workerNodeKey].removeEventListener(
+    this.workerNodes[workerNodeKey]?.removeEventListener(
       'message',
       messageListenerToEventListener<Message>(listener),
     )