Choreo Workflow: Comprehensive Walkthrough

Mix.install([
  {:choreo, "~> 0.14.1"},
  # {:choreo, path: Path.expand("../..", __DIR__), force: true},
  {:kino_vizjs, "~> 0.9.0"}
])

Section

Rendering diagrams: This livebook uses Kino.VizJS to render DOT diagrams inline and Choreo.Lab.Siren / Choreo.Lab.Sketch for Mermaid previews.

What is Choreo.Workflow?

Choreo.Workflow models business process orchestration and Saga transactions as directed graphs. Unlike drawing tools where boxes and arrows are just pictures, Choreo workflows are executable blueprints you can analyse for latency, parallelism, failure coverage, and structural correctness.

Common use cases:

• E-commerce order processing with rollback handlers
• CI/CD pipeline orchestration
• Multi-step approval flows
• Distributed Saga transactions

Node Types

Edge Types

API Approaches: Programmatic Pipe API vs Lab DSL

Choreo provides two ways to author workflows:

1. Programmatic Pipe API (Choreo.Workflow): Explicit, pipe-first builder functions (add_start/3, add_task/3, add_decision/3, add_fork/3, add_join/3, add_compensation/3, add_end/3, connect/4). Best for production services, dynamic DAG generation, and pipeline compilers.
2. Lab DSL (Choreo.Lab.DSL.Workflow): Concise, sketch-oriented syntax (workflow do ... end, variable binding, ~>, and pipe modifiers like |> condition("yes"), |> failure()). Best for Livebooks, process reviews, Saga sketches, and rapid orchestration modeling.

In this guide, the introductory legend demonstrates the programmatic pipe API, while subsequent examples showcase the expressive Lab DSL.

alias Choreo.Workflow
alias Choreo.Workflow.Analysis
import Choreo.Lab.DSL.Workflow

legend =
  Workflow.new()
  |> Workflow.add_start(:start)
  |> Workflow.add_end(:end)
  |> Workflow.add_task(:task)
  |> Workflow.add_decision(:decision)
  |> Workflow.add_fork(:fork)
  |> Workflow.add_join(:join)
  |> Workflow.add_compensation(:compensation, for: :task)
  |> Workflow.add_event(:event)

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(legend)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(legend)),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(legend))
)

Example 1: E-Commerce Saga (The Classic)

A customer places an order. We charge their card, reserve inventory, pack, and ship. If anything fails, we roll back previous steps.

order_saga =
  workflow do
    order_received = start("order_received")

    charge_card =
      task("Charge Card",
        id: :charge_card,
        timeout_ms: 5000,
        retry: 3,
        retry_backoff_ms: 1000
      )

    reserve_inventory =
      task("Reserve Inventory",
        id: :reserve_inventory,
        timeout_ms: 3000
      )

    sufficient_stock = decision("Stock OK?", id: :sufficient_stock)
    pack_items = task("Pack Items", id: :pack_items, timeout_ms: 10_000)
    ship_order = task("Ship Order", id: :ship_order, timeout_ms: 5000)

    refund_payment =
      compensation("Refund Payment",
        id: :refund_payment,
        for: charge_card
      )

    release_inventory =
      compensation("Release Inventory",
        id: :release_inventory,
        for: reserve_inventory
      )

    done = finish("done")

    # Happy path
    order_received ~> charge_card
    charge_card ~> reserve_inventory
    reserve_inventory ~> sufficient_stock
    sufficient_stock ~> pack_items |> condition("yes")
    pack_items ~> ship_order
    ship_order ~> done

    # Compensation paths
    sufficient_stock ~> refund_payment |> condition("no") |> compensation()
    sufficient_stock ~> release_inventory |> condition("no") |> compensation()
    refund_payment ~> done |> compensation()
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(order_saga)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(order_saga), height: "800px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(order_saga))
)

Critical Path Analysis

The critical path is the longest-latency chain from start to end. It tells you the minimum time any order will take, assuming no failures.

{:ok, path, latency_ms} = Analysis.critical_path(order_saga)

