aion/workflow

Workflow authoring surface.

This module is an aggregator only: it forwards declarations from the workflow submodules and contains no workflow business logic. run is the only recorded activity dispatch surface in this brief; deterministic time and random values come from AD through aion/internal/ffi.

Types

ChildHandle

</>

pub type ChildHandle(output, workflow_error) =
  child.ChildHandle(output, workflow_error)

SignalRef

</>

pub type SignalRef(payload) =
  signal.SignalRef(payload)

TimerRef

</>

pub type TimerRef =
  timer.TimerRef

Timestamp

</>

pub type Timestamp =
  run.Timestamp

WorkflowDefinition

</>

pub type WorkflowDefinition(input, output, workflow_error) =
  define.WorkflowDefinition(input, output, workflow_error)

Values

all

</>

pub fn all(
  activities: List(activity.Activity(input, output)),
) -> Result(List(output), error.ActivityError)

all_settled

</>

pub fn all_settled(
  activities: List(activity.Activity(input, output)),
) -> List(Result(output, error.ActivityError))

Spawn all activities concurrently and settle every member independently: one Result slot per activity, input order, no fail-fast, no sibling cancellation. Each member’s retry policy runs to its own final outcome.

all_settled_with_default

</>

pub fn all_settled_with_default(
  activities: List(activity.Activity(input, output)),
  workflow_default_task_queue: option.Option(String),
) -> List(Result(output, error.ActivityError))

all_settled, supplying the workflow-level default task queue for any member that selects none. See [run_with_default] for the precedence.

all_with_default

</>

pub fn all_with_default(
  activities: List(activity.Activity(input, output)),
  workflow_default_task_queue: option.Option(String),
) -> Result(List(output), error.ActivityError)

all, supplying the workflow-level default task queue for any member that selects none. See [run_with_default] for the resolution precedence.

cancel_timer

</>

pub fn cancel_timer(
  reference: timer.TimerRef,
) -> Result(Nil, error.EngineError)

continue_as_new

</>

pub fn continue_as_new(
  input: a,
  input_codec: codec.Codec(a),
) -> Result(Nil, error.WorkflowError)

define

</>

pub fn define(
  name: String,
  input_codec: codec.Codec(input),
  output_codec: codec.Codec(output),
  error_codec: codec.Codec(workflow_error),
  entry_fn: fn(input) -> Result(output, workflow_error),
) -> define.WorkflowDefinition(input, output, workflow_error)

entry_fn

</>

pub fn entry_fn(
  definition: define.WorkflowDefinition(
    input,
    output,
    workflow_error,
  ),
) -> fn(input) -> Result(output, workflow_error)

entrypoint

</>

pub fn entrypoint(
  definition: define.WorkflowDefinition(
    input,
    output,
    workflow_error,
  ),
  raw_input: dynamic.Dynamic,
) -> Result(String, String)

Drive a workflow’s typed entry from the engine’s raw spawn argument, making the hand-written run adapter a one-line shim:

pub fn run(raw_input: Dynamic) -> Result(String, String) {
  workflow.entrypoint(definition(), raw_input)
}

Success encodes with the definition’s output codec, typed failure with its error codec (byte-identical to the hand-written adapter); an undecodable input yields the documented {"aion_error":"input_decode",...} envelope.

error_codec

</>

pub fn error_codec(
  definition: define.WorkflowDefinition(
    input,
    output,
    workflow_error,
  ),
) -> codec.Codec(workflow_error)

id

</>

pub fn id() -> Result(String, error.EngineError)

input_codec

</>

pub fn input_codec(
  definition: define.WorkflowDefinition(
    input,
    output,
    workflow_error,
  ),
) -> codec.Codec(input)

map

</>

pub fn map(
  items: List(value),
  to_activity: fn(value) -> activity.Activity(input, output),
) -> Result(List(output), error.ActivityError)

map_settled

</>

