perf: use Map for invocationsQueue (O(1) lookup/delete)

Replace array-based invocationsQueue with Map<string, QueueItem> for
O(1) lookup and delete operations. This eliminates the O(n²) complexity
from findIndex + splice patterns in step.ts, hook.ts, and sleep.ts.

Changes:
- private.ts: Change invocationsQueue type from QueueItem[] to Map
- workflow.ts: Initialize with new Map() instead of []
- global.ts: Accept Map or array in WorkflowSuspension, add single-pass counting
- step.ts: Use Map.set/has/delete instead of push/findIndex/splice
- hook.ts: Use Map.set/delete instead of push/findIndex/splice
- sleep.ts: Use Map.set/get/delete instead of push/find/findIndex/splice

This is Phase 1 of the high-concurrency workflow optimization plan.
See beads issue wrk-fyx for details.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: pranaygp

.changeset/slick-cooks-clean.md

@@ -0,0 +1,7 @@
+---
+"@workflow/core": patch
+---
+
+perf: use Map for invocationsQueue (O(1) lookup/delete)
+
+Replace array-based invocationsQueue with Map for O(1) lookup and delete operations, eliminating O(n²) complexity in high-concurrency workflows.

packages/core/src/global.ts

@@ -40,10 +40,23 @@ export class WorkflowSuspension extends Error {
   hookCount: number;
   waitCount: number;
 
-  constructor(steps: QueueItem[], global: typeof globalThis) {
-    const stepCount = steps.filter((s) => s.type === 'step').length;
-    const hookCount = steps.filter((s) => s.type === 'hook').length;
-    const waitCount = steps.filter((s) => s.type === 'wait').length;
+  constructor(
+    stepsInput: QueueItem[] | Map<string, QueueItem>,
+    global: typeof globalThis
+  ) {
+    // Convert Map to array for backward compatibility
+    const steps =
+      stepsInput instanceof Map ? [...stepsInput.values()] : stepsInput;
+
+    // Single-pass counting for efficiency
+    let stepCount = 0;
+    let hookCount = 0;
+    let waitCount = 0;
+    for (const item of steps) {
+      if (item.type === 'step') stepCount++;
+      else if (item.type === 'hook') hookCount++;
+      else if (item.type === 'wait') waitCount++;
+    }
 
     // Build description parts
     const parts: string[] = [];

packages/core/src/private.ts

@@ -38,7 +38,11 @@ export { __private_getClosureVars } from './step/get-closure-vars.js';
 export interface WorkflowOrchestratorContext {
   globalThis: typeof globalThis;
   eventsConsumer: EventsConsumer;
-  invocationsQueue: QueueItem[];
+  /**
+   * Map of pending invocations keyed by correlationId.
+   * Using Map instead of Array for O(1) lookup/delete operations.
+   */
+  invocationsQueue: Map<string, QueueItem>;
   onWorkflowError: (error: Error) => void;
   generateUlid: () => string;
   generateNanoid: () => string;

packages/core/src/step.test.ts

@@ -20,7 +20,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
   return {
     globalThis: context.globalThis,
     eventsConsumer: new EventsConsumer(events),
-    invocationsQueue: [],
+    invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt), // All generated ulids use the workflow's started at time
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
@@ -99,7 +99,10 @@ describe('createUseStep', () => {
     expect((error as WorkflowSuspension).message).toBe(
       '1 step has not been run yet'
     );
-    expect(ctx.invocationsQueue).toEqual((error as WorkflowSuspension).steps);
+    // Compare Map values with WorkflowSuspension.steps array
+    expect([...ctx.invocationsQueue.values()]).toEqual(
+      (error as WorkflowSuspension).steps
+    );
     expect((error as WorkflowSuspension).steps).toMatchInlineSnapshot(`
       [
         {
@@ -142,7 +145,10 @@ describe('createUseStep', () => {
     expect((error as WorkflowSuspension).message).toBe(
       '3 steps have not been run yet'
     );
-    expect(ctx.invocationsQueue).toEqual((error as WorkflowSuspension).steps);
+    // Compare Map values with WorkflowSuspension.steps array
+    expect([...ctx.invocationsQueue.values()]).toEqual(
+      (error as WorkflowSuspension).steps
+    );
     expect((error as WorkflowSuspension).steps).toMatchInlineSnapshot(`
       [
         {
@@ -228,8 +234,9 @@ describe('createUseStep', () => {
     expect(result).toBe('Result: 42');
 
     // Verify closure variables were added to invocation queue
-    expect(ctx.invocationsQueue).toHaveLength(1);
-    expect(ctx.invocationsQueue[0]).toMatchObject({
+    expect(ctx.invocationsQueue.size).toBe(1);
+    const queueItem = [...ctx.invocationsQueue.values()][0];
+    expect(queueItem).toMatchObject({
       type: 'step',
       stepName: 'calculate',
       args: [],
@@ -263,8 +270,9 @@ describe('createUseStep', () => {
     expect(result).toBe(5);
 
     // Verify empty closure variables were added to invocation queue
-    expect(ctx.invocationsQueue).toHaveLength(1);
-    expect(ctx.invocationsQueue[0]).toMatchObject({
+    expect(ctx.invocationsQueue.size).toBe(1);
+    const queueItem = [...ctx.invocationsQueue.values()][0];
+    expect(queueItem).toMatchObject({
       type: 'step',
       stepName: 'add',
       args: [2, 3],

packages/core/src/step.ts

@@ -30,7 +30,7 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) {
         queueItem.closureVars = closureVars;
       }
 
-      ctx.invocationsQueue.push(queueItem);
+      ctx.invocationsQueue.set(correlationId, queueItem);
 
       // Track whether we've already seen a "step_started" event for this step.
       // This is important because after a retryable failure, the step moves back to
@@ -73,13 +73,9 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) {
         if (event.eventType === 'step_started') {
           // Step has started - so remove from the invocations queue (only on the first "step_started" event)
           if (!hasSeenStepStarted) {
-            const invocationsQueueIndex = ctx.invocationsQueue.findIndex(
-              (invocation) =>
-                invocation.type === 'step' &&
-                invocation.correlationId === correlationId
-            );
-            if (invocationsQueueIndex !== -1) {
-              ctx.invocationsQueue.splice(invocationsQueueIndex, 1);
+            // O(1) lookup and delete using Map
+            if (ctx.invocationsQueue.has(correlationId)) {
+              ctx.invocationsQueue.delete(correlationId);
             } else {
               setTimeout(() => {
                 reject(

packages/core/src/workflow.ts

@@ -76,7 +76,7 @@ export async function runWorkflow(
       eventsConsumer: new EventsConsumer(events),
       generateUlid: () => ulid(+startedAt),
       generateNanoid,
-      invocationsQueue: [],
+      invocationsQueue: new Map(),
     };
 
     // Subscribe to the events log to update the timestamp in the vm context

packages/core/src/workflow/hook.ts

@@ -13,8 +13,8 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) {
     const correlationId = `hook_${ctx.generateUlid()}`;
     const token = options.token ?? ctx.generateNanoid();
 
-    // Add hook creation to invocations queue
-    ctx.invocationsQueue.push({
+    // Add hook creation to invocations queue (using Map for O(1) operations)
+    ctx.invocationsQueue.set(correlationId, {
       type: 'hook',
       correlationId,
       token,
@@ -52,13 +52,8 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) {
         event?.eventType === 'hook_created' &&
         event.correlationId === correlationId
       ) {
-        // Remove this hook from the invocations queue if it exists
-        const index = ctx.invocationsQueue.findIndex(
-          (item) => item.type === 'hook' && item.correlationId === correlationId
-        );
-        if (index !== -1) {
-          ctx.invocationsQueue.splice(index, 1);
-        }
+        // Remove this hook from the invocations queue (O(1) delete using Map)
+        ctx.invocationsQueue.delete(correlationId);
         return EventConsumerResult.Consumed;
       }
 

packages/core/src/workflow/sleep.ts

@@ -14,12 +14,13 @@ export function createSleep(ctx: WorkflowOrchestratorContext) {
     // Calculate the resume time
     const resumeAt = parseDurationToDate(param);
 
-    // Add wait to invocations queue
-    ctx.invocationsQueue.push({
+    // Add wait to invocations queue (using Map for O(1) operations)
+    const waitItem: WaitInvocationQueueItem = {
       type: 'wait',
       correlationId,
       resumeAt,
-    });
+    };
+    ctx.invocationsQueue.set(correlationId, waitItem);
 
     ctx.eventsConsumer.subscribe((event) => {
       // If there are no events and we're waiting for wait_completed,
@@ -39,12 +40,11 @@ export function createSleep(ctx: WorkflowOrchestratorContext) {
         event.correlationId === correlationId
       ) {
         // Mark this wait as having the created event, but keep it in the queue
-        const waitItem = ctx.invocationsQueue.find(
-          (item) => item.type === 'wait' && item.correlationId === correlationId
-        ) as WaitInvocationQueueItem | undefined;
-        if (waitItem) {
-          waitItem.hasCreatedEvent = true;
-          waitItem.resumeAt = event.eventData.resumeAt;
+        // O(1) lookup using Map
+        const queueItem = ctx.invocationsQueue.get(correlationId);
+        if (queueItem && queueItem.type === 'wait') {
+          queueItem.hasCreatedEvent = true;
+          queueItem.resumeAt = event.eventData.resumeAt;
         }
         return EventConsumerResult.Consumed;
       }
@@ -54,13 +54,8 @@ export function createSleep(ctx: WorkflowOrchestratorContext) {
         event?.eventType === 'wait_completed' &&
         event.correlationId === correlationId
       ) {
-        // Remove this wait from the invocations queue
-        const index = ctx.invocationsQueue.findIndex(
-          (item) => item.type === 'wait' && item.correlationId === correlationId
-        );
-        if (index !== -1) {
-          ctx.invocationsQueue.splice(index, 1);
-        }
+        // Remove this wait from the invocations queue (O(1) delete using Map)
+        ctx.invocationsQueue.delete(correlationId);
 
         // Wait has elapsed, resolve the sleep
         setTimeout(() => {