fix: ensure dynamic worker can't be created at pool destroy

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

Author: jerome-benoit

SECURITY.md

@@ -6,7 +6,7 @@ Security matrix, currently there are no security vulnerabilities.
 
 | Version | Supported |
 | ------- | --------- |
-| 1.x.x   | ✔         |
+| 0.5.x   | ✔         |
 
 ## Reporting a Vulnerability
 

src/pools/abstract-pool.ts

@@ -104,6 +104,19 @@ export abstract class AbstractPool<
     Response
   >
 
+  /**
+   * Whether the pool is started or not.
+   */
+  protected started: boolean
+  /**
+   * Whether the pool is starting or not.
+   */
+  protected starting: boolean
+  /**
+   * Whether the pool is destroying or not.
+   */
+  protected destroying: boolean
+
   /**
    * The task functions added at runtime map:
    * - `key`: The task function name.
@@ -114,18 +127,6 @@ export abstract class AbstractPool<
     TaskFunctionObject<Data, Response>
   >
 
-  /**
-   * Whether the pool is started or not.
-   */
-  private started: boolean
-  /**
-   * Whether the pool is starting or not.
-   */
-  private starting: boolean
-  /**
-   * Whether the pool is destroying or not.
-   */
-  private destroying: boolean
   /**
    * Whether the minimum number of workers is starting or not.
    */
