fix(concurrency-parallel): pull-based dispatch, crash recovery, timeout semantics

src/api/concurrency-parallel/README.md

@@ -24,6 +24,16 @@ npm run concurrency-parallel:test
 limited concurrency for rate limits and connection pools; priority queue when
 tasks have different importance.
 
+## Reliability & scheduling
+
+- **Worker pool is pull-based**: tasks wait in one shared queue; the next idle
+  worker takes the next task (no head-of-line blocking behind a slow task).
+  Optional `maxPendingTasks` applies backpressure; crashed workers fail their
+  in-flight task loudly and are auto-respawned (`maxWorkerRestarts`).
+- **ConcurrencyManager honors `timeout` and `retries`** from its config, plus
+  optional `signal` (AbortSignal) for cooperative cancellation and
+  `executeAllSettled()` for per-task outcomes without fail-fast.
+
 ## Endpoints
 
 - **Concurrency:** `POST /concurrent/all`, `/concurrent/limited`,

src/api/concurrency-parallel/benchmarks/bench.ts

@@ -0,0 +1,46 @@
+/**
+ * Benchmarks: worker-pool dispatch under skewed task durations + fibonacci.
+ * Run against a built dist (node benchmarks/bench.js). Compare by building
+ * the old and new sources into dist and running this same script.
+ */
+/* eslint-disable no-console */
+import { ParallelManager } from "../src/parallel-manager";
+
+async function benchDispatch() {
+  const manager = new ParallelManager({ workerCount: 2, timeout: 600000 });
+  await manager.initializeWorkers();
+
+  // 16 tasks with alternating heavy/light durations — the adversarial (and
+  // realistic) case for eager round-robin: all heavy tasks land on one worker.
+  const tasks = [];
+  for (let i = 0; i < 16; i++) {
+    tasks.push({ iterations: i % 2 === 0 ? 15_000_000 : 50_000 });
+  }
+
+  const start = Date.now();
+  await manager.executeParallel(tasks, "compute");
+  const elapsed = Date.now() - start;
+  await manager.cleanup();
+  return elapsed;
+}
+
+async function benchFibonacci(n) {
+  const manager = new ParallelManager({ workerCount: 1, timeout: 600000 });
+  await manager.initializeWorkers();
+  const start = Date.now();
+  await manager.executeParallel([{ n }], "fibonacci");
+  const elapsed = Date.now() - start;
+  await manager.cleanup();
+  return elapsed;
+}
+
+(async () => {
+  const dispatchMs = await benchDispatch();
+  console.log(`RESULT dispatch_skewed_16tasks_2workers_ms=${dispatchMs}`);
+  const fibMs = await benchFibonacci(42);
+  console.log(`RESULT fibonacci_n42_ms=${fibMs}`);
+  process.exit(0);
+})().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});

src/api/concurrency-parallel/src/concurrency-manager.ts

@@ -5,16 +5,19 @@
  * 1. Promise-based concurrency control
  * 2. Queue management with priority
  * 3. Rate limiting and throttling
- * 4. Error handling and retries
- * 5. Performance monitoring
+ * 4. Error handling, retries, and per-task timeouts
+ * 5. Cooperative cancellation (AbortSignal) and fail-fast semantics
+ * 6. Performance monitoring
  */
 
 import {
   TaskResult,
   ConcurrencyConfig,
   PerformanceMetrics,
   TaskProcessor,
+  SettledTaskResult,
 } from "./concurrency-types.js";
