perf: optimize Flow execution and compact checkpoints

.gitignore

@@ -3,6 +3,7 @@ dist/
 coverage/
 *.log
 .DS_Store
+/.claude/
 .open-flow-dev/
 .wrangler/
 

.vscode/settings.json

@@ -1,4 +1,6 @@
 {
+  "js/ts.experimental.useTsgo": true,
+  "js/ts.tsdk.path": "./node_modules/tsrs/tsdk",
   "[typescript]": {
     "editor.defaultFormatter": "oxc.oxc-vscode",
     "editor.formatOnSave": true,

apps/server/node/application/execution-slots.ts

@@ -0,0 +1,106 @@
+import * as Effect from 'effect/Effect'
+import * as Semaphore from 'effect/Semaphore'
+
+interface Lease {
+  readonly runId: string
+  readonly abort: AbortController
+  held: boolean
+  /** Outstanding child waits; the lease holds no permit while any remain. */
+  suspended: number
+  taking: Promise<void> | undefined
+  ended: boolean
+}
+
+/** Deployment-wide Task budget, held per node invocation and lent out while it waits for children. */
+export class ExecutionSlots {
+  readonly #semaphore: Semaphore.Semaphore
+  readonly #leases = new Map<string, Lease>()
+
+  constructor(limit: number) {
+    this.#semaphore = Semaphore.makeUnsafe(limit)
+  }
+
+  /** Admits one node invocation; repeated admission of a live invocation shares its lease. */
+  acquire(runId: string, jobId: string, signal: AbortSignal): Promise<void> {
+    signal.throwIfAborted()
+    const key = JSON.stringify([runId, jobId])
+    let lease = this.#leases.get(key)
+    if (lease == null) {
+      lease = { runId, abort: new AbortController(), held: false, suspended: 0, taking: undefined, ended: false }
+      this.#leases.set(key, lease)
+    }
+    return this.#hold(lease, signal)
+  }
+
+  /** Lends the permit to child work until every overlapping child wait has resumed. */
+  suspend(runId: string, jobId: string): void {
+    const lease = this.#leases.get(JSON.stringify([runId, jobId]))
+    if (lease == null) return
+    lease.suspended++
+    if (lease.suspended == 1) this.#drop(lease)
+  }
+
+  /** Ends one child wait; the last one reacquires the permit before the invocation continues. */
+  resume(runId: string, jobId: string, signal: AbortSignal): Promise<void> {
+    const key = JSON.stringify([runId, jobId])
+    const lease = this.#leases.get(key)
+    if (lease == null || lease.suspended == 0) return Promise.resolve()
+    lease.suspended--
+    return lease.suspended > 0 ? Promise.resolve() : this.#hold(lease, signal)
+  }
+
+  release(runId: string, jobId: string): void {
+    const key = JSON.stringify([runId, jobId])
+    const lease = this.#leases.get(key)
+    if (lease != null) this.#end(key, lease, 'Node execution ended.')
+  }
+
+  releaseRun(runId: string): void {
+    for (const [key, lease] of this.#leases) if (lease.runId == runId) this.#end(key, lease, 'Run execution ended.')
+  }
+
+  #hold(lease: Lease, signal: AbortSignal): Promise<void> {
+    signal.throwIfAborted()
+    if (lease.held || lease.suspended > 0) return Promise.resolve()
+    if (lease.taking != null) return lease.taking
+    const taking = Effect.runPromise(
+      Effect.uninterruptibleMask((restore) =>
+        restore(this.#semaphore.take(1)).pipe(
+          Effect.andThen(
+            Effect.suspend(() => {
+              // A permit that arrives after the lease ended or lent itself out is returned immediately.
+              if (lease.ended || lease.suspended > 0) return Effect.asVoid(this.#semaphore.release(1))
+              lease.held = true
+              return Effect.void
+            }),
+          ),
+        ),
+      ),
+      { signal: AbortSignal.any([signal, lease.abort.signal]) },
+    ).then(
+      () => {
+        if (lease.taking == taking) lease.taking = undefined
+      },
+      (error: unknown) => {
+        // The lease itself ends only through release or releaseRun.
+        if (lease.taking == taking) lease.taking = undefined
+        throw error
+      },
+    )
+    lease.taking = taking
+    return taking
+  }
+
+  #drop(lease: Lease): void {
+    if (!lease.held) return
+    lease.held = false
+    Effect.runSync(this.#semaphore.release(1))
+  }
+
+  #end(key: string, lease: Lease, reason: string): void {
+    this.#leases.delete(key)
+    lease.ended = true
+    lease.abort.abort(new Error(reason))
+    this.#drop(lease)
+  }
+}

apps/server/node/application/run.ts

@@ -30,8 +30,9 @@ import { HttpTaskError } from '../deployment/http-transport.ts'
 import { executeOpenApi } from '../deployment/openapi.ts'
 import { WorkflowToolError } from '../deployment/workflow-contract.ts'
 import { errorKind } from '../logger.ts'
-import { supportsIsolatedVmEngine, IsolatedVmHost } from '../runtime/isolated-vm.ts'
+import { supportsIsolatedVmEngine, IsolatedVmError, IsolatedVmHost } from '../runtime/isolated-vm.ts'
 import { checkCodeActions, checkCodePermissions, checkConnectorProxies } from './connector-capabilities.ts'
+import { ExecutionSlots } from './execution-slots.ts'
 import { decodedRevision, preparedRevision } from './revision-cache.ts'
 import { ArtifactHostError, RunArtifacts } from './run-artifacts.ts'
 import { RunState } from './run-state.ts'
@@ -114,6 +115,7 @@ export class RunExecutor {
   readonly #toolCalls = new Map<string, { runId: string; jobId: string; abort: AbortController }>()
   readonly #state: RunState
   readonly #artifacts: RunArtifacts
+  readonly #slots: ExecutionSlots
 
   constructor(
     store: Store,
@@ -129,7 +131,9 @@ export class RunExecutor {
     invokeWorkflowTarget: RunControl['invokeWorkflowTarget'],
     subscribeRun: (flowId: string, runId: string, changed: () => void) => () => void,
     additionalFailureCodes: ReadonlySet<string>,
+    maxConcurrentTasks: number,
   ) {
+    this.#slots = new ExecutionSlots(maxConcurrentTasks)
     this.#invokeWorkflowTarget = invokeWorkflowTarget
     this.#subscribeRun = subscribeRun
     this.#invokeWorkflow = invokeWorkflow
@@ -313,6 +317,7 @@ export class RunExecutor {
       const emit = async (event: Parameters<NonNullable<Parameters<IsolatedVmHost['run']>[1]['emit']>>[0] | ArtifactEvent): Promise<void> => {
         if (event.type == 'run.started' && event.runId == run.runId) return
         if ((event.type == 'node.completed' || event.type == 'node.failed') && 'jobId' in event) {
+          this.#slots.release(run.runId, event.jobId)
           connectors.release(event.jobId)
           if (event.type === 'node.failed') this.#store.state.discardInvocation(run.runId, event.jobId)
           this.#cancelNodeTools(run.runId, event.jobId)
@@ -325,6 +330,7 @@ export class RunExecutor {
         .run(flow, {
           capability: (capabilities, call) => this.#invokeCapability(capabilities, call, this.#connectorContext(run, 'execute'), run, flow, emit),
           emit,
+          admitTask: (admission, signal) => this.#slots.acquire(admission.runId, admission.jobId, signal),
           emitLogs: async (events, signal) => {
             const projected = (await Promise.all(events.map((event) => projectEvent(event).catch(() => undefined)))).filter((event) => event != null)
             if (signal.aborted) return
@@ -362,6 +368,7 @@ export class RunExecutor {
               }
             }
             const output = this.#store.runs.createWait(run.runId, request, this.#resolveWaitPublicOrigin()?.href, notification)
+            this.#slots.release(run.runId, request.jobId)
             this.#runChanged(run.flowId, run.runId)
             this.#wakeMaintenance()
             return output
@@ -405,13 +412,27 @@ export class RunExecutor {
         })
         .pipe(
           Effect.matchEffect({
-            onFailure: (error) => Effect.sync(() => this.#failRun(run, startedAt, error.message.includes('execution deadline'), error, indeterminate)),
+            onFailure: (error) =>
+              Effect.sync(() =>
+                this.#failRun(
+                  run,
+                  startedAt,
+                  error.message.includes('execution deadline'),
+                  error,
+                  indeterminate || (error instanceof IsolatedVmError && error.code == 'executor-crashed'),
+                ),
+              ),
             onSuccess: (output) =>
               Effect.try({ try: () => this.#commitOutcome(run, startedAt, budgetMs, output), catch: (error) => error }).pipe(
                 Effect.catch((error) => Effect.sync(() => this.#failRun(run, startedAt, false, error))),
               ),
           }),
-          Effect.ensuring(Effect.sync(() => this.#artifacts.discard(run.runId))),
+          Effect.ensuring(
+            Effect.sync(() => {
+              this.#slots.releaseRun(run.runId)
+              this.#artifacts.discard(run.runId)
+            }),
+          ),
         )
     })
   }
@@ -432,7 +453,7 @@ export class RunExecutor {
 
   #failRun(run: StoredRun, startedAt: number, timedOut: boolean, error: unknown, indeterminate = false): void {
     const result = indeterminate
-      ? { error: { code: 'execution.terminal-unknown', message: 'A Connector action outcome is unknown.' } }
+      ? { error: { code: 'execution.terminal-unknown', message: 'An execution outcome is unknown.' } }
       : timedOut
         ? { error: { code: 'run.timeout', message: 'The Run exceeded its execution deadline.' } }
         : { error: { code: 'run.failed', message: 'The Flow could not be completed.' } }
@@ -473,8 +494,11 @@ export class RunExecutor {
     const task = node.task
     const executor = task.executor
     switch (executor.kind) {
-      case 'workflow':
-        return this.#invokeWorkflow(run, invocation.nodeId, invocation.jobId, invocation.input, node, invocation.workflowRunId)
+      case 'workflow': {
+        const result = await this.#invokeWorkflow(run, invocation.nodeId, invocation.jobId, invocation.input, node, invocation.workflowRunId)
+        if (result.kind == 'calling') this.#slots.release(run.runId, invocation.jobId)
+        return result
+      }
       case 'skip-duplicates': {
         if (run.source == null) throw new StateError('capability.denied', 'State requires a Run source.')
         const items = invocation.input.items!
@@ -757,6 +781,7 @@ export class RunExecutor {
       signal.throwIfAborted()
       this.#pendingWorkflowCalls.set(run.runId, (this.#pendingWorkflowCalls.get(run.runId) ?? 0) + 1)
       this.waitingWorkflows.add(run.runId)
+      this.#slots.suspend(run.runId, jobId)
       this.#wakeMaintenance()
       try {
         return await new Promise<Readonly<Record<string, JsonValue>>>((resolve, reject) => {
@@ -791,6 +816,7 @@ export class RunExecutor {
         if (pending == 0) this.#pendingWorkflowCalls.delete(run.runId)
         else this.#pendingWorkflowCalls.set(run.runId, pending)
         await this.#resumeWorkflowCaller(run.runId, signal)
+        if (!signal.aborted) await this.#slots.resume(run.runId, jobId, signal)
       }
     } finally {
       this.#toolCalls.delete(key)

apps/server/node/application/service-options.ts

@@ -9,6 +9,7 @@ import type { IntegrationOptions } from '../runtime/integration-runtime.ts'
 export interface ServerRuntime {
   readonly stateLimits?: Partial<typeof import('../storage/state-store.ts').stateLimits>
   readonly maxConcurrentRuns?: number
+  readonly maxConcurrentTasks?: number
   readonly maxPendingRuns?: number
   readonly runEventRetentionMs?: number
   readonly runTimeoutMs?: number
@@ -56,6 +57,7 @@ export function validateRuntime(runtime: ServerRuntime): void {
   validatePositiveInteger(runtime.runEventRetentionMs, 'Run event retention must be a positive safe integer number of milliseconds.')
   validatePositiveInteger(runtime.maxPendingRuns, 'Maximum pending Runs must be a positive safe integer.')
   validatePositiveInteger(runtime.maxConcurrentRuns, 'Maximum concurrent Runs must be a positive safe integer.')
+  validatePositiveInteger(runtime.maxConcurrentTasks, 'Maximum concurrent Tasks must be a positive safe integer.')
   validatePositiveInteger(runtime.runTimeoutMs, 'Run timeout must be a positive safe integer number of milliseconds.')
   validatePositiveInteger(runtime.taskCpuMs, 'Task CPU limit must be a positive safe integer number of milliseconds.')
 }

apps/server/node/application/service.ts

@@ -161,6 +161,7 @@ export class ServerService {
           if (event.kind == 'run.changed' && event.runId == runId) changed()
         }),
       connectorFailureCodes,
+      runtime.maxConcurrentTasks ?? 16,
     )
     const pollDefinitions = triggerDefinitions.filter((definition): definition is PollDefinition => definition.snapshot.type == 'poll')
     const integrationDefinitions = triggerDefinitions.filter((definition): definition is IntegrationDefinition => definition.snapshot.type == 'integration')

apps/server/node/main.ts

@@ -52,6 +52,10 @@ function main(): Effect.Effect<void> {
       if (!Number.isSafeInteger(maxConcurrentRuns) || maxConcurrentRuns <= 0) {
         throw new Error('OPEN_FLOW_MAX_CONCURRENT_RUNS must be a positive safe integer.')
       }
+      const maxConcurrentTasks = Number(process.env.OPEN_FLOW_MAX_CONCURRENT_TASKS ?? '16')
+      if (!Number.isSafeInteger(maxConcurrentTasks) || maxConcurrentTasks <= 0) {
+        throw new Error('OPEN_FLOW_MAX_CONCURRENT_TASKS must be a positive safe integer.')
+      }
       const runTimeoutMs = Number(process.env.OPEN_FLOW_RUN_TIMEOUT_MS ?? '1800000')
       if (!Number.isSafeInteger(runTimeoutMs) || runTimeoutMs <= 0) {
         throw new Error('OPEN_FLOW_RUN_TIMEOUT_MS must be a positive safe integer.')
@@ -127,6 +131,7 @@ function main(): Effect.Effect<void> {
         logger,
         runtime: {
           maxConcurrentRuns,
+          maxConcurrentTasks,
           maxPendingRuns,
           runEventRetentionMs,
           runTimeoutMs,