Temporalex (Temporalex v0.5.4)

Copy Markdown View Source

Durable workflows for Elixir, on Temporal.

The shortest honest call:

greeting = Greet.execute!("Fresha")

A workflow module declares what it is — behaviour, identity, address — and use Temporalex.Workflow generates the call-side surface:

defmodule Greet do
  use Temporalex.Workflow, queue: "greetings"

  @impl true
  def id(name), do: "greet-#{name}"

  @impl true
  def run(name), do: {:ok, "Hello, #{name}!"}
end

This module holds what happens between building a start and running it: the chain steps that shape a Temporalex.Start, the terminal verbs that perform it, and await for collecting a result later.

booking_id
|> Booking.new()
|> Temporalex.retry(max_attempts: 3)
|> Temporalex.fairness(salon_id)
|> Temporalex.index(salon_id: salon_id)
|> Temporalex.execute!()

Builders are inert; terminal verbs run; a bang raises. A chain has exactly one terminal verb, at the end — everything before it is data.

See docs/rfcs/0002-client-surface.md for the full design.

Summary

Functions

await(handle, opts \\ [])

Waits for the workflow's result.

await!(handle, opts \\ [])

Like await/2 but returns the result and raises on failure.

client(start, client)

Selects a named client instead of the default one.

cron(start, expression)

Cron schedule for a recurring workflow.

execute(start)

Performs the start and waits for the result: start/1 + await/2.

execute!(start)

Like execute/1 but returns the result and raises on failure.

execution_timeout(start, ms)

The whole chain's lifetime, retries and continue-as-new included. As consequential as run_timeout/2, and as deliberately undefaulted.

fail!(message)

Raises an application failure describing a business outcome.

fail!(message, opts)
fairness(start, key, weight \\ 1.0)

Fair dispatch key, typically a tenant id — tasks sharing a key are dispatched in proportion to their weight, so one noisy tenant cannot monopolise a queue.

headers(start, headers)

Header payloads — usually an interceptor's job, not the call site's.

id(start, id)

Overrides the workflow id (id/1 normally derived it in new).

index(start, attributes)

Indexed search attributes — machine-findable via temporal workflow list --query. The unindexed, human-readable counterpart is the memo.

input(start, input)

Overrides the durable input — rare: when address and input diverge.

priority(start, key)

Queue priority band — smaller is higher.

queue(start, queue)

Overrides the task queue — for starting on someone else's queue.

retry(start, policy)

Workflow retry policy.

run_timeout(start, ms)

One run's lifetime. Consequential: expiry destroys the run without compensation — which is why there is no default.

start(start)

Performs the start and returns {:ok, handle} — nobody waits.

start!(start)

Like start/1 but returns the handle and raises on failure.

timeout(start, timeout)

How long a waiter waits. Expiring never touches the workflow.

Functions

await(handle, opts \\ [])

@spec await(
  Temporalex.Client.Handle.t(),
  keyword()
) :: {:ok, term()} | {:error, Exception.t()}

Waits for the workflow's result.

A timeout is the caller giving up: the workflow keeps running and the handle stays valid, so awaiting again later is legitimate. The wait bound is opts[:timeout], else the timeout the start chain carried onto the handle, else the client's workflow_result_timeout (60 seconds by default).

await!(handle, opts \\ [])

@spec await!(
  Temporalex.Client.Handle.t(),
  keyword()
) :: term()

Like await/2 but returns the result and raises on failure.

client(start, client)

@spec client(Temporalex.Start.t(), atom()) :: Temporalex.Start.t()

Selects a named client instead of the default one.

cron(start, expression)

@spec cron(Temporalex.Start.t(), String.t()) :: Temporalex.Start.t()

Cron schedule for a recurring workflow.

execute(start)

@spec execute(Temporalex.Start.t()) :: {:ok, term()} | {:error, Exception.t()}

Performs the start and waits for the result: start/1 + await/2.