+import { ConcurrencyDefaultConfig } from "./constants";
 
 export class ConcurrencyManager {
   private config: ConcurrencyConfig;
@@ -26,6 +29,68 @@ export class ConcurrencyManager {
     this.config = config;
   }
 
+  /**
+   * Run a processor for one task with the configured timeout + retries applied.
+   * - `timeout` (ms): each attempt races a timer; a late attempt rejects with a
+   *   descriptive error instead of hanging the whole batch.
+   * - `retries`: failed attempts (including timeouts) are retried up to N extra
+   *   times before the error propagates.
+   * Both settings were always declared on ConcurrencyConfig but previously ignored.
+   */
+  private async runWithPolicy<T, R>(task: T, processor: TaskProcessor<T, R>): Promise<R> {
+    const { timeout, retries = 0 } = this.config;
+    const attempts = Math.max(0, retries) + 1;
+    let lastError: unknown;
+
+    for (let attempt = 1; attempt <= attempts; attempt++) {
+      try {
+        if (timeout === undefined || timeout <= 0) {
+          return await processor(task);
+        }
+        return await this.withTimeout(processor(task), timeout, attempt);
+      } catch (error) {
+        lastError = error;
+        if (attempt === attempts) break;
+      }
+    }
+    throw lastError;
+  }
+
+  private withTimeout<R>(promise: Promise<R>, timeoutMs: number, attempt: number): Promise<R> {
+    return new Promise<R>((resolve, reject) => {
+      const timer = setTimeout(() => {
+        reject(new Error(`Task timed out after ${timeoutMs}ms (attempt ${attempt})`));
+      }, timeoutMs);
+      promise.then(
+        (value) => {
+          clearTimeout(timer);
+          resolve(value);
+        },
+        (error) => {
+          clearTimeout(timer);
+          reject(error);
+        }
+      );
+    });
+  }
+
+  /** Throw an AbortError-shaped error if the configured signal has fired. */
+  private throwIfAborted(): void {
+    if (this.config.signal?.aborted) {
+      const reason = this.config.signal.reason;
+      throw reason instanceof Error ? reason : new Error("Execution aborted");
+    }
+  }
+
+  /** Record a completed task, keeping memory bounded (FIFO eviction). */
+  private recordTask(taskResult: TaskResult): void {
+    const cap = this.config.maxCompletedTasks ?? ConcurrencyDefaultConfig.MAX_COMPLETED_TASKS;
+    if (this.completedTasks.length >= cap) {
+      this.completedTasks.shift();
+    }
+    this.completedTasks.push(taskResult);
+  }
+
   /**
    * Basic Promise.all approach - All tasks start simultaneously
    * Good for: Independent tasks that can all run at once
@@ -36,6 +101,7 @@ export class ConcurrencyManager {
       console.warn(`🚀 Starting ${tasks.length} tasks concurrently (Promise.all)`);
     }
     this.startTime = Date.now();
+    this.throwIfAborted();
 
     try {
       // All promises start immediately - true concurrency
@@ -44,10 +110,10 @@ export class ConcurrencyManager {
         this.activeTasksCount++;
 
         try {
-          const result = await processor(task);
+          const result = await this.runWithPolicy(task, processor);
           const endTime = Date.now();
 
-          this.completedTasks.push({
+          this.recordTask({
             taskId: `task-${index}`,
             result,
             executionTime: endTime - startTime,
@@ -73,7 +139,46 @@ export class ConcurrencyManager {
   }
 
   /**
-   * Limited concurrency with a lightweight inline limiter
+   * Like executeAllConcurrent, but never fail-fast: every task runs to
+   * completion and the caller gets a per-task settled outcome. Mirrors
+   * Promise.allSettled semantics with the manager's timeout/retry policy.
+   */
+  async executeAllSettled<T, R>(
+    tasks: T[],
+    processor: TaskProcessor<T, R>
+  ): Promise<SettledTaskResult<R>[]> {
+    this.startTime = Date.now();
+    this.throwIfAborted();
+
+    return Promise.all(
+      tasks.map(async (task, index): Promise<SettledTaskResult<R>> => {
+        const startTime = Date.now();
+        this.activeTasksCount++;
+        try {
+          const result = await this.runWithPolicy(task, processor);
+          const endTime = Date.now();
+          this.recordTask({
+            taskId: `settled-task-${index}`,
+            result,
+            executionTime: endTime - startTime,
+            startTime,
+            endTime,
+          });
+          return { status: "fulfilled", value: result, taskIndex: index };
+        } catch (error) {
+          return { status: "rejected", reason: error, taskIndex: index };
+        } finally {
+          this.activeTasksCount--;
+        }
+      })
+    );
+  }
+
+  /**
+   * Limited concurrency with a lightweight inline limiter.
+   * Fail-fast is now genuine: when one task rejects (after retries), the other
+   * worker lanes stop pulling new tasks instead of silently continuing to run
+   * side effects behind an already-rejected promise.
    */
   async executeLimitedConcurrent<T, R>(tasks: T[], processor: TaskProcessor<T, R>): Promise<R[]> {
     if (typeof process.env.CI === "undefined") {
@@ -82,29 +187,36 @@ export class ConcurrencyManager {
       );
     }
     this.startTime = Date.now();
+    this.throwIfAborted();
 
     const concurrency = Math.max(1, this.config.maxConcurrent);
     const results: R[] = new Array(tasks.length);
     let nextIndex = 0;
+    let failed = false;
 
     const runWorker = async (): Promise<void> => {
       for (;;) {
+        if (failed) break;
+        if (this.config.signal?.aborted) this.throwIfAborted();
         const current = nextIndex++;
         if (current >= tasks.length) break;
 
         const startTime = Date.now();
         this.activeTasksCount++;
         try {
-          const result = await processor(tasks[current]);
+          const result = await this.runWithPolicy(tasks[current], processor);
           results[current] = result as R;
           const endTime = Date.now();
-          this.completedTasks.push({
+          this.recordTask({
             taskId: `limited-task-${current}`,
             result,
             executionTime: endTime - startTime,
             startTime,
             endTime,
           });
+        } catch (error) {
+          failed = true;
+          throw error;
         } finally {
           this.activeTasksCount--;
         }
@@ -121,7 +233,8 @@ export class ConcurrencyManager {
   }
 
   /**
-   * Priority queue without external deps
+   * Priority queue without external deps.
+   * Same fail-fast + abort semantics as executeLimitedConcurrent.
    */
   async executePriorityQueue<T, R>(
     tasks: (T & { priority?: number })[],
@@ -130,31 +243,38 @@ export class ConcurrencyManager {
     if (typeof process.env.CI === "undefined") {
       console.warn(`🏆 Starting priority queue with ${tasks.length} tasks`);
     }
+    this.throwIfAborted();
 
     // Sort by priority descending; process with limited concurrency
     const sorted = [...tasks].sort((a, b) => (b.priority || 0) - (a.priority || 0));
     const results: R[] = new Array(sorted.length);
     let nextIndex = 0;
+    let failed = false;
     const concurrency = Math.max(1, this.config.maxConcurrent);
 
     const runWorker = async (): Promise<void> => {
       for (;;) {
+        if (failed) break;
+        if (this.config.signal?.aborted) this.throwIfAborted();
         const idx = nextIndex++;
         if (idx >= sorted.length) break;
         const task = sorted[idx];
         const startTime = Date.now();
         this.activeTasksCount++;
         try {
-          const result = await processor(task);
+          const result = await this.runWithPolicy(task, processor);
           results[idx] = result as R;
           const endTime = Date.now();
-          this.completedTasks.push({
+          this.recordTask({
             taskId: `priority-task-${idx}`,
             result,
             executionTime: endTime - startTime,
             startTime,
             endTime,
           });
+        } catch (error) {
+          failed = true;
+          throw error;
         } finally {
           this.activeTasksCount--;
         }
@@ -185,6 +305,7 @@ export class ConcurrencyManager {
     const results: R[] = [];
 
     for (let i = 0; i < tasks.length; i++) {
+      this.throwIfAborted();
       const task = tasks[i];
       const startTime = Date.now();
 
@@ -193,10 +314,10 @@ export class ConcurrencyManager {
       }
 
       try {
-        const result = await processor(task);
+        const result = await this.runWithPolicy(task, processor);
         const endTime = Date.now();
 
-        this.completedTasks.push({
+        this.recordTask({
           taskId: `sequential-task-${i}`,
           result,
           executionTime: endTime - startTime,
@@ -243,6 +364,7 @@ export class ConcurrencyManager {
 
     // Split tasks into batches
     for (let i = 0; i < tasks.length; i += batchSize) {
+      this.throwIfAborted();
       const batch = tasks.slice(i, i + batchSize);
       const batchNumber = Math.floor(i / batchSize) + 1;
       const totalBatches = Math.ceil(tasks.length / batchSize);
@@ -266,11 +388,18 @@ export class ConcurrencyManager {
     return results;
   }
 
+  /**
+   * Current number of in-flight tasks (useful for monitoring/backpressure).
+   */
+  getActiveTaskCount(): number {
+    return this.activeTasksCount;
+  }
+
   /**
    * Get performance metrics for analysis
    */
   getPerformanceMetrics(): PerformanceMetrics {
-    const totalExecutionTime = Date.now() - this.startTime;
+    const totalExecutionTime = this.startTime > 0 ? Date.now() - this.startTime : 0;
     const completedTasksCount = this.completedTasks.length;
 
     return {
@@ -282,7 +411,10 @@ export class ConcurrencyManager {
             completedTasksCount
           : 0,
       concurrencyLevel: this.config.maxConcurrent,
-      throughput: completedTasksCount > 0 ? (completedTasksCount / totalExecutionTime) * 1000 : 0,
+      throughput:
+        completedTasksCount > 0 && totalExecutionTime > 0
+          ? (completedTasksCount / totalExecutionTime) * 1000
+          : 0,
     };
   }
 

src/api/concurrency-parallel/src/concurrency-parallel.test.ts

@@ -453,10 +453,11 @@ describe("Integration Tests", () => {
         return { taskId: task.id };
       };
 
-      // Note: The current implementation doesn't have built-in timeout handling
-      // This test documents the expected behavior for future implementation
-      const results = await manager.executeAllConcurrent(slowTasks, slowProcessor);
-      expect(results).toHaveLength(1);
+      // ConcurrencyConfig.timeout is now enforced: a task exceeding it rejects
+      // instead of hanging the batch (this test previously documented the gap).
+      await expect(manager.executeAllConcurrent(slowTasks, slowProcessor)).rejects.toThrow(
+        /timed out after 500ms/
+      );
     });
   });
 });