fix: properly handle task abortion worker side

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

Author: jerome-benoit

src/worker/abstract-worker.ts

@@ -342,16 +342,17 @@ export abstract class AbstractWorker<
   protected messageEventListener(
     messageEvent: MessageEvent<MessageValue<Data>>,
   ): void {
-    const { data } = messageEvent
-    this.checkMessageWorkerId(data)
+    const { data: messageData } = messageEvent
+    this.checkMessageWorkerId(messageData)
     const {
+      data,
       statistics,
       checkActive,
       taskFunctionOperation,
       taskId,
       taskOperation,
       kill,
-    } = data
+    } = messageData
     if (statistics != null) {
       // Statistics message received
       this.statistics = statistics
@@ -360,18 +361,18 @@ export abstract class AbstractWorker<
       checkActive ? this.startCheckActive() : this.stopCheckActive()
     } else if (taskFunctionOperation != null) {
       // Task function operation message received
-      this.handleTaskFunctionOperationMessage(data)
+      this.handleTaskFunctionOperationMessage(messageData)
     } else if (taskId != null && data != null) {
       // Task message received
-      this.run(data)
+      this.run(messageData)
     } else if (taskOperation === 'abort' && taskId != null) {
       // Abort task operation message received
       if (this.taskAbortFunctions.has(taskId)) {
         this.taskAbortFunctions.get(taskId)?.()
       }
     } else if (kill === true) {
       // Kill message received
-      this.handleKillMessage(data)
+      this.handleKillMessage(messageData)
     }
   }
 

tests/pools/abstract-pool.test.mjs

@@ -1943,7 +1943,11 @@ describe({
       await expect(pool.mapExecute([undefined], 'unknown')).rejects.toThrow(
         new Error("Task function 'unknown' not found"),
       )
-      let results = await pool.mapExecute([{}, {}, {}, {}])
+      let results = await pool.mapExecute(
+        Array(4).fill({}),
+        'jsonIntegerSerialization',
+        Array(4).fill(AbortSignal.timeout(1000)),
+      )
       expect(results).toStrictEqual([
         { ok: 1 },
         { ok: 1 },
@@ -1964,6 +1968,7 @@ describe({
           },
         ],
         'factorial',
+        Array(4).fill(AbortSignal.timeout(1000)),
       )
       expect(results).toStrictEqual([
         3628800n,
@@ -1976,6 +1981,12 @@ describe({
       results = await pool.mapExecute(
         new Set([{ n: 10 }, { n: 20 }, { n: 30 }, { n: 40 }]),
         'factorial',
+        new Set([
+          AbortSignal.timeout(1000),
+          AbortSignal.timeout(1500),
+          AbortSignal.timeout(2000),
+          AbortSignal.timeout(2500),
+        ]),
       )
       expect(results).toStrictEqual([
         3628800n,

tests/pools/thread/dynamic.test.mjs

@@ -33,13 +33,21 @@ describe({
     })
 
     it('Verify that the function is executed in a worker thread', async () => {
-      let result = await pool.execute({
-        function: TaskFunctions.fibonacci,
-      })
+      let result = await pool.execute(
+        {
+          function: TaskFunctions.fibonacci,
+        },
+        'default',
+        AbortSignal.timeout(2000),
+      )
       expect(result).toBe(354224848179261915075n)
-      result = await pool.execute({
-        function: TaskFunctions.factorial,
-      })
+      result = await pool.execute(
+        {
+          function: TaskFunctions.factorial,
+        },
+        'default',
+        AbortSignal.timeout(2000),
+      )
       expect(result).toBe(
         93326215443944152681699238856266700490715968264381621468592963895217599993229915608941463976156518286253697920827223758251185210916864000000000000000000000000n,
       )

tests/pools/thread/fixed.test.mjs

@@ -81,13 +81,21 @@ describe({
     })
 
     it('Verify that the function is executed in a worker thread', async () => {
-      let result = await pool.execute({
-        function: TaskFunctions.fibonacci,
-      })
+      let result = await pool.execute(
+        {
+          function: TaskFunctions.fibonacci,
+        },
+        'default',
+        AbortSignal.timeout(2000),
+      )
       expect(result).toBe(354224848179261915075n)
-      result = await pool.execute({
-        function: TaskFunctions.factorial,
-      })
+      result = await pool.execute(
+        {
+          function: TaskFunctions.factorial,
+        },
+        'default',
+        AbortSignal.timeout(2000),
+      )
       expect(result).toBe(
         93326215443944152681699238856266700490715968264381621468592963895217599993229915608941463976156518286253697920827223758251185210916864000000000000000000000000n,
       )
@@ -258,18 +266,17 @@ describe({
       ).toBe(true)
     })
 
-    it.only('Verify that task can be aborted', async () => {
+    it('Verify that task can be aborted', async () => {
       let error
 
       try {
         await asyncErrorPool.execute({}, 'default', AbortSignal.timeout(500))
       } catch (e) {
         error = e
       }
-      console.info(error)
       expect(error).toBeInstanceOf(Error)
       expect(error.name).toBe('TimeoutError')
-      expect(error.message).toBe('The operation was aborted due to timeout')
+      expect(error.message).toBe('Signal timed out.')
       expect(error.stack).toBeDefined()
 
       const abortController = new AbortController()

tests/utils.test.mjs

@@ -168,10 +168,10 @@ describe('Utils test suite', () => {
     expect(isAsyncFunction({})).toBe(false)
     expect(isAsyncFunction({ a: 1 })).toBe(false)
     expect(isAsyncFunction(() => {})).toBe(false)
-    expect(isAsyncFunction(() => {})).toBe(false)
+    expect(isAsyncFunction(function () {})).toBe(false)
     expect(isAsyncFunction(function named() {})).toBe(false)
     expect(isAsyncFunction(async () => {})).toBe(true)
-    expect(isAsyncFunction(async () => {})).toBe(true)
+    expect(isAsyncFunction(async function () {})).toBe(true)
     expect(isAsyncFunction(async function named() {})).toBe(true)
     class TestClass {
       testSync() {}