IO.puts("Critical path: #{Enum.join(path, " → ")}")
IO.puts("Total latency: #{latency_ms}ms")

Parallelizable Tasks

Which tasks have no dependencies on each other and could run concurrently?

Analysis.parallelizable_tasks(order_saga)
|> Enum.with_index()
|> Enum.each(fn {tasks, level} ->
  IO.puts("Level #{level}: #{Enum.join(tasks, ", ")}")
end)

In this linear saga, every level has exactly one task — nothing can be parallelised. Let's fix that in the next example.

Missing Compensations

Tasks with retry configured should have compensation handlers. Let's check:

Analysis.missing_compensations(order_saga)
|> IO.inspect(label: "Tasks with retry but no compensation")

charge_card has retry: 3 but no direct compensation edge from it — the refund is triggered from the decision node instead. Depending on your design, this may or may not be acceptable. validate/1 will flag it if you prefer every retried task to have a direct compensation route.

Example 2: Parallelised Order Processing

Real systems do independent work in parallel. Charging the card and checking fraud can happen simultaneously. So can packing and sending the confirmation email.

parallel_order =
  workflow do
    order_received = start("order_received")
    fork_checks = fork("Parallel Checks", id: :fork_checks)
    charge_card = task("Charge Card\n5s", id: :charge_card, timeout_ms: 5000)
    fraud_check = task("Fraud Check\n2s", id: :fraud_check, timeout_ms: 2000)
    join_checks = join("Checks Complete", id: :join_checks)
    checks_passed = decision("Passed?", id: :checks_passed)
    fork_fulfillment = fork("Fulfillment", id: :fork_fulfillment)
    pack_items = task("Pack Items\n8s", id: :pack_items, timeout_ms: 8000)
    send_confirmation = task("Send Email\n1s", id: :send_confirmation, timeout_ms: 1000)
    join_fulfillment = join("Fulfillment Done", id: :join_fulfillment)
    ship_order = task("Ship\n3s", id: :ship_order, timeout_ms: 3000)
    done = finish("done")

    refund_payment = compensation("Refund", id: :refund_payment, for: charge_card)
    cancel_fraud_flag = compensation("Clear Flag", id: :cancel_fraud_flag, for: fraud_check)

    # Start → parallel checks
    order_received ~> fork_checks
    fork_checks ~> charge_card
    fork_checks ~> fraud_check
    charge_card ~> join_checks
    fraud_check ~> join_checks
    join_checks ~> checks_passed

    # Decision branches
    checks_passed ~> fork_fulfillment |> condition("yes")
    checks_passed ~> refund_payment |> condition("no") |> compensation()
    checks_passed ~> cancel_fraud_flag |> condition("no") |> compensation()

    # Parallel fulfillment
    fork_fulfillment ~> pack_items
    fork_fulfillment ~> send_confirmation
    pack_items ~> join_fulfillment
    send_confirmation ~> join_fulfillment
    join_fulfillment ~> ship_order
    ship_order ~> done
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(parallel_order)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(parallel_order), height: "900px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(parallel_order))
)

Critical Path (Parallel)

{:ok, path, latency_ms} = Analysis.critical_path(parallel_order)
IO.puts("Critical path: #{Enum.join(path, " → ")}")
IO.puts("Total latency: #{latency_ms}ms")

Notice the total is less than the sum of all timeouts because fork/join pairs only count the slowest branch.

Parallel Levels

Analysis.parallelizable_tasks(parallel_order)
|> Enum.with_index()
|> Enum.each(fn {tasks, level} ->
  IO.puts("Level #{level}: #{Enum.join(tasks, ", ")}")
end)

Now we see charge_card and fraud_check at the same level — they can run simultaneously. Same for pack_items and send_confirmation.

Example 3: CI/CD Pipeline with Swimlanes

Swimlanes group nodes by team or service. Here we model a deployment pipeline with Dev, QA, and Platform teams.

Rendering Options: Flowchart vs. Swimlane

Choreo supports two distinct ways to render swimlanes to Mermaid:

