From 05cf50b3a39134c0cf2cea2068072c545f9806ba Mon Sep 17 00:00:00 2001
From: Nikhil-Doye <nikhil.doye@gmail.com>
Date: Mon, 27 Oct 2025 15:00:32 -0400
Subject: [PATCH] Add concurrency limiting functionality to execution engine

- Introduced ConcurrencyLimiter class to manage concurrent operations and prevent resource exhaustion.
- Implemented run, runAll, and runAllOrThrow methods for executing functions with concurrency control.
- Enhanced ExecutionEngine to utilize concurrency limiting when executing parallel groups and nodes, improving resource management and execution efficiency.
- Integrated mapWithConcurrency and mapWithConcurrencySettled functions for mapping operations with concurrency control.
---
 src/services/concurrencyLimiter.ts | 116 +++++++++++++++++++++++++++++
 src/services/executionEngine.ts    |  58 ++++++++++++---
 2 files changed, 164 insertions(+), 10 deletions(-)
 create mode 100644 src/services/concurrencyLimiter.ts

diff --git a/src/services/concurrencyLimiter.ts b/src/services/concurrencyLimiter.ts
new file mode 100644
index 0000000..a1f4cce
--- /dev/null
+++ b/src/services/concurrencyLimiter.ts
@@ -0,0 +1,116 @@
+/**
+ * Concurrency Limiter
+ * Controls the number of concurrent operations to prevent resource exhaustion
+ */
+
+export class ConcurrencyLimiter {
+  private maxConcurrency: number;
+  private running: number = 0;
+  private queue: Array<() => void> = [];
+
+  constructor(maxConcurrency: number = 5) {
+    this.maxConcurrency = Math.max(1, maxConcurrency);
+  }
+
+  /**
+   * Execute a function with concurrency limiting
+   */
+  async run<T>(fn: () => Promise<T>): Promise<T> {
+    // Wait until we have capacity
+    while (this.running >= this.maxConcurrency) {
+      await new Promise((resolve) => this.queue.push(resolve));
+    }
+
+    this.running++;
+
+    try {
+      return await fn();
+    } finally {
+      this.running--;
+
+      // Process next item in queue if any
+      const nextResolve = this.queue.shift();
+      if (nextResolve) {
+        nextResolve();
+      }
+    }
+  }
+
+  /**
+   * Execute multiple functions with concurrency limiting
+   * Returns results in same order as input functions
+   */
+  async runAll<T>(
+    fns: Array<() => Promise<T>>
+  ): Promise<PromiseSettledResult<T>[]> {
+    const promises = fns.map((fn) => this.run(fn));
+    return Promise.allSettled(promises);
+  }
+
+  /**
+   * Execute multiple functions with concurrency limiting
+   * Throws on first error
+   */
+  async runAllOrThrow<T>(fns: Array<() => Promise<T>>): Promise<T[]> {
+    const promises = fns.map((fn) => this.run(fn));
+    return Promise.all(promises);
+  }
+
+  /**
+   * Get current concurrency status
+   */
+  getStatus(): {
+    running: number;
+    queued: number;
+    maxConcurrency: number;
+  } {
+    return {
+      running: this.running,
+      queued: this.queue.length,
+      maxConcurrency: this.maxConcurrency,
+    };
+  }
+
+  /**
+   * Reset the limiter
+   */
+  reset(): void {
+    this.running = 0;
+    this.queue = [];
+  }
+}
+
+/**
+ * Create a limiter for a specific concurrency level
+ */
+export function createLimiter(maxConcurrency: number = 5): ConcurrencyLimiter {
+  return new ConcurrencyLimiter(maxConcurrency);
+}
+
+/**
+ * Map function with concurrency control
+ * Similar to Promise.all but with concurrency limits
+ */
+export async function mapWithConcurrency<T, R>(
+  items: T[],
+  fn: (item: T) => Promise<R>,
+  maxConcurrency: number = 5
+): Promise<R[]> {
+  const limiter = new ConcurrencyLimiter(maxConcurrency);
+  const promises = items.map((item) => limiter.run(() => fn(item)));
+  return Promise.all(promises);
+}
+
+/**
+ * Map function with concurrency control (settled version)
+ * Similar to Promise.allSettled but with concurrency limits
+ */
+export async function mapWithConcurrencySettled<T, R>(
+  items: T[],
+  fn: (item: T) => Promise<R>,
+  maxConcurrency: number = 5
+): Promise<PromiseSettledResult<R>[]> {
+  const limiter = new ConcurrencyLimiter(maxConcurrency);
+  const promises = items.map((item) => limiter.run(() => fn(item)));
+  return Promise.allSettled(promises);
+}
diff --git a/src/services/executionEngine.ts b/src/services/executionEngine.ts
index c5d1d6e..660a9cf 100644
--- a/src/services/executionEngine.ts
+++ b/src/services/executionEngine.ts
@@ -10,6 +10,10 @@ import {
   ExecutionErrorEvent,
 } from "./executionEventBus";
 import { workflowValidationEngine } from "./workflowValidationEngine";
