fix(#805): Fix handling of aborted tasks (#807)

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Author: metcoder95

src/index.ts

@@ -455,7 +455,7 @@ class ThreadPool {
   _distributeTask (task: TaskInfo, workers: PiscinaWorker[]): boolean {
     // We need to verify if the task is aborted already or not
     // otherwise we might be distributing aborted tasks to workers
-    if (task.aborted) return false;
+    if (task.aborted) return true;
 
     const candidate = this.balancer(task.interface, workers);
 

test/pool.test.ts

@@ -103,4 +103,31 @@ test('Explicit resource management (asyncDispose)', { skip: nodeVersion !== 24 }
       tasks.push(pool.run('new Promise(resolve => setTimeout(resolve, 500))'));
     }
   }
-})
+})
+
+test('#805 - Concurrent Aborts', async (t: TestContext) => {
+  const pool = new Piscina({
+    filename: resolve(__dirname, 'fixtures/eval.js'),
+    maxThreads: 1,
+    minThreads: 1,
+    concurrentTasksPerWorker: 1,
+  });
+
+  t.after(() => pool.close());
+
+  t.plan(2);
+  const tasks = [];
+  const controller = new AbortController();
+  const controller2 = new AbortController();
+  const controller3 = new AbortController();
+
+  tasks.push(t.assert.rejects(pool.run('new Promise(resolve => setTimeout(resolve, 1500))', { signal: controller.signal })));
+  tasks.push(t.assert.rejects(pool.run('new Promise(resolve => setTimeout(resolve, 1500))', { signal: controller2.signal })));
+  tasks.push(pool.run('new Promise(resolve => setTimeout(resolve, 1000))', { signal: controller3.signal }));
+
+
+  controller.abort();
+  controller2.abort();
+
+  await Promise.all(tasks);
+});