Reactor Saga Orchestrator

Reactor composition allows you to break down complex workflows into smaller, focused, and reusable sub-reactors. Instead of building a single monolithic reactor, you use the compose macro to integrate other reactors into a parent orchestrator.

Benefits of Composition

• Complexity Management: Break large workflows into manageable pieces.
• Reusability: A single sub-reactor (e.g., UserManagementReactor) can be used in multiple different parent workflows.
• Testability: Each sub-reactor can be tested in isolation before being integrated into a larger system.
• Error Boundaries: Failures in a sub-reactor can be managed by the parent reactor.

The compose macro

When composing, you define a name for the composed step and the module of the sub-reactor. You can also provide a block to map inputs from the parent reactor to the sub-reactor's arguments.

# Basic composition
compose :step_name, SubReactorModule

# Composition with argument mapping
compose :step_name, SubReactorModule do
  argument :sub_reactor_input, input(:parent_input)
end

defmodule OrderProcessor do
  use Reactor
  
  compose :user_management, UserManagementReactor
  compose :inventory_check, InventoryReactor  
  compose :payment_processing, PaymentReactor
  compose :fulfillment, FulfillmentReactor
end

A Reactor is defined by using the Reactor module in an Elixir module. It consists of four main components:

1. Inputs: Defined using input :name, these are the parameters the reactor accepts (similar to function arguments).
2. Steps: Units of work defined using step :name. Each step contains:
   • Arguments: Defined using argument :name, source. The source can be an input(:name) or a result(:step_name) from a previous step.
   • Run function: A run fn %{args}, _context -> ... end block that performs the work. It should return {:ok, value} or {:error, reason}.
3. Dependencies: Automatically created by the argument definitions. If a step uses result(:other_step), it will wait for :other_step to complete.
4. Return: Defined using return :step_name, specifying which step's result should be the final output of the reactor.

defmodule UserRegistration do
  use Reactor

  input :email
  input :password

  step :validate_email do
    argument :email, input(:email)
    
    run fn %{email: email}, _context ->
      if String.contains?(email, "@") do
        {:ok, email}
      else
        {:error, "Email must contain @"}
      end
    end
  end

  step :hash_password do
    argument :password, input(:password)
    
    run fn %{password: password}, _context ->
      hashed = :crypto.hash(:sha256, password) |> Base.encode16()
      {:ok, hashed}
    end
  end

  step :create_user do
    argument :email, result(:validate_email)
    argument :password_hash, result(:hash_password)
    
    run fn %{email: email, password_hash: password_hash}, _context ->
      user = %{id: :rand.uniform(10000), email: email, password_hash: password_hash}
      {:ok, user}
    end
  end

  return :create_user
end

Reactor determines execution order using argument dependencies rather than explicit sequencing. By declaring a dependency on a previous step's result using result(:step_name), Reactor automatically builds an execution graph.

Key Benefits:

• Automatic Concurrency: Steps that do not share dependencies run simultaneously.
• Natural Data Flow: Dependencies express the actual movement of data through the workflow.
• Dynamic Workflows: Steps can emit new steps that resolve dependencies automatically.

Mental Model: Instead of thinking Step A -> Step B, think Step B requires the output of Step A.

defmodule UserWorkflow do
  use Reactor
  
  step :fetch_user do
    argument :user_id, input(:user_id)
  end
  
  step :fetch_profile do
    argument :user_id, input(:user_id)  # Can run concurrently with fetch_user
  end
  
  step :merge_data do
    argument :user, result(:fetch_user)      # Waits for fetch_user
    argument :profile, result(:fetch_profile) # Waits for fetch_profile
  end
end

Reactor supports two types of error recovery: Compensation and Undo.

• Compensation: A step can return {:continue, result} during a failure to allow the reactor to proceed with a fallback value. Use stub(:compensate, ...) to test this.
• Undo: If a step fails and no compensation is provided, Reactor may trigger the .undo/4 logic for previously completed steps. Use expect(:undo, ...) to verify that cleanup logic is executed.

# Testing compensation returning a continue value
test "compensation returns continue value" do
  MyApp.Steps.ProcessPayment
  |> stub(:run, fn _args, _context, _opts -> 
    {:error, :temporary_failure}
  end)
  |> stub(:compensate, fn _reason, _args, _context, _opts ->
    {:continue, %{id: "fallback_payment", status: :manual_review}}
  end)
  
  # ... rest of test
end

# Testing undo logic
test "later step failure triggers undo of earlier steps" do
  MyApp.Steps.ProcessPayment
  |> stub(:run, fn _args, _context, _opts -> 
    {:ok, %{id: "payment_123"}}
  end)
  |> expect(:undo, fn _result, _args, _context, _opts ->
    :ok
  end)
  
  # ... rest of test
end

Reactor uses two distinct mechanisms to handle failures, depending on when the error occurs in the workflow:

Compensation

Trigger: Occurs when a step fails during its run() execution. Purpose: To decide how the reactor should react to a specific error in the current step (e.g., retrying or failing). Return values for compensate/4:

• :retry: Attempt the step again.
• {:continue, value}: Continue the workflow using the provided value instead of the failed result.
• :ok: Successfully handled the error, but the reactor will still proceed to trigger rollback (undo) for previous steps.
• {:error, reason}: Fail the entire reactor immediately.

Undo

Trigger: Occurs when a step succeeded, but a subsequent step in the workflow failed. Purpose: To roll back the side effects of the successful step to maintain system consistency. Return values for undo/4:

• :ok: Successfully rolled back the changes.
• {:error, reason}: Failed to undo the changes. This is a critical failure that will cause the entire reactor to fail.

Error Handling Flow

1. Step A succeeds.
2. Step B succeeds.
3. Step C fails.
4. Reactor begins compensation for Step B.
5. Reactor begins compensation for Step A.
6. Reactor returns the compensated error.

sequenceDiagram
    participant Reactor
    participant StepA
    participant StepB
    participant StepC
    
    Reactor->>StepA: run()
    StepA-->>Reactor: {:ok, result}
    
    Reactor->>StepB: run()
    StepB-->>Reactor: {:ok, result}
    
    Reactor->>StepC: run()
    StepC-->>Reactor: {:error, reason}
    
    Note over Reactor: Begin compensation
    Reactor->>StepB: compensate()
    StepB-->>Reactor: {:continue, context}
    
    Reactor->>StepA: compensate()
    StepA-->>Reactor: {:continue, context}
    
    Reactor-->>Reactor: Return compensated error

The Reactor struct is the central data structure containing all information required for workflow execution. It is primarily composed of the following fields:

• context: Runtime context shared across all steps.
• id: Unique identifier for the reactor.
• input_descriptions: Metadata about expected inputs.
• inputs: The actual input data provided.
• intermediate_results: A map of step outputs used for dependency resolution.
• middleware: A stack of middleware for cross-cutting concerns.
• plan: The dependency graph (DAG) constructed by the planner.
• return: The final value returned by the reactor.
• state: The current lifecycle state (:pending, :executing, :halted, :failed, or :successful).
• steps: The list of defined steps.
• undo: An undo stack tracking successful undoable steps for rollback (LIFO).

defstruct context: %{},
          id: nil,
          input_descriptions: %{},
          inputs: [],
          intermediate_results: %{},
          middleware: [],
          plan: nil,
          return: nil,
          state: :pending,
          steps: [],
          undo: []