Workflow Execution

Overview

The Workflow Engine executes workflow templates as DAGs (Directed Acyclic Graphs) of steps, managing state transitions, parallelism, retries, and failure handling.

Execution Architecture

                         WORKFLOW EXECUTION ARCHITECTURE

  ┌─────────────────────────────────────────────────────────────────────────────┐
  │                           WORKFLOW ENGINE                                    │
  │                                                                             │
  │  ┌─────────────────────────────────────────────────────────────────────┐   │
  │  │                        WORKFLOW RUNNER                               │   │
  │  │                                                                      │   │
  │  │  ┌────────────┐    ┌────────────┐    ┌────────────┐                │   │
  │  │  │ Template   │───►│ Execution  │───►│ Context    │                │   │
  │  │  │ Parser     │    │ Planner    │    │ Builder    │                │   │
  │  │  └────────────┘    └────────────┘    └────────────┘                │   │
  │  │         │                │                 │                        │   │
  │  │         └────────────────┼─────────────────┘                        │   │
  │  │                          ▼                                          │   │
  │  │  ┌─────────────────────────────────────────────────────────────┐   │   │
  │  │  │                    DAG EXECUTOR                              │   │   │
  │  │  │                                                              │   │   │
  │  │  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │   │   │
  │  │  │  │ Ready    │  │ Running  │  │ Waiting  │  │ Completed│   │   │   │
  │  │  │  │ Queue    │  │ Set      │  │ Set      │  │ Set      │   │   │   │
  │  │  │  └──────────┘  └──────────┘  └──────────┘  └──────────┘   │   │   │
  │  │  │                                                              │   │   │
  │  │  │  ┌──────────────────────────────────────────────────────┐   │   │   │
  │  │  │  │                 STEP DISPATCHER                       │   │   │   │
  │  │  │  └──────────────────────────────────────────────────────┘   │   │   │
  │  │  └─────────────────────────────────────────────────────────────┘   │   │
  │  └─────────────────────────────────────────────────────────────────────┘   │
  │                                    │                                        │
  │                                    ▼                                        │
  │  ┌─────────────────────────────────────────────────────────────────────┐   │
  │  │                        STEP EXECUTOR POOL                            │   │
  │  │                                                                      │   │
  │  │  ┌────────────┐  ┌────────────┐  ┌────────────┐  ┌────────────┐   │   │
  │  │  │ Executor 1 │  │ Executor 2 │  │ Executor 3 │  │ Executor N │   │   │
  │  │  └────────────┘  └────────────┘  └────────────┘  └────────────┘   │   │
  │  │                                                                      │   │
  │  └─────────────────────────────────────────────────────────────────────┘   │
  │                                                                             │
  └─────────────────────────────────────────────────────────────────────────────┘

Workflow Run State Machine

                         WORKFLOW RUN STATES

                            ┌──────────┐
                            │ CREATED  │
                            └────┬─────┘
                                 │ start()
                                 ▼
                            ┌──────────┐
                            │ RUNNING  │◄──────────────────┐
                            └────┬─────┘                   │
                                 │                         │
             ┌───────────────────┼───────────────────┐     │
             │                   │                   │     │
             ▼                   ▼                   ▼     │
        ┌──────────┐       ┌──────────┐       ┌──────────┐│
        │ WAITING  │       │ PAUSED   │       │ FAILING  ││
        │ APPROVAL │       │          │       │          ││
        └────┬─────┘       └────┬─────┘       └────┬─────┘│
             │                  │                  │      │
             │ approve()        │ resume()         │      │
             │                  │                  │      │
             └───────────────►──┴──────────────────┘      │
                                │                         │
                                └─────────────────────────┘
                                │
        ┌───────────────────────┼───────────────────┐
        │                       │                   │
        ▼                       ▼                   ▼
   ┌──────────┐           ┌──────────┐       ┌──────────┐
   │COMPLETED │           │ FAILED   │       │ CANCELLED│
   └──────────┘           └──────────┘       └──────────┘

State Transitions

