From 83fa0a36c7d67362593915f3b19867f8fbbcbcce Mon Sep 17 00:00:00 2001
From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= <jerome.benoit@sap.com>
Date: Sat, 1 Jul 2023 00:55:35 +0200
Subject: [PATCH] fix: ensure worker started message cannot be received for non
 existing worker
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

Signed-off-by: Jérôme Benoit <jerome.benoit@sap.com>
---
 src/pools/abstract-pool.ts    | 29 ++++++++++++++++++++++++-----
 src/pools/worker.ts           |  4 ++--
 src/utility-types.ts          |  2 +-
 src/worker/abstract-worker.ts |  2 +-
 4 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 6fde2491..ba05d929 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -19,7 +19,8 @@ import {
   type PoolType,
   PoolTypes,
   type TasksQueueOptions,
-  type WorkerType
+  type WorkerType,
+  WorkerTypes
 } from './pool'
 import type {
   IWorker,
@@ -784,9 +785,13 @@ export abstract class AbstractPool<
     return message => {
       if (message.workerId != null && message.started != null) {
         // Worker started message received
-        this.workerNodes[
-          this.getWorkerNodeKey(this.getWorkerById(message.workerId) as Worker)
-        ].info.started = message.started
+        const worker = this.getWorkerById(message.workerId)
+        if (worker != null) {
+          this.workerNodes[this.getWorkerNodeKey(worker)].info.started =
+            message.started
+        } else {
+          throw new Error('Worker started message received from unknown worker')
+        }
       } else if (message.id != null) {
         // Task execution response received
         const promiseResponse = this.promiseResponseMap.get(message.id)
@@ -850,7 +855,7 @@ export abstract class AbstractPool<
   private pushWorkerNode (worker: Worker): number {
     this.workerNodes.push({
       worker,
-      info: { id: worker.threadId ?? worker.id, started: false },
+      info: { id: this.getWorkerId(worker), started: false },
       usage: this.getWorkerUsage(),
       tasksQueue: new Queue<Task<Data>>()
     })
@@ -862,6 +867,20 @@ export abstract class AbstractPool<
     return this.workerNodes.length
   }
 
+  /**
+   * Gets the worker id.
+   *
+   * @param worker - The worker.
+   * @returns The worker id.
+   */
+  private getWorkerId (worker: Worker): number | undefined {
+    if (this.worker === WorkerTypes.thread) {
+      return worker.threadId
+    } else if (this.worker === WorkerTypes.cluster) {
+      return worker.id
+    }
+  }
+
   // /**
   //  * Sets the given worker in the pool worker nodes.
   //  *
diff --git a/src/pools/worker.ts b/src/pools/worker.ts
index c92478d4..e0636d99 100644
--- a/src/pools/worker.ts
+++ b/src/pools/worker.ts
@@ -125,7 +125,7 @@ export interface TaskStatistics {
  */
 export interface WorkerInfo {
   /**
-   * Worker Id.
+   * Worker id.
    */
   id: number | undefined
   /**
@@ -163,7 +163,7 @@ export interface WorkerUsage {
  */
 export interface IWorker {
   /**
-   * Worker Id.
+   * Worker id.
    */
   id?: number
   threadId?: number
diff --git a/src/utility-types.ts b/src/utility-types.ts
index 1f724274..0443b4ce 100644
--- a/src/utility-types.ts
+++ b/src/utility-types.ts
@@ -54,7 +54,7 @@ export interface WorkerStatistics {
 export interface MessageValue<Data = unknown, ErrorData = unknown>
   extends Task<Data> {
   /**
-   * Worker Id.
+   * Worker id.
    */
   readonly workerId?: number
   /**
diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts
index 242c4698..0646a9bf 100644
--- a/src/worker/abstract-worker.ts
+++ b/src/worker/abstract-worker.ts
@@ -37,7 +37,7 @@ export abstract class AbstractWorker<
   Response = unknown
 > extends AsyncResource {
   /**
-   * Worker Id.
+   * Worker id.
    */
   protected abstract id: number
   /**
-- 
2.34.1