execute!(start)

@spec execute!(Temporalex.Start.t()) :: term()

Like execute/1 but returns the result and raises on failure.

execution_timeout(start, ms)

@spec execution_timeout(Temporalex.Start.t(), pos_integer()) :: Temporalex.Start.t()

The whole chain's lifetime, retries and continue-as-new included. As consequential as run_timeout/2, and as deliberately undefaulted.

fail!(message)

@spec fail!(String.t() | atom()) :: no_return()

Raises an application failure describing a business outcome.

Temporalex.fail!("amount exceeds limit", type: "AmountTooLarge", retry: false)

Called from an activity, this fails the current attempt. Temporal retries it under the activity's retry policy unless retry: false. type: is the stable string that retry policies and workflow matches key on.

Called from workflow code, this fails the workflow. That is final unless the workflow was started with a retry policy, in which case retry: false makes it final.

Options: type: (a non-empty string — the wire value Temporal matches retry policies against; anything else, nil included, is refused rather than silently replaced when encoded — omit the option to get the default type), retry: (true or false, default true), details:. Use Temporalex.Failure.application!/2 to set a nested :cause.

fail!(message, opts)

@spec fail!(
  String.t() | atom(),
  keyword()
) :: no_return()

fairness(start, key, weight \\ 1.0)

@spec fairness(Temporalex.Start.t(), String.t() | integer(), float()) ::
  Temporalex.Start.t()

Fair dispatch key, typically a tenant id — tasks sharing a key are dispatched in proportion to their weight, so one noisy tenant cannot monopolise a queue.

headers(start, headers)

@spec headers(Temporalex.Start.t(), keyword() | map()) :: Temporalex.Start.t()

Header payloads — usually an interceptor's job, not the call site's.

id(start, id)

@spec id(Temporalex.Start.t(), String.t() | :generate) :: Temporalex.Start.t()

Overrides the workflow id (id/1 normally derived it in new).

index(start, attributes)

@spec index(Temporalex.Start.t(), keyword() | map()) :: Temporalex.Start.t()

Indexed search attributes — machine-findable via temporal workflow list --query. The unindexed, human-readable counterpart is the memo.

input(start, input)

@spec input(Temporalex.Start.t(), term()) :: Temporalex.Start.t()

Overrides the durable input — rare: when address and input diverge.

priority(start, key)

@spec priority(Temporalex.Start.t(), pos_integer()) :: Temporalex.Start.t()

Queue priority band — smaller is higher.

queue(start, queue)

@spec queue(Temporalex.Start.t(), String.t()) :: Temporalex.Start.t()

Overrides the task queue — for starting on someone else's queue.

retry(start, policy)

@spec retry(
  Temporalex.Start.t(),
  keyword()
) :: Temporalex.Start.t()

Workflow retry policy.

run_timeout(start, ms)

@spec run_timeout(Temporalex.Start.t(), pos_integer()) :: Temporalex.Start.t()

One run's lifetime. Consequential: expiry destroys the run without compensation — which is why there is no default.

start(start)

@spec start(Temporalex.Start.t()) ::
  {:ok, Temporalex.Client.Handle.t()} | {:error, Exception.t()}

Performs the start and returns {:ok, handle} — nobody waits.

A duplicate start (same workflow id, still running) attaches to the existing execution by default (id_conflict_policy: :use_existing); pass id_conflict_policy: :fail on new/2 to make duplicates loud errors.

start!(start)

@spec start!(Temporalex.Start.t()) :: Temporalex.Client.Handle.t()

Like start/1 but returns the handle and raises on failure.

timeout(start, timeout)

@spec timeout(Temporalex.Start.t(), pos_integer() | :infinity) :: Temporalex.Start.t()

How long a waiter waits. Expiring never touches the workflow.

On an execute! ending this bounds the call; on a start! ending it is carried on the handle as the default for a later await!.