Step Execution State Machine

                           STEP STATES

                           ┌──────────┐
                           │ PENDING  │
                           └────┬─────┘
                                │ schedule()
                                ▼
                           ┌──────────┐
                           │ QUEUED   │
                           └────┬─────┘
                                │ dispatch()
                                ▼
                           ┌──────────┐
                           │ RUNNING  │◄─────────┐
                           └────┬─────┘          │
                                │                │ retry()
            ┌───────────────────┼───────────────┐│
            │                   │               ││
            ▼                   ▼               ▼│
       ┌──────────┐       ┌──────────┐    ┌──────────┐
       │SUCCEEDED │       │ FAILED   │    │ RETRYING │
       └──────────┘       └────┬─────┘    └──────────┘
                               │
                               ▼
                    ┌─────────────────────┐
                    │  FAILURE HANDLER    │
                    │  ┌───────────────┐  │
                    │  │ fail          │──┼─► Mark workflow failing
                    │  │ continue      │──┼─► Continue to next step
                    │  │ rollback      │──┼─► Trigger rollback path
                    │  │ goto:{nodeId} │──┼─► Jump to specific node
                    │  └───────────────┘  │
                    └─────────────────────┘

Step States

DAG Execution Algorithm

class DAGExecutor:
    def __init__(self, workflow_run: WorkflowRun):
        self.run = workflow_run
        self.template = workflow_run.template
        self.pending = set(node.id for node in template.nodes)
        self.running = set()
        self.completed = set()
        self.failed = set()
        self.outputs = {}  # nodeId -> outputs

    async def execute(self):
        """Main execution loop."""
        self.run.status = WorkflowStatus.RUNNING
        self.run.started_at = datetime.utcnow()

        while self.pending or self.running:
            # Find ready nodes (all dependencies satisfied)
            ready = self.find_ready_nodes()

            # Dispatch ready nodes
            for node_id in ready:
                asyncio.create_task(self.execute_node(node_id))
                self.pending.remove(node_id)
                self.running.add(node_id)

            # Wait for any node to complete
            if self.running:
                await self.wait_for_completion()

            # Check for deadlock
            if not ready and self.pending and not self.running:
                raise DeadlockException(self.pending)

        # Determine final status
        if self.failed:
            self.run.status = WorkflowStatus.FAILED
        else:
            self.run.status = WorkflowStatus.COMPLETED

        self.run.completed_at = datetime.utcnow()

    def find_ready_nodes(self) -> List[str]:
        """Find nodes whose dependencies are all complete."""
        ready = []
        for node_id in self.pending:
            node = self.template.get_node(node_id)

            # Check condition
            if node.condition:
                if not self.evaluate_condition(node.condition):
                    self.mark_skipped(node_id)
                    continue

            # Check all incoming edges
            incoming = self.template.get_incoming_edges(node_id)
            dependencies_met = all(
                edge.from_node in self.completed
                for edge in incoming
                if self.evaluate_edge_condition(edge)
            )

            if dependencies_met:
                ready.append(node_id)

        return ready

    async def execute_node(self, node_id: str):
        """Execute a single node."""
        node = self.template.get_node(node_id)
        step_run = StepRun(
            workflow_run_id=self.run.id,
            node_id=node_id,
            status=StepStatus.RUNNING
        )

        try:
            # Resolve inputs
            inputs = self.resolve_inputs(node)

            # Get step executor
            executor = self.step_registry.get_executor(node.type)

            # Execute with timeout
            async with asyncio.timeout(node.timeout):
                outputs = await executor.execute(inputs, node.config)

            # Store outputs
            self.outputs[node_id] = outputs
            step_run.outputs = outputs
            step_run.status = StepStatus.SUCCEEDED

            self.running.remove(node_id)
            self.completed.add(node_id)

        except Exception as e:
            await self.handle_step_failure(node, step_run, e)

    async def handle_step_failure(self, node, step_run, error):
        """Handle step failure according to retry and failure policies."""
        step_run.attempt_number += 1

        # Check retry policy
        if step_run.attempt_number <= node.retry_policy.max_retries:
            if self.is_retryable(error, node.retry_policy):
                step_run.status = StepStatus.RETRYING
                delay = self.calculate_backoff(node.retry_policy, step_run.attempt_number)
                await asyncio.sleep(delay)
                await self.execute_node(node.id)  # Retry
                return

        # No more retries - handle failure
        step_run.status = StepStatus.FAILED
        step_run.error = str(error)

        match node.on_failure:
            case "fail":
                self.run.status = WorkflowStatus.FAILING
                self.failed.add(node.id)
            case "continue":
                self.completed.add(node.id)  # Continue as if succeeded
            case "rollback":
                await self.trigger_rollback(node)
            case _ if node.on_failure.startswith("goto:"):
                target = node.on_failure.split(":")[1]
                self.pending.add(target)  # Add target to pending

        self.running.remove(node.id)