@@ -757,11 +758,17 @@ export abstract class AbstractPool<
 
   private isWorkerNodeBackPressured(workerNodeKey: number): boolean {
     const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode == null) {
+      return false
+    }
     return workerNode.info.ready && workerNode.info.backPressure
   }
 
   private isWorkerNodeBusy(workerNodeKey: number): boolean {
     const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode == null) {
+      return false
+    }
     if (this.opts.enableTasksQueue === true) {
       return (
         workerNode.info.ready &&
@@ -774,6 +781,9 @@ export abstract class AbstractPool<
 
   private isWorkerNodeIdle(workerNodeKey: number): boolean {
     const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode == null) {
+      return false
+    }
     if (this.opts.enableTasksQueue === true) {
       return (
         workerNode.info.ready &&
@@ -786,6 +796,9 @@ export abstract class AbstractPool<
 
   private isWorkerNodeStealing(workerNodeKey: number): boolean {
     const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode == null) {
+      return false
+    }
     return (
       workerNode.info.ready &&
       (workerNode.info.continuousStealing ||
@@ -802,10 +815,6 @@ export abstract class AbstractPool<
       | undefined
     try {
       return await new Promise<boolean>((resolve, reject) => {
-        if (this.workerNodes[workerNodeKey] == null) {
-          resolve(true)
-          return
-        }
         taskFunctionOperationListener = (
           message: MessageValue<Response>,
         ): void => {
@@ -885,10 +894,6 @@ export abstract class AbstractPool<
     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(
@@ -1333,18 +1338,29 @@ export abstract class AbstractPool<
     this.started = false
   }
 
-  private async sendKillMessageToWorker(workerNodeKey: number): Promise<void> {
+  private async sendKillMessageToWorker(
+    workerNodeKey: number,
+    timeout = 1000,
+  ): Promise<void> {
+    let timeoutHandle: number | undefined
     let killMessageListener:
       | ((message: MessageValue<Response>) => void)
       | undefined
     try {
       await new Promise<void>((resolve, reject) => {
-        if (this.workerNodes[workerNodeKey] == null) {
-          resolve()
-          return
-        }
+        timeoutHandle = timeout >= 0
+          ? setTimeout(() => {
+            resolve()
+          }, timeout)
+          : undefined
         killMessageListener = (message: MessageValue<Response>): void => {
-          this.checkMessageWorkerId(message)
+          if (
+            this.workerNodes.length === 0 ||
+            this.workerNodes[workerNodeKey] == null
+          ) {
+            resolve()
+            return
+          }
           if (message.kill === 'success') {
             resolve()
           } else if (message.kill === 'failure') {
@@ -1359,6 +1375,9 @@ export abstract class AbstractPool<
         this.sendToWorker(workerNodeKey, { kill: true })
       })
     } finally {
+      if (timeoutHandle != null) {
+        clearTimeout(timeoutHandle)
+      }
       if (killMessageListener != null) {
         this.deregisterWorkerMessageListener(workerNodeKey, killMessageListener)
       }
@@ -1819,7 +1838,12 @@ export abstract class AbstractPool<
   }
 
   private cannotStealTask(): boolean {
-    return this.workerNodes.length <= 1 || this.info.queuedTasks === 0
+    return (
+      !this.started ||
+      this.destroying ||
+      this.workerNodes.length <= 1 ||
+      this.info.queuedTasks === 0
+    )
   }
 
   private handleTask(workerNodeKey: number, task: Task<Data>): void {
@@ -2051,6 +2075,9 @@ export abstract class AbstractPool<
     const { workerId } = event.detail
     const sourceWorkerNode =
       this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
+    if (sourceWorkerNode == null) {
+      return
+    }
     const workerNodes = this.workerNodes
       .slice()
       .sort(
@@ -2088,12 +2115,18 @@ export abstract class AbstractPool<
   protected readonly workerMessageListener = (
     message: MessageValue<Response>,
   ): void => {
-    if (message.kill != null) {
+    const { kill, ready, taskFunctionsProperties, taskId, workerId } = message
+    const workerReadyMessage = ready != null && taskFunctionsProperties != null
+    // Late worker ready message received
+    if (this.destroying && workerReadyMessage) {
+      return
+    }
+    // Kill messages response are handled in dedicated listeners
+    if (kill != null) {
       return
     }
     this.checkMessageWorkerId(message)
-    const { workerId, ready, taskId, taskFunctionsProperties } = message
-    if (ready != null && taskFunctionsProperties != null) {
+    if (workerReadyMessage) {
       // Worker ready response received from worker
       this.handleWorkerReadyResponse(message)
     } else if (taskFunctionsProperties != null) {
@@ -2282,7 +2315,7 @@ export abstract class AbstractPool<
   private readonly abortTask = (
     event: CustomEvent<WorkerNodeEventDetail>,
   ): void => {
-    if (!this.started || this.destroying) {
+    if (!this.started) {
       return
     }
     const { taskId, workerId } = event.detail

src/pools/selection-strategies/abstract-worker-choice-strategy.ts

@@ -117,6 +117,7 @@ export abstract class AbstractWorkerChoiceStrategy<
 
   /**
    * Check the worker node key.
+   *
    * @param workerNodeKey - The worker node key to check.
    * @returns The worker node key if it is valid, otherwise undefined.
    */

src/pools/thread/dynamic.ts

@@ -52,7 +52,11 @@ export class DynamicThreadPool<
 
   /** @inheritDoc */
   protected override shallCreateDynamicWorker(): boolean {
-    return (!this.full && this.internalBusy()) || this.empty
+    return (
+      this.started &&
+      !this.destroying &&
+      ((!this.full && this.internalBusy()) || this.empty)
+    )
   }
 
   /** @inheritDoc */

src/pools/thread/fixed.ts

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

tests/pools/abstract-pool.test.mjs

@@ -1440,7 +1440,10 @@ describe({
       const elapsedTime = performance.now() - startTime
       expect(tasksFinished).toBeLessThanOrEqual(numberOfWorkers * maxMultiplier)
       expect(elapsedTime).toBeGreaterThanOrEqual(2000)
-      expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 100)
+      // Worker kill message response timeout is 1000ms
+      expect(elapsedTime).toBeLessThanOrEqual(
+        tasksFinishedTimeout + 1000 * tasksFinished + 100,
+      )
     })
 
     it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
@@ -1468,7 +1471,9 @@ describe({
       await pool.destroy()
       const elapsedTime = performance.now() - startTime
       expect(tasksFinished).toBe(0)
-      expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 100)
+      expect(elapsedTime).toBeLessThanOrEqual(
+        tasksFinishedTimeout + 1000 * tasksFinished + 100,
+      )
     })
 
     it('Verify that hasTaskFunction() is working', async () => {