Chapter 37: Workflow Orchestration

1. Why Workflow Orchestration

The Problem

Modern systems require coordinating multi-step processes that span services, involve human approvals, run for hours or days, and must handle failures gracefully. Consider these scenarios:

• E-commerce order: Reserve inventory → charge payment → ship order → send notifications (each step can fail)
• Data pipeline: Extract → transform → validate → load (with retries, scheduling, dependencies)
• Document processing: Upload → OCR → classify → extract → validate → index (long-running, need progress tracking)
• Account provisioning: Create user → provision resources → send credentials → wait for activation (hours/days, timeouts)

What Workflow Orchestration Provides

Orchestration vs Choreography

2. Temporal

Temporal is a durable execution platform that guarantees workflows run to completion despite failures. It's the successor to AWS SWF and Uber's Cadence.

Core Concepts

• Workflow: Durable function representing the orchestration logic (must be deterministic)
• Activity: Individual task that can have side effects (API calls, DB writes)
• Worker: Process that executes workflow/activity code
• Task Queue: Named queue that routes work to workers
• Event History: Complete log of workflow execution (enables replay)

Architecture

Key Features

// Activity automatically retries on failure
ActivityOptions {
  startToCloseTimeout: 5min
  retryPolicy: {
    initialInterval: 1s
    maxInterval: 1min
    maxAttempts: 10
    backoffCoefficient: 2.0
  }
}

// Wait for approval or timeout
select {
  case approved := <-approvalSignal:
    // Continue workflow
  case <-sleep(24h):
    // Timeout, cancel order
}

Signals, Queries, Updates

3. Temporal Programming Model

Workflow Functions (Deterministic)

Workflows must be deterministic for replay to work. Rules:

• No direct I/O (no HTTP calls, no DB access, no random numbers)
• All side effects must be in Activities
• Use workflow-provided time/random functions (deterministic)
• No threading or async/await outside workflow APIs

// Order processing workflow (Python-style pseudocode)
@workflow.defn
class OrderWorkflow:
    def __init__(self):
        self.order_status = "pending"

    @workflow.run
    async def run(self, order_id: str) -> str:
        # Step 1: Reserve inventory (activity)
        inventory_result = await workflow.execute_activity(
            reserve_inventory,
            args=[order_id],
            start_to_close_timeout=timedelta(minutes=5)
        )

        if not inventory_result.success:
            return "insufficient_inventory"

        # Step 2: Charge payment (activity with retries)
        try:
            payment_result = await workflow.execute_activity(
                charge_payment,
                args=[order_id],
                start_to_close_timeout=timedelta(minutes=2),
                retry_policy=RetryPolicy(
                    max_attempts=3,
                    initial_interval=timedelta(seconds=1),
                    backoff_coefficient=2.0
                )
            )
        except ActivityError as e:
            # Payment failed, compensate: release inventory
            await workflow.execute_activity(
                release_inventory,
                args=[order_id]
            )
            return "payment_failed"

        # Step 3: Ship order (activity)
        shipping_result = await workflow.execute_activity(
            ship_order,
            args=[order_id],
            start_to_close_timeout=timedelta(hours=1)
        )

        # Step 4: Wait for delivery confirmation or timeout
        delivery_confirmed = False
        try:
            await workflow.wait_condition(
                lambda: self.order_status == "delivered",
                timeout=timedelta(days=7)
            )
            delivery_confirmed = True
        except TimeoutError:
            # Timeout, trigger investigation
            await workflow.execute_activity(
                investigate_delivery,
                args=[order_id]
            )

        # Step 5: Send notification
        await workflow.execute_activity(
            send_notification,
            args=[order_id, delivery_confirmed]
        )

        return "completed"

    @workflow.signal
    def update_status(self, status: str):
        self.order_status = status

    @workflow.query
    def get_status(self) -> str:
        return self.order_status

Activities (Side Effects)

// Activities perform actual work (can fail, non-deterministic)
@activity.defn
async def charge_payment(order_id: str) -> PaymentResult:
    # This is where real I/O happens
    response = await stripe_client.charge(
        order_id=order_id,
        amount=get_order_amount(order_id)
    )

    # Activity can be retried automatically
    if response.status == "insufficient_funds":
        raise ApplicationError("Payment declined")

    return PaymentResult(
        success=True,
        transaction_id=response.id
    )