Input Resolution

Inputs to steps can come from multiple sources:

interface InputResolver {
  resolve(binding: InputBinding, context: ExecutionContext): any;
}

class StandardInputResolver implements InputResolver {
  resolve(binding: InputBinding, context: ExecutionContext): any {
    switch (binding.source.type) {
      case "literal":
        return binding.source.value;

      case "context":
        // Navigate context path: "release.name" -> context.release.name
        return this.navigatePath(context, binding.source.path);

      case "output":
        // Get output from previous step
        const stepOutputs = context.stepOutputs[binding.source.nodeId];
        return stepOutputs?.[binding.source.outputName];

      case "secret":
        // Fetch from vault (never cached)
        return this.secretsClient.fetch(binding.source.secretName);

      case "expression":
        // Evaluate JavaScript expression
        return this.expressionEvaluator.evaluate(
          binding.source.expression,
          context
        );
    }
  }
}

Execution Context

The execution context provides data available to all steps:

interface ExecutionContext {
  // Workflow identifiers
  workflowRunId: UUID;
  templateId: UUID;
  templateVersion: number;

  // Input values
  inputs: Record<string, any>;

  // Domain objects (loaded at start)
  release?: Release;
  promotion?: Promotion;
  environment?: Environment;
  targets?: Target[];

  // Step outputs (accumulated during execution)
  stepOutputs: Record<string, Record<string, any>>;

  // Tenant context
  tenantId: UUID;
  userId: UUID;

  // Metadata
  startedAt: DateTime;
  correlationId: string;
}

Concurrency Control

Parallelism Within Workflows

interface ParallelConfig {
  maxConcurrency: number;     // Max simultaneous steps
  failFast: boolean;          // Stop all on first failure
}

// Example: Parallel deployment to multiple targets
const parallelDeploy: StepNode = {
  id: "parallel-deploy",
  type: "parallel",
  config: {
    maxConcurrency: 5,
    failFast: false
  },
  children: [
    { id: "deploy-target-1", type: "deploy-docker", ... },
    { id: "deploy-target-2", type: "deploy-docker", ... },
    { id: "deploy-target-3", type: "deploy-docker", ... },
  ]
};

Global Concurrency Limits

interface ConcurrencyLimits {
  maxWorkflowsPerTenant: number;        // Concurrent workflow runs
  maxStepsPerWorkflow: number;          // Concurrent steps per workflow
  maxDeploymentsPerEnvironment: number; // Prevent deployment conflicts
}

// Default limits
const defaults: ConcurrencyLimits = {
  maxWorkflowsPerTenant: 10,
  maxStepsPerWorkflow: 20,
  maxDeploymentsPerEnvironment: 1  // One deployment at a time
};

Checkpoint and Resume

Workflows support checkpointing for long-running executions:

interface WorkflowCheckpoint {
  workflowRunId: UUID;
  checkpointedAt: DateTime;

  // Execution state
  pendingNodes: string[];
  completedNodes: string[];
  failedNodes: string[];

  // Accumulated data
  stepOutputs: Record<string, Record<string, any>>;

  // Context snapshot
  contextSnapshot: ExecutionContext;
}

class CheckpointManager {
  // Save checkpoint after each step completion
  async saveCheckpoint(run: WorkflowRun): Promise<void> {
    const checkpoint: WorkflowCheckpoint = {
      workflowRunId: run.id,
      checkpointedAt: new Date(),
      pendingNodes: Array.from(run.executor.pending),
      completedNodes: Array.from(run.executor.completed),
      failedNodes: Array.from(run.executor.failed),
      stepOutputs: run.executor.outputs,
      contextSnapshot: run.context
    };

    await this.repository.save(checkpoint);
  }

