Oi (oi v0.7.0)

Copy Markdown

Oi provides a simple API for Orchid and ready-to-use functions for bind OrchidSymbiont, OrchidStratum and OrchidIntervention.

The purpose is building a front-end-friendly interface(exclipit nodes&edges).

Core Concepts

Oi pipeline

There're three phases in Oi's lifecycle:

  1. Build a Oi.Topology.Graph of step nodes and edges
  2. Oi.compile/2(topology level) generate static Oi.Compiled struct (bundles + plan, reusable)
  3. Oi.execute/2 can resolve data, apply orchid_adapters, run via pluggable executor

Compiled

immutable compilation product. Compile once, dispatch many times with different data / executors.

See Oi.Compiled.

Data format during Oi.execute/2

Oi provides unified data: replaces the separate inputs: / interventions:.

  • Ports with no incoming edge → memory (external inputs).
  • Ports with an incoming edge → interventions (data originates inside the graph, user is overriding it). Intervention values use the format {type, value} where type is an atom naming the intervention strategy (:override, :offset, or a custom module like MyApp.MergeStrategy). Plain values default to {:override, value}.

When the payload is a tuple, wrap it explicitly in %Orchid.Param{} to preserve type information:

{:override, %Orchid.Param{name: :key, type: {:tuple, 2}, payload: {1, 2}}}

Two formats supported:

# Nested (recommended)
data: %{step1: %{in: "foo"}, step2: %{result: {:override, "bar"}}}

# Tuple keys
data: %{{:step1, :in} => "foo", {:step2, :result} => {:override, "bar"}}

Pluggable Executor

Pluggable task fan-out per stage. Built-in: Oi.Executor.Sync (serial), Oi.Executor.TaskSup (Task.Supervisor within per session).

Custom executors implement the Oi.Executor behaviour.

Session / Multiple Oi instances

Optional multi-tenant isolation via Oi.Runtime.Session, wrapping per-tenant OrchidSymbiont.Runtime + Task.Supervisor.

Orchid Extension

Oi intentionally does not implement its own workflow plugin system. Use Orchid hooks through :orchid_opts for step-level concerns such as logging, metrics, retry, and caching.

Oi also support some ready-to-use functions to load/modify orchid's recipe and options via Oi.Adapters.

Errors

All public functions return either {:ok, result} or {:error, reason}. Errors are tagged tuples keyed by phase:

Graph phase (build + validate)

  • {:error, {:self_loop, node_id}} -- add_edge/2, node points to itself
  • {:error, [{:orphan_edge, ...}]} -- validate/1, edge refs nonexistent node
  • {:error, [{:port_not_found, ...}]} -- validate/1, port not in inputs/outputs
  • {:error, [{:cycle_detected, [node_id]}]} -- validate/1, nodes in a cycle

Dispatch phase (execute)

  • {:error, :invalid_data_format} -- malformed :data input
  • {:error, {:missing_input, io_key}} -- required upstream port not in memory
  • {:error, {:bad_executor_return, mod, val}} -- executor returned unexpected shape

Orchid phase (transparent pass-through)

  • {:error, {:orchid_error, recipe_name, %Orchid.Error{}}}

Validation errors accumulate -- the :error tuple wraps a list so the caller sees all problems at once, not just the first one found.

Summary

Types

name()

Oi's name, split several scopes.

Functions

compile(graph, cluster \\ %Oi.Topology.Cluster{})

Compile graph into static bundles + plan.

execute(compiled, opts \\ [])

Execute compiled plan with inputs and interventions.

run(graph_or_compiled, opts \\ [])

All-in-one convenience: compile + execute in a single call.

Types

name()

@type name() :: String.t()

Oi's name, split several scopes.

Functions

compile(graph, cluster \\ %Oi.Topology.Cluster{})

@spec compile(Oi.Topology.Graph.t(), Oi.Topology.Cluster.t()) ::
  {:ok, Oi.Compiled.t()} | {:error, term()}

Compile graph into static bundles + plan.

Pure topology — no interventions. Reusable across different intervention sets and inputs.

execute(compiled, opts \\ [])

@spec execute(
  Oi.Compiled.t(),
  keyword()
) :: {:ok, Oi.Result.t()} | {:error, term()}

Execute compiled plan with inputs and interventions.

Oi.execute(compiled, data: %{
  step1: %{in: "foo"},
  step2: %{result: {:override, "bar"}}
})

Supports tuple-key format as well: %{{:step, :port} => value}.

Ports with incoming edges → intervention; ports without → memory. Intervention values use {type, value} where type is an atom naming the intervention strategy (:override, :offset, or a custom module). Plain values default to {:override, value}. For tuple payloads, wrap in %Orchid.Param{} to preserve type info.

Other options

  • :executor — Oi.Executor.Sync (default), Oi.Executor.TaskSup, or custom module implementing Oi.Executor
  • :executor_opts — passed to the executor (e.g. [sup: MyTaskSup])
  • :orchid_adapters — adapter pipeline; each is a 1-arity or 2-arity function over {recipe, opts}
  • :orchid_baggage — merged into Orchid run baggage
  • :orchid_opts — extra keyword opts forwarded to Orchid.run/3
  • :concurrency — fallback for executor if :executor_opts has none (default: System.schedulers_online())
  • :timeout — fallback for executor (default: :infinity)
  • :name — optional scope name, merged into baggage as :scope_id

run(graph_or_compiled, opts \\ [])

@spec run(
  Oi.Compiled.t() | Oi.Topology.Graph.t(),
  keyword()
) :: {:ok, Oi.Result.t()} | {:error, term()}

All-in-one convenience: compile + execute in a single call.

Accepts either a Graph.t() or a Compiled.t(). When given a graph, extracts :cluster from opts (defaults to empty %Cluster{}), compiles, then executes with the remaining options. When given a compiled struct, delegates directly to execute/2.

Options

Same as execute/2, plus:

  • :cluster — Oi.Topology.Cluster.t() (default: %Cluster{}), only used when graph_or_compiled is a Graph.t()

Examples

graph = graph do
  step(MyStep)
end

{:ok, result} = Oi.run(graph, data: %{my_step: %{in: "hello"}})

# With explicit cluster
{:ok, result} = Oi.run(graph,
  data: %{my_step: %{in: "hello"}},
  cluster: %Cluster{node_colors: %{my_step: :fast}}
)