Durable Workflow Engine

call_workflow/3 can be used inside parallel blocks. In this specific context, child workflows are executed inline (synchronously) within the parallel task rather than being sent to the queue and suspending the parent.

Behavior:

• The parent's state is saved before the child runs and restored after it completes.
• Results are available immediately in the data[:__results__] map after the parallel block finishes.

Critical Limitation: Child workflows used inside a parallel block cannot use waits (such as sleep, wait_for_event, etc.). Because they are running inline, they cannot suspend the execution thread. Attempting to use a waiting child inside a parallel block will result in an error.

workflow "enrich_order" do
  step :init, fn input ->
    {:ok, %{order_id: input["order_id"]}}
  end

  parallel on_error: :complete_all do
    step :enrich_customer, fn data ->
      case call_workflow(MyApp.CustomerLookup, %{"id" => data.order_id}, ref: :customer) do
        {:ok, result} -> {:ok, assign(data, :customer, result)}
        {:error, reason} -> {:error, reason}
      end
    end

    step :enrich_inventory, fn data ->
      case call_workflow(MyApp.InventoryCheck, %{"id" => data.order_id}, ref: :inventory) do
        {:ok, result} -> {:ok, assign(data, :inventory, result)}
        {:error, reason} -> {:error, reason}
      end
    end
  end

  step :process, fn data ->
    results = data[:__results__]
    # Handle results from parallel call_workflow steps...
    {:ok, data}
  end
end

Workflows are defined by modules that use Durable. You use the workflow "name" do ... end block to encapsulate a sequence of steps.

Steps

Steps are the atomic units of work within a workflow. They are defined using step :name, [options], fn ctx -> ... end.

• Context (ctx): Each step receives a context object containing accumulated data. Use assign(ctx, :key, value) to persist new data into the workflow state.
• Retries: You can configure automatic retries for steps using the retry option, specifying max_attempts and a backoff strategy (e.g., :exponential).
• Error Handling: Raising an error within a step (e.g., when an AI response is invalid) will trigger the configured retry logic.

Branching

Use branch on: fn ctx -> ... end do ... end to implement conditional logic.

• Only one path in a branch executes based on the value returned by the function.
• Steps defined after a branch block will execute regardless of which branch was taken, once that branch completes.

Human-in-the-Loop

Use Durable.Wait to pause a workflow for external input. The wait_for_input/2 function allows you to wait for a specific input type with a defined timeout.

defmodule MyApp.DocumentProcessor do
  use Durable
  use Durable.Helpers

  workflow "process_document" do
    step :fetch, fn ctx ->
      doc = DocumentStore.get(ctx["doc_id"])
      {:ok, %{doc: doc}}
    end

    step :classify, [retry: [max_attempts: 3, backoff: :exponential]], fn ctx ->
      # ... logic ...
      {:ok, assign(ctx, :doc_type, doc_type)}
    end

    branch on: fn ctx -> ctx.doc_type end do
      :invoice ->
        step :extract_invoice, [retry: [max_attempts: 3]], fn ctx ->
          # ... logic ...
          {:ok, assign(ctx, :extracted, extracted)}
        end
      _ ->
        step :flag_for_review, fn ctx ->
          {:ok, assign(ctx, :needs_review, true)}
        end
    end

    step :store, fn ctx ->
      # ... logic ...
      {:ok, ctx}
    end
  end
end

# Start the workflow
{:ok, workflow_id} = Durable.start(MyApp.DocumentProcessor, %{"doc_id" => "doc_123"})

When a parallel block completes, the results of each step are collected into a map under the __results__ key in the context. Results are stored as tagged tuples: {:ok, data} or {:error, reason}.

You can access these results in two ways:

1. Using Helper Functions

If you are inside a step, use these helpers to interact with the results:

• parallel_results(): Returns the entire map of results.
• parallel_result(step_name): Returns the tagged tuple for a specific step.
• parallel_ok?(step_name): Returns true if the specific step succeeded.

2. Direct Context Access

Access the map directly via ctx[:__results__][step_name].

step :handle_results, fn ctx ->
  # Using helpers
  results = parallel_results()
  case parallel_result(:payment) do
    {:ok, payment} -> # handle success
    {:error, reason} -> # handle error
  end

  # Direct access
  case ctx[:__results__][:payment] do
    {:ok, payment} -> {:ok, Map.put(ctx, :payment_id, payment.id)}
    {:error, _} -> {:goto, :handle_payment_failure, ctx}
  end
end