  // Resume from checkpoint after service restart
  async resumeFromCheckpoint(workflowRunId: UUID): Promise<WorkflowRun> {
    const checkpoint = await this.repository.get(workflowRunId);

    const run = new WorkflowRun();
    run.executor.pending = new Set(checkpoint.pendingNodes);
    run.executor.completed = new Set(checkpoint.completedNodes);
    run.executor.failed = new Set(checkpoint.failedNodes);
    run.executor.outputs = checkpoint.stepOutputs;
    run.context = checkpoint.contextSnapshot;

    // Resume execution
    await run.executor.execute();
    return run;
  }
}

Timeout Handling

interface TimeoutConfig {
  stepTimeout: number;       // Per-step timeout (seconds)
  workflowTimeout: number;   // Total workflow timeout (seconds)
}

class TimeoutHandler {
  async executeWithTimeout<T>(
    operation: () => Promise<T>,
    timeoutSeconds: number,
    onTimeout: () => Promise<void>
  ): Promise<T> {
    const controller = new AbortController();
    const timeoutId = setTimeout(
      () => controller.abort(),
      timeoutSeconds * 1000
    );

    try {
      const result = await operation();
      clearTimeout(timeoutId);
      return result;
    } catch (error) {
      if (error.name === 'AbortError') {
        await onTimeout();
        throw new TimeoutException(timeoutSeconds);
      }
      throw error;
    }
  }
}

Event Emission

The workflow engine emits events for observability:

type WorkflowEvent =
  | { type: "workflow.started"; workflowRunId: UUID; templateId: UUID }
  | { type: "workflow.completed"; workflowRunId: UUID; status: string }
  | { type: "workflow.failed"; workflowRunId: UUID; error: string }
  | { type: "step.started"; workflowRunId: UUID; nodeId: string }
  | { type: "step.completed"; workflowRunId: UUID; nodeId: string; outputs: any }
  | { type: "step.failed"; workflowRunId: UUID; nodeId: string; error: string }
  | { type: "step.retrying"; workflowRunId: UUID; nodeId: string; attempt: number };

class WorkflowEventEmitter {
  private subscribers: Map<string, ((event: WorkflowEvent) => void)[]> = new Map();

  emit(event: WorkflowEvent): void {
    const handlers = this.subscribers.get(event.type) || [];
    for (const handler of handlers) {
      handler(event);
    }

    // Also emit to event bus for external consumers
    this.eventBus.publish("workflow.events", event);
  }
}

Execution Monitoring

Real-time Progress

interface WorkflowProgress {
  workflowRunId: UUID;
  status: WorkflowStatus;

  // Step progress
  totalSteps: number;
  completedSteps: number;
  runningSteps: number;
  failedSteps: number;

  // Current activity
  currentNodes: string[];

  // Timing
  startedAt: DateTime;
  estimatedCompletion?: DateTime;

  // Step details
  steps: StepProgress[];
}

interface StepProgress {
  nodeId: string;
  nodeName: string;
  status: StepStatus;
  startedAt?: DateTime;
  completedAt?: DateTime;
  attempt: number;
  logs?: string;
}

WebSocket Streaming

// Client subscribes to workflow progress
const ws = new WebSocket(`/api/v1/workflow-runs/${runId}/stream`);

ws.onmessage = (event) => {
  const progress: WorkflowProgress = JSON.parse(event.data);
  updateUI(progress);
};

// Server streams updates
class WorkflowStreamHandler {
  async stream(runId: UUID, connection: WebSocket): Promise<void> {
    const subscription = this.eventBus.subscribe(`workflow.${runId}.*`);

    for await (const event of subscription) {
      const progress = await this.buildProgress(runId);
      connection.send(JSON.stringify(progress));

      if (progress.status === 'completed' || progress.status === 'failed') {
        break;
      }
    }

    connection.close();
  }
}

References

• Workflow Templates
• Workflow Engine Module
• Promotion Manager