Oi.Dispatch.Config (oi v0.8.0)

Copy Markdown

Immutable dispatch configuration.

Built once at the dispatch boundary, threaded to Worker and the plugin chain. Controls executor selection, concurrency, timeout, and Orchid plugin pipeline.

Options

  • :executor — module implementing Oi.Executor (default: Oi.Executor.Sync)
  • :executor_opts — keyword opts passed to executor's run/3
  • :orchid_adapters — ordered list of adapters; each is a 1-arity fn {recipe, opts} -> {recipe, opts} or 2-arity fn {recipe, opts}, conf -> {recipe, opts}
  • :orchid_baggage — map merged into every Orchid run's 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 if :executor_opts has none (default: :infinity)
  • :name — optional scope name, merged into baggage as :scope_id
  • :checkpoint — optional function called before each stage; see "Checkpoint" below

Checkpoint

A checkpoint is a function fn event, drafting -> :cont | :halt end invoked by the orchestrator before each stage runs. event is a map with :stage_index, :stage_count, :clusters (cluster names of the stage's bundles) and :node_ids. The passed drafting's memory holds everything produced so far — inspect it, then return :cont to run the stage or :halt to stop the whole dispatch with the current memory as a partial result.

Summary

Types

checkpoint()
checkpoint_action()
checkpoint_event()

Event map passed to a :checkpoint function before a stage runs.

data()

Unified user-facing data for Oi.execute/2.

t()

Functions

apply_orchid_adapters(conf, orchid_tuple)

Run every plugin in order over the {recipe, run_opts} tuple. Each plugin may rewrite the recipe or append to run_opts.

assemble_run_opts(conf, drafting)

Assemble keyword opts for Orchid.run/3. Delegates to Options.assemble_run_opts/3.

build_drafting(data, compiled)

Build a Drafting from user :data and compiled graph topology. Delegates to Options.build_drafting_inputs/2.

new(opts \\ [])

Types

checkpoint()

@type checkpoint() :: (checkpoint_event(), Oi.Dispatch.Drafting.t() ->
                   checkpoint_action())

checkpoint_action()

@type checkpoint_action() :: :cont | :halt

checkpoint_event()

@type checkpoint_event() :: %{
  stage_index: non_neg_integer(),
  stage_count: non_neg_integer(),
  clusters: [Oi.Topology.Cluster.cluster_name()],
  node_ids: [Oi.Topology.Graph.Node.id()]
}

Event map passed to a :checkpoint function before a stage runs.

data()

@type data() :: map()

Unified user-facing data for Oi.execute/2.

See Oi.Dispatch module docs for the full data format specification. Two shapes supported:

  • Nested: %{step: %{port: value}}
  • Tuple keys: %{{:step, :port} => value}

t()

@type t() :: %Oi.Dispatch.Config{
  checkpoint: checkpoint() | nil,
  concurrency: pos_integer(),
  executor: module(),
  executor_opts: keyword(),
  name: Oi.name() | nil,
  orchid_adapters: [
    ({Orchid.Recipe.t(), keyword()}, t() -> {Orchid.Recipe.t(), keyword()})
    | ({Orchid.Recipe.t(), keyword()} -> {Orchid.Recipe.t(), keyword()})
  ],
  orchid_baggage: map(),
  orchid_opts: keyword(),
  timeout: timeout()
}

Functions

apply_orchid_adapters(conf, orchid_tuple)

@spec apply_orchid_adapters(
  t(),
  {Orchid.Recipe.t(), keyword()}
) :: {Orchid.Recipe.t(), keyword()}

Run every plugin in order over the {recipe, run_opts} tuple. Each plugin may rewrite the recipe or append to run_opts.

assemble_run_opts(conf, drafting)

@spec assemble_run_opts(t(), Oi.Dispatch.Drafting.t()) :: keyword()

Assemble keyword opts for Orchid.run/3. Delegates to Options.assemble_run_opts/3.

build_drafting(data, compiled)

@spec build_drafting(data(), Oi.Compiled.t()) ::
  {:ok, Oi.Dispatch.Drafting.t()} | {:error, term()}

Build a Drafting from user :data and compiled graph topology. Delegates to Options.build_drafting_inputs/2.

new(opts \\ [])

@spec new(keyword()) :: t()