@activity.defn
async def ship_order(order_id: str) -> ShippingResult:
    # Call shipping API
    tracking_number = await shipping_api.create_shipment(order_id)

    # Write to database
    await db.update_order_status(order_id, "shipped", tracking_number)

    return ShippingResult(tracking_number=tracking_number)

Child Workflows

// Complex workflows can spawn child workflows
@workflow.run
async def parent_workflow(customer_id: str):
    # Process multiple orders in parallel
    order_ids = await workflow.execute_activity(get_pending_orders, [customer_id])

    # Start child workflow for each order
    child_handles = []
    for order_id in order_ids:
        handle = await workflow.start_child_workflow(
            OrderWorkflow,
            args=[order_id],
            id=f"order-{order_id}"
        )
        child_handles.append(handle)

    # Wait for all to complete
    results = await asyncio.gather(*[h.result() for h in child_handles])

    return results

4. Apache Airflow

Apache Airflow is a platform for programmatically authoring, scheduling, and monitoring data pipelines as Directed Acyclic Graphs (DAGs).

Core Concepts

• DAG: Directed Acyclic Graph defining task dependencies
• Operator: Template for a task (PythonOperator, BashOperator, SQLOperator)
• Task: Instance of an operator
• Scheduler: Triggers DAGs based on schedule, monitors completion
• Executor: Runs tasks (LocalExecutor, CeleryExecutor, KubernetesExecutor)
• XCom: Cross-communication between tasks (small data sharing)

Architecture

DAG Definition

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'data_pipeline',
    default_args=default_args,
    description='ETL pipeline for sales data',
    schedule_interval='0 2 * * *',  # 2 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'sales']
) as dag:

    # Task 1: Extract data from API
    extract = PythonOperator(
        task_id='extract_sales_data',
        python_callable=extract_sales_data,
        op_kwargs={'date': '{{ ds }}'}  # Templated execution date
    )

    # Task 2: Transform data
    transform = PythonOperator(
        task_id='transform_data',
        python_callable=transform_sales_data
    )

    # Task 3: Validate data quality
    validate = PythonOperator(
        task_id='validate_data',
        python_callable=validate_sales_data
    )

    # Task 4: Load to warehouse
    load = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_to_warehouse
    )

    # Task 5: Run analytics
    analytics = BashOperator(
        task_id='run_analytics',
        bash_command='dbt run --models sales_analytics'
    )

    # Define dependencies
    extract >> transform >> validate >> load >> analytics

    # Alternative syntax:
    # extract.set_downstream(transform)
    # transform.set_downstream([validate, load])  # Parallel

Executors

XComs (Cross-Communication)

def extract_sales_data(**context):
    data = fetch_from_api()

    # Push data to XCom (small data only, stored in DB)
    context['ti'].xcom_push(key='sales_data', value=data)

    return len(data)  # Also auto-pushed to XCom with key='return_value'

def transform_sales_data(**context):
    # Pull data from previous task
    ti = context['ti']
    sales_data = ti.xcom_pull(
        task_ids='extract_sales_data',
        key='sales_data'
    )

    transformed = transform(sales_data)
    return transformed

Backfills & Catchup

# Backfill: Re-run historical DAG runs
# If DAG was paused/failed, catch up on missed runs
with DAG(
    'daily_reports',
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=True  # Run all missed intervals since start_date
) as dag:
    # Tasks...

# Command line: backfill specific date range
# airflow dags backfill -s 2024-01-01 -e 2024-01-31 daily_reports

5. AWS Step Functions

AWS Step Functions is a serverless orchestration service using state machines defined in Amazon States Language (ASL) JSON.

Workflow Types

State Types

• Task: Invoke Lambda, API Gateway, DynamoDB, ECS, Batch, SNS, SQS, etc.
• Choice: Conditional branching (if-else logic)
• Parallel: Execute branches concurrently
• Map: Iterate over array, execute state for each item
• Wait: Delay for duration or until timestamp
• Pass: Pass input to output (transform, inject data)
• Succeed/Fail: Terminal states

State Machine Diagram

ASL Definition

