Add stress test benchmarks for large concurrent step counts

.changeset/brave-walls-trade.md

@@ -0,0 +1,9 @@
+---
+"@workflow/world-local": patch
+---
+
+perf: optimize for high-concurrency workflows
+
+- Add in-memory cache for file existence checks to avoid expensive fs.access() calls
+- Increase default concurrency limit from 20 to 100
+- Improve HTTP connection pooling with undici Agent (100 connections, 30s keepalive)

packages/core/e2e/bench.bench.ts

@@ -293,4 +293,89 @@ describe('Workflow Performance Benchmarks', () => {
     },
     { time: 5000, warmupIterations: 1, teardown }
   );
+
+  // Stress tests for large concurrent step counts
+  // These reproduce reported issues with Promise.race/Promise.all at scale
+
+  bench(
+    'stress test: Promise.all with 100 concurrent steps',
+    async () => {
+      const { runId } = await triggerWorkflow(
+        'promiseAllStressTestWorkflow',
+        [100]
+      );
+      const { run } = await getWorkflowReturnValue(runId);
+      stageTiming('stress test: Promise.all with 100 concurrent steps', run);
+    },
+    { time: 30000, iterations: 1, warmupIterations: 0, teardown }
+  );
+
+  // TODO: Re-enable after performance optimizations (see beads issue wrk-fyx)
+  bench.skip(
+    'stress test: Promise.all with 500 concurrent steps',
+    async () => {
+      const { runId } = await triggerWorkflow(
+        'promiseAllStressTestWorkflow',
+        [500]
+      );
+      const { run } = await getWorkflowReturnValue(runId);
+      stageTiming('stress test: Promise.all with 500 concurrent steps', run);
+    },
+    { time: 60000, iterations: 1, warmupIterations: 0, teardown }
+  );
+
+  // TODO: Re-enable after performance optimizations (see beads issue wrk-fyx)
+  bench.skip(
+    'stress test: Promise.all with 1000 concurrent steps',
+    async () => {
+      const { runId } = await triggerWorkflow(
+        'promiseAllStressTestWorkflow',
+        [1000]
+      );
+      const { run } = await getWorkflowReturnValue(runId);
+      stageTiming('stress test: Promise.all with 1000 concurrent steps', run);
+    },
+    { time: 120000, iterations: 1, warmupIterations: 0, teardown }
+  );
+
+  bench(
+    'stress test: Promise.race with 100 concurrent steps',
+    async () => {
+      const { runId } = await triggerWorkflow(
+        'promiseRaceStressTestLargeWorkflow',
+        [100]
+      );
+      const { run } = await getWorkflowReturnValue(runId);
+      stageTiming('stress test: Promise.race with 100 concurrent steps', run);
+    },
+    { time: 30000, iterations: 1, warmupIterations: 0, teardown }
+  );
+
+  // TODO: Re-enable after performance optimizations (see beads issue wrk-fyx)
+  bench.skip(
+    'stress test: Promise.race with 500 concurrent steps',
+    async () => {
+      const { runId } = await triggerWorkflow(
+        'promiseRaceStressTestLargeWorkflow',
+        [500]
+      );
+      const { run } = await getWorkflowReturnValue(runId);
+      stageTiming('stress test: Promise.race with 500 concurrent steps', run);
+    },
+    { time: 60000, iterations: 1, warmupIterations: 0, teardown }
+  );
+
+  // TODO: Re-enable after performance optimizations (see beads issue wrk-fyx)
+  bench.skip(
+    'stress test: Promise.race with 1000 concurrent steps',
+    async () => {
+      const { runId } = await triggerWorkflow(
+        'promiseRaceStressTestLargeWorkflow',
+        [1000]
+      );
+      const { run } = await getWorkflowReturnValue(runId);
+      stageTiming('stress test: Promise.race with 1000 concurrent steps', run);
+    },
+    { time: 120000, iterations: 1, warmupIterations: 0, teardown }
+  );
 });

packages/world-local/src/fs.ts

@@ -9,6 +9,17 @@ const ulid = monotonicFactory(() => Math.random());
 
 const Ulid = z.string().ulid();
 
