Raxol.Workflow.Graph (Raxol v2.6.0)

View Source

Immutable graph builder for Raxol.Workflow.

Build a graph by piping add_node/3, add_edge/3, add_guarded_edge/4, add_conditional_edge/4 (and the upcoming add_join/4 / add_channel/4) onto a new/1 seed, then freeze with compile/2.

Example

alias Raxol.Workflow.Graph

{:ok, compiled} =
  Graph.new(:summarizer)
  |> Graph.add_node(:fetch, fn s -> {:ok, Map.put(s, :data, "x")} end)
  |> Graph.add_node(:render, fn s -> {:ok, Map.put(s, :rendered, true)} end)
  |> Graph.add_edge(:__start__, :fetch)
  |> Graph.add_edge(:fetch, :render)
  |> Graph.add_edge(:render, :__end__)
  |> Graph.compile()

Reserved node ids

  • :__start__ -- virtual entry node, present in every graph
  • :__end__ -- virtual exit node, present in every graph

Authors must add at least one edge from :__start__ and at least one edge into :__end__. compile/2 enforces these rules along with reachability and absence-of-orphans.

Summary

Types

t()
validation_error()

Functions

add_channel(graph, name, opts)

Declare a typed reducer for a state-map key.

add_conditional_edge(graph, from, candidates, chooser)

Add a conditional fan-out edge from from. The chooser receives the state and returns the next node id (or a list of ids for parallel fan-out). candidates declares the set of possible destinations so compile/2 can validate them.

add_edge(graph, from, to)

Add a static edge from from to to.

add_guarded_edge(graph, from, to, guard)

Add a guarded edge from from to to. The guard receives the workflow state; the edge is taken only when the guard returns truthy.

add_join(graph, target, upstream, opts \\ [])

Mark target as a join (barrier) node fed by the listed upstream branches.

add_node(graph, id, body)

Add a node to the graph. The body argument selects the node shape

add_node(graph, id, fun, compensate_fun)

Add a FunctionNode with an optional compensation function.

compile(graph, opts \\ [])

Freeze the graph for execution.

new(id)

Construct a new, empty graph with the given id.

Types

t()

@type t() :: %Raxol.Workflow.Graph{
  channels: %{required(atom() | binary()) => Raxol.Workflow.Channel.t()},
  edges: [Raxol.Workflow.Edge.t()],
  id: atom() | binary(),
  nodes: %{required(Raxol.Workflow.Node.id()) => Raxol.Workflow.Node.t()}
}

validation_error()

@type validation_error() ::
  {:missing_start_edge, atom()}
  | {:missing_end_edge, atom()}
  | {:unknown_node, [Raxol.Workflow.Node.id()]}
  | {:orphan_nodes, [Raxol.Workflow.Node.id()]}
  | {:unreachable_from_start, [Raxol.Workflow.Node.id()]}
  | {:cannot_reach_end, [Raxol.Workflow.Node.id()]}
  | {:join_target_unknown, Raxol.Workflow.Node.id()}
  | {:join_upstream_unknown, Raxol.Workflow.Node.id(),
     [Raxol.Workflow.Node.id()]}
  | {:join_upstream_shared, [Raxol.Workflow.Node.id()]}
  | {:join_target_unknown, Raxol.Workflow.Node.id()}
  | {:join_upstream_unknown, Raxol.Workflow.Node.id(),
     [Raxol.Workflow.Node.id()]}
  | {:join_upstream_shared, Raxol.Workflow.Node.id(),
     [Raxol.Workflow.Node.id()]}

Functions

add_channel(graph, name, opts)

@spec add_channel(t(), atom() | binary(), keyword()) :: t()

Declare a typed reducer for a state-map key.

Channels are metadata only -- no nodes or edges are added. At a join, the runtime applies each channel's with reducer pairwise across all branch contributions to the channel's into key.

Options:

  • :into -- the state-map key the reducer writes to. Required.
  • :with -- a 2-arity reducer (left, right) -> merged. Required.