{
  "Comment": "Order processing state machine",
  "StartAt": "ValidateInput",
  "States": {
    "ValidateInput": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:validate-input",
      "Next": "CheckValid",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["ValidationError"],
          "ResultPath": "$.error",
          "Next": "InvalidInput"
        }
      ]
    },
    "CheckValid": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.isValid",
          "BooleanEquals": true,
          "Next": "ProcessOrder"
        }
      ],
      "Default": "InvalidInput"
    },
    "ProcessOrder": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:putItem",
      "Parameters": {
        "TableName": "Orders",
        "Item": {
          "orderId": {"S.$": "$.orderId"},
          "status": {"S": "processing"}
        }
      },
      "Next": "ChargePayment"
    },
    "ChargePayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:charge-payment",
      "ResultPath": "$.paymentResult",
      "Next": "CheckPayment"
    },
    "CheckPayment": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.paymentResult.success",
          "BooleanEquals": true,
          "Next": "ParallelShipAndNotify"
        }
      ],
      "Default": "PaymentFailed"
    },
    "ParallelShipAndNotify": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "ShipOrder",
          "States": {
            "ShipOrder": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:123456789:function:ship-order",
              "End": true
            }
          }
        },
        {
          "StartAt": "SendEmail",
          "States": {
            "SendEmail": {
              "Type": "Task",
              "Resource": "arn:aws:states:::sns:publish",
              "Parameters": {
                "TopicArn": "arn:aws:sns:us-east-1:123456789:order-notifications",
                "Message.$": "$.orderId"
              },
              "End": true
            }
          }
        }
      ],
      "Next": "Success"
    },
    "PaymentFailed": {
      "Type": "Fail",
      "Error": "PaymentError",
      "Cause": "Payment processing failed"
    },
    "InvalidInput": {
      "Type": "Fail",
      "Error": "InvalidInput",
      "Cause": "Input validation failed"
    },
    "Success": {
      "Type": "Succeed"
    }
  }
}

Map State (Dynamic Parallelism)

{
  "ProcessOrders": {
    "Type": "Map",
    "ItemsPath": "$.orders",
    "MaxConcurrency": 10,
    "Iterator": {
      "StartAt": "ProcessSingleOrder",
      "States": {
        "ProcessSingleOrder": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:us-east-1:123456789:function:process-order",
          "End": true
        }
      }
    },
    "ResultPath": "$.processedOrders",
    "Next": "AllOrdersProcessed"
  }
}

6. State Machines

Formal Definition

A finite state machine (FSM) is a mathematical model defined by:

• States: Finite set of conditions (S = {s1, s2, ..., sn})
• Initial state: Starting point (s0 ∈ S)
• Transitions: Rules for moving between states (δ: S × E → S)
• Events: Inputs that trigger transitions (E = {e1, e2, ..., em})
• Actions: Operations performed during transitions (optional)
• Final states: Terminal conditions (F ⊆ S)

Order State Machine Example

Advantages of State Machines

Implementation Pattern

// State machine implementation (pseudocode)
class OrderStateMachine {
  state: State = State.CREATED

  // State transition table
  transitions = {
    CREATED: {
      SUBMIT: State.PAYMENT_PENDING,
      CANCEL: State.CANCELLED
    },
    PAYMENT_PENDING: {
      PAYMENT_SUCCESS: State.PAID,
      PAYMENT_FAILED: State.PAYMENT_FAILED,
      CANCEL: State.CANCELLED
    },
    PAID: {
      SHIP: State.SHIPPED,
      REFUND: State.CANCELLED
    },
    SHIPPED: {
      PICKUP: State.IN_TRANSIT
    },
    IN_TRANSIT: {
      DELIVER: State.DELIVERED
    }
  }

  // Trigger event
  trigger(event: Event) {
    allowed_transitions = transitions[this.state]

    if event not in allowed_transitions:
      throw InvalidTransitionError(
        f"Cannot {event} from state {this.state}"
      )

    old_state = this.state
    new_state = allowed_transitions[event]

    // Execute exit action for old state
    on_exit(old_state)

    // Perform transition
    this.state = new_state

    // Execute entry action for new state
    on_entry(new_state)

    // Log transition
    log_transition(old_state, event, new_state)
  }

  // State entry/exit hooks
  on_entry(state: State) {
    match state:
      PAYMENT_PENDING: initiate_payment()
      SHIPPED: notify_customer()
      DELIVERED: close_order()
  }

  on_exit(state: State) {
    // Cleanup actions when leaving state
  }
}

Hierarchical State Machines

Complex workflows benefit from nested states (substates):

7. Compensation & Rollback

The Problem