1. Flowchart Syntax (syntax: :flowchart, default): Renders swimlanes as nested subgraphs (subgraph Name ... end) within a standard Mermaid flowchart. Useful for maximum compatibility with older Mermaid versions.
2. Swimlane Syntax (syntax: :swimlane): Renders using the native Mermaid 11.16+ swimlane-beta layout. This produces a dedicated swimlane chart structure where processes are organized cleanly in parallel lanes, highlighting handoffs and responsibility boundaries.

cicd =
  workflow do
    swimlane "Dev Team", id: :dev, fillcolor: "#dbeafe" do
      git_push = event("Git Push", id: :git_push)

      run_tests =
        task("Unit Tests\n3m",
          id: :run_tests,
          timeout_ms: 180_000
        )

      build_image =
        task("Build Docker Image\n5m",
          id: :build_image,
          timeout_ms: 300_000
        )
    end

    swimlane "QA Team", id: :qa, fillcolor: "#dcfce7" do
      deploy_staging =
        task("Deploy to Staging\n2m",
          id: :deploy_staging,
          timeout_ms: 120_000
        )

      run_integration_tests =
        task("Integration Tests\n10m",
          id: :run_integration_tests,
          timeout_ms: 600_000,
          retry: 1
        )

      tests_passed = decision("Tests Pass?", id: :tests_passed)
    end

    swimlane "Platform Team", id: :platform, fillcolor: "#fef3c7" do
      deploy_production =
        task("Deploy to Prod\n3m",
          id: :deploy_production,
          timeout_ms: 180_000
        )

      run_smoke_tests =
        task("Smoke Tests\n2m",
          id: :run_smoke_tests,
          timeout_ms: 120_000
        )

      rollback_production =
        compensation("Rollback Prod",
          id: :rollback_production,
          for: deploy_production
        )
    end

    live = finish("live")

    # Flow
    git_push ~> run_tests
    run_tests ~> build_image
    build_image ~> deploy_staging
    deploy_staging ~> run_integration_tests
    run_integration_tests ~> tests_passed
    tests_passed ~> deploy_production |> condition("yes")
    tests_passed ~> rollback_production |> condition("no") |> compensation()
    deploy_production ~> run_smoke_tests
    run_smoke_tests ~> live
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(cicd, syntax: :swimlane), height: "900px"),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(cicd), height: "900px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(cicd))
)

Bottleneck Detection

Analysis.bottlenecks(cicd, latency_threshold: 300_000)
|> IO.inspect(label: "High-latency bottlenecks (> 5m)")

Analysis.bottlenecks(cicd, retry_threshold: 1)
|> IO.inspect(label: "High-retry bottlenecks")

Simulation

Analysis.simulate(cicd)
|> Enum.sort_by(fn {_id, vals} -> vals.cumulative_latency end, :desc)
|> Enum.take(5)
|> Enum.each(fn {id, vals} ->
  total = div(vals.cumulative_latency, 1000)
  IO.puts("#{id}: #{total}s total (task: #{div(vals.task_latency, 1000)}s)")
end)

Example 4: Approval Workflow with Retry and Timeout

Human-in-the-loop processes need timeout handling. If a manager doesn't approve within 24 hours, escalate.

approval =
  workflow do
    request_submitted = start("request_submitted")

    validate_request =
      task("Auto-Validate\n500ms",
        id: :validate_request,
        timeout_ms: 500
      )

    valid = decision("Valid?", id: :valid)

    manager_review =
      task("Manager Review\n24h",
        id: :manager_review,
        timeout_ms: 86_400_000
      )

    approved = decision("Approved?", id: :approved)

    process_request =
      task("Process\n2s",
        id: :process_request,
        timeout_ms: 2000
      )

    completed = finish("completed")

    auto_escalate =
      task("Escalate to VP\n48h",
        id: :auto_escalate,
        timeout_ms: 172_800_000
      )

    notify_rejection =
      task("Notify Rejection\n1s",
        id: :notify_rejection,
        timeout_ms: 1000
      )

    rejected = finish("rejected")
    escalated = finish("escalated")

    # Happy / sad paths
    request_submitted ~> validate_request
    validate_request ~> valid
    valid ~> manager_review |> condition("yes")
    valid ~> notify_rejection |> condition("no")
    notify_rejection ~> rejected
    manager_review ~> approved
    approved ~> process_request |> condition("yes")
    approved ~> notify_rejection |> condition("no")
    process_request ~> completed

    # Timeout path
    manager_review ~> auto_escalate |> timeout()
    auto_escalate ~> escalated
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(approval)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(approval), height: "900px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(approval))
)