+// In-memory cache of created files to avoid expensive fs.access() calls
+// This is safe because we only write once per file path (no overwrites without explicit flag)
+const createdFilesCache = new Set<string>();
+
+/**
+ * Clear the created files cache. Useful for testing or when files are deleted externally.
+ */
+export function clearCreatedFilesCache(): void {
+  createdFilesCache.clear();
+}
+
 export function ulidToDate(maybeUlid: string): Date | null {
   const ulid = Ulid.safeParse(maybeUlid);
   if (!ulid.success) {
@@ -53,8 +64,20 @@ export async function write(
   opts?: WriteOptions
 ): Promise<void> {
   if (!opts?.overwrite) {
+    // Fast path: check in-memory cache first to avoid expensive fs.access() calls
+    // This provides significant performance improvement when creating many files
+    if (createdFilesCache.has(filePath)) {
+      throw new WorkflowAPIError(
+        `File ${filePath} already exists and 'overwrite' is false`,
+        { status: 409 }
+      );
+    }
+
+    // Slow path: check filesystem for files created before this process started
     try {
       await fs.access(filePath);
+      // File exists on disk, add to cache for future checks
+      createdFilesCache.add(filePath);
       throw new WorkflowAPIError(
         `File ${filePath} already exists and 'overwrite' is false`,
         { status: 409 }
@@ -74,6 +97,8 @@ export async function write(
     await fs.writeFile(tempPath, data);
     tempFileCreated = true;
     await fs.rename(tempPath, filePath);
+    // Track this file in cache so future writes know it exists
+    createdFilesCache.add(filePath);
   } catch (error) {
     // Only try to clean up temp file if it was actually created
     if (tempFileCreated) {

packages/world-local/src/queue.ts

@@ -16,14 +16,20 @@ const LOCAL_QUEUE_MAX_VISIBILITY =
 
 // The local workers share the same Node.js process and event loop,
 // so we need to limit concurrency to avoid overwhelming the system.
-const DEFAULT_CONCURRENCY_LIMIT = 20;
+const DEFAULT_CONCURRENCY_LIMIT = 100;
 const WORKFLOW_LOCAL_QUEUE_CONCURRENCY =
   parseInt(process.env.WORKFLOW_LOCAL_QUEUE_CONCURRENCY ?? '0', 10) ||
   DEFAULT_CONCURRENCY_LIMIT;
 
-// Create a custom agent with unlimited headers timeout for long-running steps
+// Create a custom agent optimized for high-concurrency local workflows:
+// - headersTimeout: 0 allows long-running steps
+// - connections: 100 allows many parallel connections to the same host
+// - pipelining: 1 (default) for HTTP/1.1 compatibility
+// - keepAliveTimeout: 30s keeps connections warm for rapid step execution
 const httpAgent = new Agent({
   headersTimeout: 0,
+  connections: 100,
+  keepAliveTimeout: 30_000,
 });
 
 export function createQueue(config: Partial<Config>): Queue {

workbench/example/workflows/97_bench.ts

@@ -86,3 +86,42 @@ export async function streamWorkflow() {
   const doubled = await doubleNumbers(stream);
   return doubled;
 }
+
+//////////////////////////////////////////////////////////
+// Stress test workflows for large concurrent step counts
+//////////////////////////////////////////////////////////
+
+async function stressTestStep(i: number) {
+  'use step';
+  // Minimal work to isolate the overhead of concurrent step tracking
+  return i;
+}
+
+// Stress test: Promise.all with many concurrent steps
+export async function promiseAllStressTestWorkflow(count: number) {
+  'use workflow';
+  const promises: Promise<number>[] = [];
+  for (let i = 0; i < count; i++) {
+    promises.push(stressTestStep(i));
+  }
+  const results = await Promise.all(promises);
+  return results.length;
+}
+
+// Stress test: Promise.race with many concurrent steps (uses Map pattern from report)
+export async function promiseRaceStressTestLargeWorkflow(count: number) {
+  'use workflow';
+  const runningTasks = new Map<number, Promise<number>>();
+  for (let i = 0; i < count; i++) {
+    runningTasks.set(i, stressTestStep(i));
+  }
+
+  const done: number[] = [];
+  while (runningTasks.size > 0) {
+    const result = await Promise.race(runningTasks.values());
+    done.push(result);
+    runningTasks.delete(result);
+  }
+
+  return done.length;
+}