Gleam activity API

Policies and routing

A new activity has one attempt unless a retry policy is attached. The policy sets total maximum attempts and an explicit exponential, linear or fixed backoff. Supply timeouts and heartbeat intervals for work that needs them.

Task queue selects a worker pool inside a namespace. Node selects locality within that pool. Fallback supplies alternate queues. Labels attach activity metadata. Execution tier selects the linked in-VM path or a remote worker implementation.

Type references and declarations carry named input/output contracts for generated workers and manifests. The accessors expose the values already attached to an activity.

Public types and signatures

pub type Backoff {
  Exponential(
    initial: duration.Duration,
    multiplier: Float,
    max: duration.Duration,
  )
  Linear(
    initial: duration.Duration,
    increment: duration.Duration,
    max: duration.Duration,
  )
  Fixed(delay: duration.Duration)
}

pub type RetryPolicy {
  RetryPolicy(max_attempts: Int, backoff: Backoff)
}

pub type Tier {
  InVm
  RemotePython
  RemoteRust
}

pub opaque type Activity(i, o)

pub opaque type TypeRef(a)

pub opaque type Declaration

pub fn new(
  name: String,
  input: i,
  input_codec: codec.Codec(i),
  output_codec: codec.Codec(o),
  run: fn(i) -> Result(o, error.ActivityError),
) -> Activity(i, o)

pub fn retry(activity: Activity(i, o), policy: RetryPolicy) -> Activity(i, o)

pub fn fallback(activity: Activity(i, o), queues: List(String)) -> Activity(i, o)

pub fn timeout(
  activity: Activity(i, o),
  timeout_duration: duration.Duration,
) -> Activity(i, o)

pub fn heartbeat(
  activity: Activity(i, o),
  heartbeat_interval: duration.Duration,
) -> Activity(i, o)

pub fn label(
  activity: Activity(i, o),
  key: String,
  value: String,
) -> Activity(i, o)

pub fn task_queue(activity: Activity(i, o), name: String) -> Activity(i, o)

pub fn node(activity: Activity(i, o), name: String) -> Activity(i, o)

pub fn execution_tier(activity: Activity(i, o), tier: Tier) -> Activity(i, o)

pub fn name(activity: Activity(i, o)) -> String

pub fn labels(activity: Activity(i, o)) -> List(#(String, String))

pub fn input(activity: Activity(i, o)) -> i

pub fn input_codec(activity: Activity(i, o)) -> codec.Codec(i)

pub fn output_codec(activity: Activity(i, o)) -> codec.Codec(o)

pub fn runner(
  activity: Activity(i, o),
) -> fn(i) -> Result(o, error.ActivityError)

pub fn retry_policy(activity: Activity(i, o)) -> Option(RetryPolicy)

pub fn timeout_duration(activity: Activity(i, o)) -> Option(duration.Duration)

pub fn heartbeat_interval(
  activity: Activity(i, o),
) -> Option(duration.Duration)

pub fn selected_task_queue(activity: Activity(i, o)) -> Option(String)

pub fn selected_fallback(activity: Activity(i, o)) -> List(String)

pub fn selected_node(activity: Activity(i, o)) -> Option(String)

pub fn selected_tier(activity: Activity(i, o)) -> Option(Tier)

pub fn type_ref(type_name: String, value_codec: codec.Codec(a)) -> TypeRef(a)

pub fn type_ref_codec(reference: TypeRef(a)) -> codec.Codec(a)

pub fn declare(
  name: String,
  tier: Tier,
  input: TypeRef(i),
  output: TypeRef(o),
) -> Declaration

pub fn tier_to_string(tier: Tier) -> String

pub fn declaration_name(declaration: Declaration) -> String

pub fn declaration_tier(declaration: Declaration) -> Tier

pub fn declaration_input_type(declaration: Declaration) -> String

pub fn declaration_output_type(declaration: Declaration) -> String

The implementation is gleam/aion_flow/src/aion/activity.gleam. See worker admission for schema matching and placement.

On main, gleam/aion_flow/src/aion/activity.gleam:1–493 declares these activity constructors, codecs and policies.