pub fn map_settled(
  items: List(value),
  to_activity: fn(value) -> activity.Activity(input, output),
) -> List(Result(output, error.ActivityError))

Dynamically produce one activity per input element, then settle like [all_settled]: one Result slot per item, item order, no fail-fast.

map_settled_with_default

</>

pub fn map_settled_with_default(
  items: List(value),
  to_activity: fn(value) -> activity.Activity(input, output),
  workflow_default_task_queue: option.Option(String),
) -> List(Result(output, error.ActivityError))

map_settled, supplying the workflow-level default task queue for any produced activity that selects none. See [run_with_default] for the precedence.

map_with_default

</>

pub fn map_with_default(
  items: List(value),
  to_activity: fn(value) -> activity.Activity(input, output),
  workflow_default_task_queue: option.Option(String),
) -> Result(List(output), error.ActivityError)

map, supplying the workflow-level default task queue for any produced activity that selects none. See [run_with_default] for the precedence.

name

</>

pub fn name(
  definition: define.WorkflowDefinition(
    input,
    output,
    workflow_error,
  ),
) -> String

now

</>

pub fn now() -> Result(run.Timestamp, error.EngineError)

output_codec

</>

pub fn output_codec(
  definition: define.WorkflowDefinition(
    input,
    output,
    workflow_error,
  ),
) -> codec.Codec(output)

race

</>

pub fn race(
  activities: List(activity.Activity(input, output)),
) -> Result(output, error.ActivityError)

race_with_default

</>

pub fn race_with_default(
  activities: List(activity.Activity(input, output)),
  workflow_default_task_queue: option.Option(String),
) -> Result(output, error.ActivityError)

race, supplying the workflow-level default task queue for any member that selects none. See [run_with_default] for the resolution precedence.

random

</>

pub fn random() -> Result(Float, error.EngineError)

random_int

</>

pub fn random_int(
  min: Int,
  max: Int,
) -> Result(Int, error.EngineError)

receive

</>

pub fn receive(
  reference: signal.SignalRef(payload),
) -> Result(payload, error.ReceiveError)

run

</>

pub fn run(
  activity: activity.Activity(input, output),
) -> Result(output, error.ActivityError)

run_with_default

</>

pub fn run_with_default(
  activity: activity.Activity(input, output),
  workflow_default_task_queue: option.Option(String),
) -> Result(output, error.ActivityError)

run, supplying the workflow-level default task queue applied when the activity selects none. Precedence (activity override > workflow default > the named "default" queue) is resolved once at the engine schedule seam.

sleep

</>

pub fn sleep(
  duration: duration.Duration,
) -> Result(Nil, error.EngineError)

spawn

</>

pub fn spawn(
  name: String,
  workflow_fn: fn(input) -> Result(output, workflow_error),
  input: input,
  input_codec: codec.Codec(input),
  output_codec: codec.Codec(output),
  error_codec: codec.Codec(workflow_error),
) -> Result(
  child.ChildHandle(output, workflow_error),
  error.EngineError,
)

spawn_and_wait

</>

pub fn spawn_and_wait(
  name: String,
  workflow_fn: fn(input) -> Result(output, workflow_error),
  input: input,
  input_codec: codec.Codec(input),
  output_codec: codec.Codec(output),
  error_codec: codec.Codec(workflow_error),
) -> Result(output, error.ChildError(workflow_error))

start_timer

</>

pub fn start_timer(
  name: String,
  duration: duration.Duration,
) -> Result(timer.TimerRef, error.EngineError)

timer_id

</>

pub fn timer_id(reference: timer.TimerRef) -> String

timestamp_to_milliseconds

</>

pub fn timestamp_to_milliseconds(timestamp: run.Timestamp) -> Int

with_timeout

</>

pub fn with_timeout(
  operation: fn() -> Result(value, inner_error),
  deadline: duration.Duration,
) -> Result(value, error.TimeoutResultError(inner_error))