Distributed transactions can't use traditional ACID transactions across services. If step 5 of 10 fails, you need to undo steps 1-4. This is compensation.

Saga Pattern

A saga is a sequence of local transactions. If one fails, execute compensating transactions to undo previous steps.

Compensation Implementation

// Temporal compensation pattern
@workflow.run
async def booking_workflow(booking_id: str):
    compensations = []

    try:
        # Step 1: Reserve hotel
        hotel_result = await workflow.execute_activity(reserve_hotel, [booking_id])
        compensations.append(lambda: cancel_hotel_reservation(hotel_result.reservation_id))

        # Step 2: Book flight
        flight_result = await workflow.execute_activity(book_flight, [booking_id])
        compensations.append(lambda: cancel_flight(flight_result.ticket_id))

        # Step 3: Charge customer
        payment_result = await workflow.execute_activity(charge_payment, [booking_id])
        compensations.append(lambda: refund_payment(payment_result.transaction_id))

        # Step 4: Send confirmation
        await workflow.execute_activity(send_confirmation, [booking_id])

        return "success"

    except Exception as e:
        # Execute compensations in reverse order
        for compensation in reversed(compensations):
            try:
                await workflow.execute_activity(compensation)
            except Exception as comp_error:
                # Log compensation failure, may need manual intervention
                await workflow.execute_activity(
                    log_compensation_failure,
                    [booking_id, str(comp_error)]
                )

        raise e  # Re-raise to mark workflow as failed

Compensation Strategies

Idempotency

Compensation actions must be idempotent (safe to retry). Use idempotency keys:

// Idempotent refund activity
async def refund_payment(transaction_id: str):
    idempotency_key = f"refund-{transaction_id}"

    # Check if refund already processed
    existing_refund = await db.get_refund_by_key(idempotency_key)
    if existing_refund:
        return existing_refund  # Already done, return cached result

    # Process refund
    refund = await payment_gateway.refund(transaction_id)

    # Store with idempotency key
    await db.store_refund(refund, idempotency_key)

    return refund

Human-in-the-Loop

Some failures require human approval to proceed or compensate:

@workflow.run
async def high_value_order(order_id: str, amount: float):
    # Process order normally
    result = await process_order(order_id)

    # If payment is large, wait for manual approval
    if amount > 10000:
        approved = await workflow.wait_condition(
            lambda: self.approval_status == "approved",
            timeout=timedelta(hours=24)
        )

        if not approved:
            # Timeout, compensate
            await compensate_order(order_id)
            return "approval_timeout"

    # Continue...

@workflow.signal
def approve_order(self):
    self.approval_status = "approved"

8. Workflow Patterns

1. Saga Orchestration

Centralized orchestrator directs saga steps and compensations (already covered above).

2. Fan-Out/Fan-In

Execute multiple tasks in parallel, then wait for all to complete before proceeding.

// Temporal fan-out/fan-in
@workflow.run
async def process_batch(item_ids: List[str]):
    # Fan-out: start activities in parallel
    activity_futures = []
    for item_id in item_ids:
        future = workflow.execute_activity(
            process_item,
            args=[item_id],
            start_to_close_timeout=timedelta(minutes=5)
        )
        activity_futures.append(future)

    # Fan-in: wait for all to complete
    results = await asyncio.gather(*activity_futures)

    # Aggregate results
    summary = aggregate_results(results)
    return summary

3. Human Approval

Pause workflow until human provides input (signal/callback).

// Temporal human approval
@workflow.run
async def approval_workflow(request_id: str):
    # Submit request
    await workflow.execute_activity(submit_request, [request_id])

    # Wait for approval signal (or timeout)
    try:
        approved = await workflow.wait_condition(
            lambda: self.approval_status is not None,
            timeout=timedelta(days=3)
        )
    except TimeoutError:
        # Timeout, auto-reject
        await workflow.execute_activity(notify_timeout, [request_id])
        return "timeout"

    if self.approval_status == "approved":
        await workflow.execute_activity(execute_request, [request_id])
        return "approved"
    else:
        await workflow.execute_activity(cancel_request, [request_id])
        return "rejected"

@workflow.signal
def approve(self):
    self.approval_status = "approved"

@workflow.signal
def reject(self):
    self.approval_status = "rejected"

4. Scheduled Workflows

Trigger workflows on a schedule (cron).