Failure Scenarios

Which tasks have explicit compensation or failure paths?

Analysis.validate(approval)

Uncompensated Paths

Which error paths don't lead to a terminal node via compensations?

Analysis.uncompensated_paths(approval)
|> IO.inspect(label: "Uncompensated error paths")

Example 5: Fixing a Broken Workflow

Start with a workflow that has structural problems and use validate/1 as a checklist.

broken =
  workflow do
    step_a = task("Step A", id: :step_a)
    step_b = task("Step B", id: :step_b)
    orphan_step = task("Orphan", id: :orphan_step)
    dead_end = task("Dead End", id: :dead_end)
    finish = finish("finish")
    rollback = compensation("Rollback", id: :rollback, for: step_a)

    step_a ~> step_b
    step_b ~> finish
    # orphan_step has no inputs
    # dead_end has no outputs
    # rollback has no incoming edge
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(broken, syntax: :swimlane)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(broken), height: "900px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(broken))
)

Diagnosis

Analysis.validate(broken)
|> Enum.each(fn {sev, msg} ->
  icon = if sev == :error, do: "❌", else: "⚠️"
  IO.puts("#{icon} #{msg}")
end)

The validator catches:

• No start node
• No end node connected from dead_end
• Orphan tasks
• Dead-end tasks
• Unreachable compensation node

The Fix

fixed =
  workflow do
    start = start("start")
    step_a = task("Step A", id: :step_a)
    step_b = task("Step B", id: :step_b)
    finish = finish("finish")
    rollback = compensation("Rollback", id: :rollback, for: step_a)

    start ~> step_a
    step_a ~> step_b
    step_b ~> finish
    step_a ~> rollback |> compensation()
  end

IO.puts("After fix:")
Analysis.validate(fixed) |> IO.inspect()

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(fixed)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(fixed), height: "600px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(fixed))
)

Advanced: Custom Theming

brand_theme =
  Choreo.Theme.custom(
    colors: %{
      start: "#10b981",
      end: "#ef4444",
      task: "#3b82f6",
      decision: "#8b5cf6",
      compensation: "#f87171"
    },
    node_fontcolor: "white",
    edge_color: "#94a3b8",
    graph_bgcolor: "#0f172a"
  )


Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(parallel_order, theme: brand_theme)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(parallel_order, theme: brand_theme), height: "900px"),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(parallel_order, theme: brand_theme))
)

Example 6: Multigraph (Parallel Transitions)

Sometimes tasks transition between states along multiple paths (e.g., synchronous data payload and asynchronous notification).

multi_flow =
  workflow do
    start = start("start")
    process = task("process")
    done = finish("done")

    start ~> process
    process ~> done |> on("Payload") |> weight(1000)
    process ~> done |> on("Event") |> weight(200)
  end

IO.puts("Number of edges: #{length(Workflow.edges(multi_flow))}")

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Workflow.to_mermaid(multi_flow)),
  Graphviz: Kino.VizJS.render(Workflow.to_dot(multi_flow)),
  Sketch: Choreo.Lab.Sketch.new(Workflow.to_mermaid(multi_flow))
)

Cheat Sheet

Lab DSL Syntax

Programmatic Pipe API & Analysis

Summary

Workflow diagrams as code mean your business processes are version-controlled, reviewable, and auditable. Every time you add a task, change a timeout, or introduce a new decision branch, you can immediately see the impact on critical path latency, parallelisation opportunities, and failure coverage — before you ship to production.