+import {
+  mapWithConcurrencySettled,
+  ConcurrencyLimiter,
+} from "./concurrencyLimiter";
 
 export interface ExecutionContext {
   nodeId: string;
@@ -540,12 +544,21 @@ export class ExecutionEngine {
       "parallel"
     );
 
-    // Execute groups in parallel
-    const groupPromises = parallelGroups.map((group) =>
-      this.executeParallelGroup(group, plan, onNodeUpdate, executionId)
+    // Get the max concurrency from the plan or use default
+    const maxConcurrency =
+      (plan as any).maxConcurrency || this.maxConcurrentExecutions;
+
+    // Execute groups in parallel with concurrency limiting
+    const groupFunctions = parallelGroups.map(
+      (group) => () =>
+        this.executeParallelGroup(group, plan, onNodeUpdate, executionId)
     );
 
-    await Promise.allSettled(groupPromises);
+    await mapWithConcurrencySettled(
+      groupFunctions,
+      (fn) => fn(),
+      maxConcurrency
+    );
   }
 
   // Execute a parallel group
@@ -560,17 +573,42 @@ export class ExecutionEngine {
     ) => void,
     executionId: string
   ): Promise<void> {
-    const nodePromises = group.nodes.map((nodeId) => {
-      const context = plan.nodes.find((n) => n.nodeId === nodeId);
-      if (!context) return Promise.resolve();
+    const maxConcurrency =
+      group.maxConcurrency ||
+      (plan as any).maxConcurrency ||
+      this.maxConcurrentExecutions;
 
-      return this.executeNode(context, plan, onNodeUpdate, executionId);
+    const nodeExecutionFunctions = group.nodes.map((nodeId) => {
+      return async () => {
+        const context = plan.nodes.find((n) => n.nodeId === nodeId);
+        if (!context) return Promise.resolve();
+        return this.executeNode(context, plan, onNodeUpdate, executionId);
+      };
     });
 
     if (group.waitForAll) {
-      await Promise.allSettled(nodePromises);
+      // Execute all nodes with concurrency limiting
+      await mapWithConcurrencySettled(
+        nodeExecutionFunctions,
+        (fn) => fn(),
+        maxConcurrency
+      );
     } else {
-      await Promise.race(nodePromises);
+      // Execute with concurrency limiting and return first result
+      const promises = nodeExecutionFunctions.map((fn) =>
+        (async () => {
+          try {
+            return await fn();
+          } catch (error) {
+            return error;
+          }
+        })()
+      );
+
+      // Use Promise.race with limited concurrency
+      const limiter = new ConcurrencyLimiter(maxConcurrency);
+      const limitedPromises = promises.map((p) => limiter.run(() => p));
+      await Promise.race(limitedPromises);
     }
   }