Durable.Context (Durable v1.5.0)
View SourceContext management for workflow execution.
The context is a key-value store that persists across steps within a workflow. It provides a way to share state between steps and is automatically persisted to the database after each step completes.
Usage
defmodule MyApp.OrderWorkflow do
use Durable
use Durable.Context
workflow "process_order" do
step :init do
order = input().order
put_context(:order_id, order.id)
put_context(:items, order.items)
end
step :calculate_total do
items = get_context(:items)
total = Enum.sum(Enum.map(items, & &1.price))
put_context(:total, total)
end
step :finalize do
%{
order_id: get_context(:order_id),
total: get_context(:total)
}
end
end
endContext Storage
During execution, context is stored in the process dictionary for fast access. After each step completes, the context is persisted to the database. When a workflow resumes (e.g., after a sleep or wait), the context is restored from the database.
Summary
Functions
Injects context management functions into the calling module.
Appends a value to a list in the context.
Returns which attempt of the current step this is, starting at 1.
Returns the entire context map.
Returns the current step name.
Deletes a key from the context.
Gets a value from the context by key.
Gets a value from the context by key with a default.
Checks if a key exists in the context.
Increments a numeric value in the context.
Returns the initial workflow input.
Returns true on the current step's last attempt: if it returns an error,
the step won't be tried again.
Returns how many attempts the current step gets: its retry: [max_attempts: n],
or 1 if it has no retry:.
Deep merges a map into the context.
Checks if a parallel step succeeded.
Returns a specific parallel step's result by name.
Returns the full parallel results map from context.
Merges a map into the context.
Puts a single key-value pair into the context.
Returns true once this node has started shutting down.
Updates a context value using a function.
Returns the current workflow ID.
Functions
Injects context management functions into the calling module.
Appends a value to a list in the context.
If the key doesn't exist, creates a new list with the value.
Examples
append_context(:events, %{type: :clicked, timestamp: DateTime.utc_now()})@spec attempt() :: pos_integer()
Returns which attempt of the current step this is, starting at 1.
The count carries over when a run resumes after a crash, and starts again at
1 after Durable.retry/2.
Examples
step :charge, [retry: [max_attempts: 5]], fn ctx ->
Logger.info("charging, attempt #{attempt()} of #{max_attempts()}")
...
end@spec context() :: map()
Returns the entire context map.
Examples
ctx = context()
# => %{order_id: 123, items: [...]}@spec current_step() :: atom() | nil
Returns the current step name.
Examples
step = current_step()Deletes a key from the context.
Examples
delete_context(:temporary_data)
delete_context("temporary_data")Gets a value from the context by key.
Returns nil if the key doesn't exist.
Examples
order_id = get_context(:order_id)Gets a value from the context by key with a default.
Examples
count = get_context(:retry_count, 0)
count = get_context("retry_count", 0)Checks if a key exists in the context.
Examples
if has_context?(:order_id) do
# ...
endIncrements a numeric value in the context.
If the key doesn't exist, starts from 0.
Examples
increment_context(:retry_count, 1)
increment_context(:processed_items, 1)@spec input() :: map()
Returns the initial workflow input.
Examples
order = input().order@spec last_attempt?() :: boolean()
Returns true on the current step's last attempt: if it returns an error,
the step won't be tried again.
Examples
step :charge, [retry: [max_attempts: 5]], fn ctx ->
case Payments.charge(ctx.order) do
{:ok, _} = ok -> ok
{:error, reason} ->
if last_attempt?(), do: Orders.mark_payment_failed(ctx.order)
{:error, reason}
end
end@spec max_attempts() :: pos_integer()
Returns how many attempts the current step gets: its retry: [max_attempts: n],
or 1 if it has no retry:.
@spec merge_context(map()) :: :ok
Deep merges a map into the context.
Examples
merge_context(%{settings: %{notifications: true}})Checks if a parallel step succeeded.
Examples
step :handle, fn ctx ->
if parallel_ok?(:payment) do
# payment succeeded
else
# payment failed
end
endReturns a specific parallel step's result by name.
Examples
step :handle, fn ctx ->
case parallel_result(:payment) do
{:ok, payment} -> # handle success
{:error, reason} -> # handle error
end
end@spec parallel_results() :: map()
Returns the full parallel results map from context.
The results map contains tagged tuples: %{step_name => {:ok, data} | {:error, reason}}
Examples
parallel do
step :payment, fn ctx -> {:ok, %{id: 123}} end
step :delivery, fn ctx -> {:error, :not_found} end
end
step :handle, fn ctx ->
results = parallel_results()
# => %{payment: {:ok, %{id: 123}}, delivery: {:error, :not_found}}
end@spec put_context(map()) :: :ok
Merges a map into the context.
Examples
put_context(%{order_id: 123, customer_id: 456})Puts a single key-value pair into the context.
Examples
put_context(:order_id, 123)
put_context("order_id", 123)@spec shutting_down?() :: boolean()
Returns true once this node has started shutting down.
The node finishes the step it's on, then hands the rest of the run to another
node. A step that runs longer than the :shutdown_grace_period can check this
and stop early. Snoozing doesn't use an attempt, and another node carries on
straight away:
Examples
step :crawl, fn ctx ->
for page <- pages_left(ctx), not shutting_down?(), do: crawl(page)
if shutting_down?(),
do: {:snooze, 0, "node stopping"},
else: {:ok, ctx}
endUpdates a context value using a function.
Examples
update_context(:retry_count, &(&1 + 1))
update_context(:items, &[new_item | &1])@spec workflow_id() :: String.t() | nil
Returns the current workflow ID.
Examples
id = workflow_id()