// Airflow scheduled DAG
with DAG(
    'daily_report',
    schedule_interval='0 9 * * *',  # 9 AM daily
    start_date=datetime(2024, 1, 1)
) as dag:
    generate_report = PythonOperator(
        task_id='generate_report',
        python_callable=generate_daily_report
    )

// Temporal scheduled workflow
async def main():
    client = await Client.connect("localhost:7233")

    # Schedule workflow to run daily
    await client.schedule_workflow(
        workflow=DailyReportWorkflow,
        schedule_id="daily-report",
        schedule={
            "cron": "0 9 * * *",  # 9 AM daily
            "timezone": "America/New_York"
        }
    )

5. Event-Driven Triggers

Start workflow in response to events (S3 upload, SQS message, webhook).

// AWS Step Functions triggered by EventBridge
{
  "source": ["aws.s3"],
  "detail-type": ["Object Created"],
  "detail": {
    "bucket": {"name": ["my-upload-bucket"]}
  }
}
// -> Triggers Step Functions execution

// Temporal workflow started from event
async def event_handler(event: S3Event):
    client = await Client.connect("localhost:7233")

    await client.start_workflow(
        ProcessUploadWorkflow,
        args=[event.bucket, event.key],
        id=f"upload-{event.key}",
        task_queue="upload-processing"
    )

6. Polling Pattern

Periodically check external system until condition met.

// Temporal polling with sleep
@workflow.run
async def wait_for_external_system(job_id: str):
    max_attempts = 60  # 1 hour (60 * 1 min)

    for attempt in range(max_attempts):
        # Check status
        status = await workflow.execute_activity(
            check_job_status,
            args=[job_id],
            start_to_close_timeout=timedelta(seconds=30)
        )

        if status == "completed":
            return "success"
        elif status == "failed":
            return "failed"

        # Wait 1 minute before next check
        await asyncio.sleep(60)

    # Timeout
    return "timeout"

7. Dynamic Workflow

Workflow structure determined at runtime (data-driven).

// Dynamic task graph based on config
@workflow.run
async def dynamic_pipeline(config: PipelineConfig):
    results = {}

    for stage in config.stages:
        # Determine activity based on stage type
        if stage.type == "transform":
            activity = transform_activity
        elif stage.type == "validate":
            activity = validate_activity
        elif stage.type == "enrich":
            activity = enrich_activity
        else:
            raise ValueError(f"Unknown stage type: {stage.type}")

        # Execute activity with stage-specific config
        result = await workflow.execute_activity(
            activity,
            args=[results, stage.config],
            start_to_close_timeout=timedelta(minutes=stage.timeout)
        )

        results[stage.name] = result

    return results

9. Comparison

Architecture Comparison

10. When to Use What

Decision Tree

Use Temporal When...

• Building microservice orchestration (saga pattern)
• Long-running processes (hours, days, weeks)
• Need human-in-the-loop approvals
• Complex state management across steps
• Requirement for exactly-once execution guarantees
• Want to write workflows in your application language
• Need multi-cloud or on-prem deployment

Use Airflow When...

• Building data pipelines (ETL, ELT)
• Need complex scheduling (cron, backfills, data-aware schedules)
• Task dependencies form a DAG
• Team is Python-focused
• Need extensive integrations (Airflow has 1000+ operators)
• Batch processing workloads

Use Step Functions When...

• Already on AWS and want managed service
• Tight integration with AWS services (Lambda, ECS, DynamoDB, etc.)
• Workflows are relatively simple (< 25k events)
• Want zero infrastructure management
• High-volume, short-duration workflows (Express)

Use Argo Workflows When...

• Already on Kubernetes
• ML pipelines (can integrate with Kubeflow)
• CI/CD workflows
• Each step needs isolated environment (containers)
• Need resource management per step (CPU/memory limits)

Use Simple Queue When...

• Single-step async processing (no multi-step coordination)
• No need for workflow visibility
• Simple retry logic sufficient
• Low complexity (adding orchestration would be overkill)

Hybrid Approaches

Often systems use multiple tools:

• Airflow + Temporal: Airflow schedules batch jobs, each triggers Temporal workflows for per-item processing
• Step Functions + Lambda: Step Functions orchestrates, Lambda executes individual tasks
• Kafka + Temporal: Kafka for event streaming, Temporal for stateful workflow orchestration
• Argo + Airflow: Airflow schedules pipelines, Argo executes containerized steps on K8s

Cost Considerations