Channel names must be unique within a graph.

add_conditional_edge(graph, from, candidates, chooser)

@spec add_conditional_edge(
  t(),
  Raxol.Workflow.Node.id(),
  [Raxol.Workflow.Node.id()],
  (any() -> any())
) ::
  t()

Add a conditional fan-out edge from from. The chooser receives the state and returns the next node id (or a list of ids for parallel fan-out). candidates declares the set of possible destinations so compile/2 can validate them.

add_edge(graph, from, to)

@spec add_edge(t(), Raxol.Workflow.Node.id(), Raxol.Workflow.Node.id()) :: t()

Add a static edge from from to to.

add_guarded_edge(graph, from, to, guard)

@spec add_guarded_edge(
  t(),
  Raxol.Workflow.Node.id(),
  Raxol.Workflow.Node.id(),
  (any() -> any())
) :: t()

Add a guarded edge from from to to. The guard receives the workflow state; the edge is taken only when the guard returns truthy.

add_join(graph, target, upstream, opts \\ [])

@spec add_join(t(), Raxol.Workflow.Node.id(), [Raxol.Workflow.Node.id()], keyword()) ::
  t()

Mark target as a join (barrier) node fed by the listed upstream branches.

The runtime holds at the join until every node id in upstream has committed a result on the same run. It then merges each branch's terminal state (per any declared add_channel/3 reducer; last-write-wins by branch order for keys without a channel), runs the target node's body once with the merged state, and continues from target's outgoing edges.

Options:

  • :reduce -- explicit (list_of_branch_states -> merged_state) that overrides per-channel + last-write-wins. Default: nil.
  • :timeout_ms -- max wall-clock to wait for all branches before failing with {:branch_timeout, missing}. Default: inherits the run's :run_timeout_ms.
  • :parallelism -- branch execution concurrency. :branches (default) runs every branch concurrently via Task.async_stream. A positive integer caps max_concurrency (1 selects the serial path, useful when consumers need branch-index-order side effects).

compile/2 validates that target and every upstream id refer to declared nodes.

add_node(graph, id, body)

@spec add_node(t(), Raxol.Workflow.Node.id(), term()) :: t()

Add a node to the graph. The body argument selects the node shape:

  • A 1-arity function builds a FunctionNode.
  • A {module, opts} tuple builds a BehaviourNode.
  • A struct builds a TypedNode.

add_node(graph, id, fun, compensate_fun)

@spec add_node(t(), Raxol.Workflow.Node.id(), (any() -> any()), (any() ->
                                                             {:ok, any()}
                                                             | {:error, any()})) ::
  t()

Add a FunctionNode with an optional compensation function.

Convenience for add_node/3 |> Node.function/3. The compensate_fun runs in reverse order under failure_policy: :compensate when the run errors after this node succeeded. It receives the current state and returns {:ok, new_state} | {:error, reason}.

compile(graph, opts \\ [])

@spec compile(
  t(),
  keyword()
) :: {:ok, Raxol.Workflow.Compiled.t()} | {:error, validation_error()}

Freeze the graph for execution.

Runs the full validation pipeline (no orphan nodes, every node reachable from :__start__, every node has a path to :__end__, conditional candidates exist in the graph) and returns a Raxol.Workflow.Compiled struct or a structured error.

Options

  • :failure_policy -- one of :retry, :halt, :compensate (default :halt)
  • :max_attempts -- when :failure_policy is :retry, the maximum number of attempts per node (default 3). Includes the initial attempt; 1 disables retries.
  • :retry_backoff_ms -- base backoff delay between retries; each subsequent retry doubles (default 100).
  • :step_timeout_ms -- per-node timeout (default 60_000)
  • :run_timeout_ms -- total run timeout (default 3_600_000)
  • :saver -- Raxol.Workflow.Checkpoint.Saver module (defaults to Ets)

These options are stored on Compiled and consumed by Raxol.Workflow.Runtime.

new(id)

@spec new(atom() | binary()) :: t()

Construct a new, empty graph with the given id.