Creates a graph node that archives the execution when unblocked. Archived executions are excluded from load/2 unless include_archived: true is passed.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`archive()` doctest graph (a useless machine that archives itself immediately;)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         archive(:archive, [:name])
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution.archived_at == nil
true
iex> execution = Journey.set(execution, :name, "Mario")
iex> {:ok, _, _} = Journey.get(execution, :archive, wait: :any)
iex> Journey.load(execution)
nil
iex> execution = Journey.load(execution, include_archived: true)
iex> execution.archived_at == nil
false

Options

• :f_on_save — see f_on_save callbacks. On success, result is {:ok, archived_at}.

Creates a self-computing node.

name is an atom uniquely identifying the node in this graph.

gated_by defines when this node becomes eligible to compute. Accepts either:

• A list of atom node names, e.g. [:a, :b], indicating the node becomes unblocked when all of the listed nodes have a value.
• A keyword list with conditions, e.g. [a: fn node -> node.node_value > 10 end], for conditional dependencies.
• A mixed list combining atoms and keyword conditions, e.g. [:a, :b, c: fn node -> node.node_value > 5 end].
• A structured condition (see unblocked_when/1 ) allowing for logical operators (:and, :or) and custom value predicates (e.g. unblocked_when({:and, [{:a, &provided?/1}, {:b, &provided?/1}]})).

f_compute is the function that computes the value of the node, once the upstream dependencies are satisfied. The function can accept either one or two arguments:

• Arity 1: fn values_map -> ... end - Receives a map of node names to their values, for all nodes whose value has been set or computed
• Arity 2: fn values_map, value_nodes_map -> ... end - Additionally receives value node data from upstream nodes

The function must return a tuple:

• {:ok, value} or
• {:error, reason}.

The value_nodes_map (when using arity-2) contains detailed information for each upstream dependency, keyed by node name. Each entry is a map with the following fields:

• :node_value - The current value of the node
• :metadata - Metadata set via Journey.set/3
• :revision - The revision number when this value was set
• :set_time - Unix timestamp when the value was set

This is useful for accessing contextual information like author IDs, timestamps, or revision tracking. The function is called when the upstream nodes are set, and the value is set to the result of the function.

Note that return values will be JSON-serialized for storage. If the returned value or reason contains atoms (e.g., {:ok, :pending} or {:ok, %{status: :active}}), those atoms will be converted to strings when retrieved via get_value/3.

In the case of a failure, the function is automatically retried, up to max_retries times (default: 3). If the function still fails after all retry attempts, the node's value remains unset. If the function does not return within abandon_after_seconds (default: 60), it is considered abandoned, and it will be retried (up to max_retries times).

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`compute()` doctest graph (pig-latinize-ish a name)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         compute(
...>           :pig_latin_ish_name,
...>           [:name],
...>           fn %{name: name} ->
...>             {:ok, "#{name}-ay"}
...>           end,
...>           max_retries: 4, # Optional (default: 3)
...>           abandon_after_seconds: 60, # Optional (default: 60)
...>           f_on_save: fn _execution_id, _node_name, _result ->
...>             # Optional callback invoked when the value is saved.
...>             # Useful for notifying other systems (e.g. a LiveView
...>             # via PubSub.notify()). The 2-arity form is deprecated.
...>             :ok
...>           end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start() |> Journey.set(:name, "Alice")
iex> {:ok, "Alice-ay", 3} = execution |> Journey.get(:pig_latin_ish_name, wait: :any)
iex> execution |> Journey.values() |> redact([:execution_id, :last_updated_at])
%{name: "Alice", pig_latin_ish_name: "Alice-ay", execution_id: "...", last_updated_at: 1_234_567_890}

Keyword List Syntax for Conditional Dependencies

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "threshold alert example",
...>       "v1.0.0",
...>       [
...>         input(:temperature),
...>         # Using keyword list syntax for conditional dependency
...>         compute(
...>           :high_temp_alert,
...>           [temperature: fn node -> node.node_value > 30 end],
...>           fn %{temperature: temp} ->
...>             {:ok, "High temperature alert: #{temp}°C"}
...>           end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution = Journey.set(execution, :temperature, 25)
iex> Journey.get(execution, :high_temp_alert)
{:error, :not_set}
iex> execution = Journey.set(execution, :temperature, 35)
iex> {:ok, "High temperature alert: 35°C", 4} = Journey.get(execution, :high_temp_alert, wait: :any)

Using Value Node Data in Compute Functions

Value node data can be accessed by defining an arity-2 compute function. This is useful for accessing contextual information like author IDs, timestamps, revisions, or data provenance from upstream nodes.

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "compute with value node data example",
...>       "v1.0.0",
...>       [
...>         input(:title),
...>         compute(
...>           :title_with_author,
...>           [:title],
...>           # Arity-2 function receives value node data from dependencies
...>           fn %{title: title}, value_nodes_map ->
...>             author = get_in(value_nodes_map, [:title, :metadata, "author_id"]) || "unknown"
...>             {:ok, "#{title} by #{author}"}
...>           end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution = Journey.set(execution, :title, "Hello", metadata: %{"author_id" => "user123"})
iex> {:ok, result, _} = Journey.get(execution, :title_with_author, wait: :any)
iex> result
"Hello by user123"

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

At-least-once semantics

f_compute is invoked at-least-once: retries on {:error, _}, abandonment after abandon_after_seconds, and worker crashes mid-computation can all cause the same logical computation to be re-invoked with the same inputs. The engine does not deduplicate invocations.

This is great for pure computation, but if your function has a consequential non-idempotent side effect (e.g. charging the customer's credit card), your function will need to provide its own protection against duplicates. One possible approach is to compose and use an idempotency key that includes the id of the execution and the node's name.

Options

• :f_on_save — see f_on_save callbacks. On success, result is {:ok, value} where value is what f_compute returned.

Computation Retention

By default, all completed computation records are kept in the database. For nodes that accumulate many computations over time (especially downstream of tick_recurring nodes), you can limit how many are retained:

compute(:send_summary, deps, &send/1,
  keep_latest_completed_computations: 50
)

After each successful completion, this deletes older computations beyond the retention window, while always preserving the 10 oldest successful computations. Set to :all to explicitly keep everything, or nil (default) to inherit from the graph-level setting.

Can also be set graph-wide via Journey.new_graph/4's :keep_latest_completed_computations option.

Creates a history-tracking node that maintains a chronological log of changes to one or more nodes.

name is an atom uniquely identifying this history node.

gated_by defines which nodes to track. Accepts the same formats as compute/4:

• A single-item list like [:node_name] to track one node
• A list like [:a, :b] to track multiple nodes (all must be set)
• Complex conditions using unblocked_when/1 (e.g., unblocked_when({:or, [{:a, &provided?/1}, {:b, &provided?/1}]}))

The historian will track changes to ALL nodes in the dependency tree and record only those that have changed since the last recording.

Options

• :max_entries (optional) - Maximum number of history entries to keep (FIFO). Defaults to 1000. Set to nil for unlimited history.
• :f_on_save — see f_on_save callbacks. On success, result is {:ok, history}.

History Format

The history is returned as a list of entries in newest-first order (most recent changes at index 0). Each entry is a map containing:

• "value" - The value of the changed node
• "node" - The name of the node (as string)
• "timestamp" - Unix timestamp when recorded
• "metadata" - Metadata from the node (if any)
• "revision" - Revision number of the node when recorded

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`historian()` doctest graph (tracks content changes)",
...>       "v1.0.0",
...>       [
...>         input(:content),
...>         historian(:content_history, [:content])
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution = Journey.set(execution, :content, "First version")
iex> {:ok, history1, history1_rev} = Journey.get(execution, :content_history, wait: :any)
iex> length(history1)
1
iex> [%{"value" => "First version", "node" => "content", "timestamp" => _ts}] = history1
iex> execution = Journey.set(execution, :content, "Second version")
iex> {:ok, history2, _} = Journey.get(execution, :content_history, wait: {:newer_than, history1_rev})
iex> length(history2)
2
iex> [%{"value" => "Second version", "node" => "content", "timestamp" => _ts}, _] = history2

With custom max_entries limit:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "historian with max_entries",
...>       "v1.0.0",
...>       [
...>         input(:status),
...>         historian(:status_history, [:status], max_entries: 2)
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution = Journey.set(execution, :status, "pending")
iex> {:ok, history, rev1} = Journey.get(execution, :status_history, wait: :any)
iex> Enum.map(history, fn entry -> entry["value"] end)
["pending"]
iex> execution = Journey.set(execution, :status, "active")
iex> {:ok, history, rev2} = Journey.get(execution, :status_history, wait: {:newer_than, rev1})
iex> Enum.map(history, fn entry -> entry["value"] end)
["active", "pending"]
iex> execution = Journey.set(execution, :status, "completed")
iex> {:ok, history, _rev} = Journey.get(execution, :status_history, wait: {:newer_than, rev2})
iex> # Since status_history is limited to `max_entries: 2`, we'll only see the 2 latest values (newest first).
iex> Enum.map(history, fn entry -> entry["value"] end)
["completed", "active"]

With unlimited history:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "historian with unlimited history",
...>       "v1.0.0",
...>       [
...>         input(:audit_event),
...>         # Explicitly opt-in to unlimited history for audit trail
...>         historian(:audit_log, [:audit_event], max_entries: nil)
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution = Journey.set(execution, :audit_event, "login")
iex> {:ok, history, _} = Journey.get(execution, :audit_log, wait: :any)
iex> length(history)
1
iex> [%{"value" => "login", "node" => "audit_event", "timestamp" => _ts}] = history

Tracking multiple nodes with :or condition:

iex> import Journey.Node
iex> import Journey.Node.Conditions
iex> import Journey.Node.UpstreamDependencies
iex> graph = Journey.new_graph(
...>       "historian multi-node example",
...>       "v1.0.0",
...>       [
...>         input(:email),
...>         input(:phone),
...>         # Track changes to either email or phone
...>         historian(
...>           :contact_history,
...>           unblocked_when({
...>             :or,
...>             [{:email, &provided?/1}, {:phone, &provided?/1}]
...>           })
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> execution = Journey.set(execution, :email, "user@example.com")
iex> {:ok, history1, history1_rev} = Journey.get(execution, :contact_history, wait: :any)
iex> length(history1)
1
iex> [%{"value" => "user@example.com", "node" => "email"}] = history1
iex> execution = Journey.set(execution, :phone, "555-1234")
iex> {:ok, history2, _} = Journey.get(execution, :contact_history, wait: {:newer_than, history1_rev})
iex> length(history2)
2
iex> # Newest first: phone, then email
iex> [%{"value" => "555-1234", "node" => "phone"}, %{"value" => "user@example.com", "node" => "email"}] = history2

Creates a graph input node. The value of an input node is set with Journey.set/3. The name of the node must be an atom.

Accepts an optional keyword list of options:

• :f_on_save - Callback invoked after the input value is persisted via Journey.set/3. Only fires with {:ok, value} (inputs don't error). See f_on_save callbacks for the shared contract.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`input()` doctest graph (just a few input nodes)",
...>       "v1.0.0",
...>       [
...>         input(:first_name),
...>         input(:last_name),
...>         input(:zip_code)
...>        ]
...>     )
iex> execution = graph |> Journey.start() |> Journey.set(:first_name, "Mario")
iex> Journey.values(execution) |> redact([:execution_id, :last_updated_at])
%{first_name: "Mario", execution_id: "...", last_updated_at: 1_234_567_890}

With f_on_save:

input(:comment, f_on_save: fn execution_id, node_name, {:ok, value} ->
  Phoenix.PubSub.broadcast(MyApp.PubSub, "execution:#{execution_id}", {node_name, value})
  :ok
end)

Creates a loop node — a self-iterating computation that threads state across iterations durably.

loop/4 is the durable analogue of Elixir's Stream.unfold/2 and Enum.reduce_while/3: the step function f_loop is invoked repeatedly, threading state forward via values_map.<name>, until either the function signals completion or :max_iterations is reached.

Each iteration is recorded as its own computation row, providing per-iteration durability: if a worker crashes mid-iteration, retries resume from the last successful iteration rather than restarting from iteration 1.

name is an atom uniquely identifying the node in this graph.

gated_by defines when the loop becomes eligible to start (same syntax as compute/4).

f_loop is the step function. It can accept either one or two arguments:

• Arity 1: fn values_map -> ... end
• Arity 2: fn values_map, value_nodes_map -> ... end

On each invocation:

• values_map contains the current values of all upstream dependencies.
• On the first iteration, the <name> key is absent from values_map — use bracket access (values_map[:<name>]) rather than dot access to read it safely.
• On subsequent iterations, values_map[:<name>] reflects the value carried by the most recent :cont_* return.

The function must return one of four tuples:

Options

• :max_iterations (required) — positive integer; the maximum number of iterations. :infinity is not allowed.

• :max_retries (default: 3) — per-iteration retry budget for {:error, _} returns.

• :abandon_after_seconds (default: 60) — per-iteration timeout.

• :f_on_save — see f_on_save callbacks. Loop-specific result shapes:

  • {:ok, value} — terminal :ok from the step function, or cap-promoted :cont_with_fallback at :max_iterations.
  • {:error, "max_iterations_reached"} — cap-failed :cont_no_fallback at :max_iterations.

  For per-iteration progress visibility, use Journey.Tools.introspect/1.

• Plus :heartbeat_interval_seconds, :heartbeat_timeout_seconds, :keep_latest_completed_computations, :keep_oldest_completed_computations (same semantics as compute/4).

Visibility

The loop node's value (visible via Journey.values/1, unblocked_when predicates, f_on_save callbacks) is set in exactly two cases:

1. The step function returns {:ok, value}.
2. :max_iterations is reached and the most recent return was {:cont_with_fallback, value}.

A loop node's value reflects the last successful write, not the outcome of the current run — even across re-runs triggered by upstream changes.

During iteration, the loop's accumulating state is visible only to the step function itself (via values_map[:<name>]); it is not externally visible until terminal resolution.

Carried values go through JSONB

The value in {:cont_with_fallback, value} / {:cont_no_fallback, value} is persisted as JSON between iterations, following the same rules as Journey.set/3: atoms become strings, and map keys must be strings. Use strings (not atoms) for any value you intend to pattern-match on in a later iteration — e.g. prefer {:cont_with_fallback, "keep_going"} over {:cont_with_fallback, :keep_going}, since the latter will be read back as "keep_going" on the next iteration.

At-least-once semantics

Step functions inherit compute/4's at-least-once contract — see At-least-once semantics for the general guidance on idempotency.

If your loop function performs non-idempotent operations with consequential side effects (e.g. charging a credit card), consider moving the operation to a downstream compute/4 node, where looping doesn't complicate the semantics.

Example: iteratively shrink a string until it fits

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`loop()` doctest graph (shrink a string)",
...>       "v1.0.0",
...>       [
...>         input(:document),
...>         input(:target_length),
...>         loop(
...>           :summary,
...>           [:document, :target_length],
...>           fn values ->
...>             current = values[:summary] || values.document
...>
...>             cond do
...>               String.length(current) <= values.target_length ->
...>                 {:ok, current}
...>
...>               String.length(current) <= 1 ->
...>                 {:ok, current}
...>
...>               true ->
...>                 shorter = String.slice(current, 0, div(String.length(current), 2))
...>                 {:cont_with_fallback, shorter}
...>             end
...>           end,
...>           max_iterations: 10
...>         )
...>       ]
...>     )
iex> execution =
...>   graph
...>   |> Journey.start()
...>   |> Journey.set(:document, "abcdefghijklmnop")
...>   |> Journey.set(:target_length, 4)
iex> {:ok, "abcd", _} = Journey.get(execution, :summary, wait: :any)

Creates a graph node that mutates the value of another node.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`mutate()` doctest graph (a useless machine;)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         mutate(
...>           :remove_pii,
...>           [:name],
...>           fn %{name: _name} ->
...>             # Return the new value for the "name" node.
...>             {:ok, "redacted"}
...>           end,
...>           mutates: :name # The name of an existing node whose value will be mutated.
...>         )
...>       ]
...>     )
iex> execution =
...>     graph
...>     |> Journey.start()
...>     |> Journey.set(:name, "Mario")
iex> {:ok, "updated :name", 3} = execution |> Journey.get(:remove_pii, wait: :any)
iex> execution |> Journey.values() |> redact([:execution_id, :last_updated_at])
%{name: "redacted", remove_pii: "updated :name",  execution_id: "...", last_updated_at: 1_234_567_890}

Options

• :update_revision_on_change (default: false) - When true, updating the mutated node's value also increments its revision, triggering downstream nodes to recompute. When false, mutations update the value without triggering recomputation. Setting this to true for a node that mutates an upstream dependency will raise a validation error to prevent cycles.
• :f_on_save — see f_on_save callbacks. On success, result is {:ok, "updated :target"}.

External Polling Pattern

A common use case is polling external data sources and triggering downstream recomputations:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "current market pricing graph",
...>       "v1.0.0",
...>       [
...>         tick_recurring(:poll_schedule, [], fn _ -> {:ok, System.system_time(:second) + 2} end),
...>         input(:cached_price),
...>         mutate(:fetch_current_price, [:poll_schedule],
...>           fn %{cached_price: cached_price} -> {:ok, (cached_price || 0) + 1} end,
...>           mutates: :cached_price,
...>           update_revision_on_change: true
...>         ),
...>         compute(:on_cached_price_update, [:cached_price],
...>           fn %{cached_price: new_price} -> {:ok, "price changed to #{new_price}"} end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start() |> Journey.set(:cached_price, 5)
iex> {:ok, "price changed to 5", initial_rev} = Journey.get(execution, :on_cached_price_update, wait: :any)
iex> background_sweeps_task = Journey.Scheduler.Background.Periodic.start_background_sweeps_in_test(execution.id)
iex> {:ok, "updated :cached_price", _} = Journey.get(execution, :fetch_current_price, wait: :any)
iex> {:ok, "price changed to 6", _} = Journey.get(execution, :on_cached_price_update, wait: {:newer_than, initial_rev})
iex> Journey.Scheduler.Background.Periodic.stop_background_sweeps_in_test(background_sweeps_task)

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

Creates a graph node that declares its readiness at a specific time, once.

Once this node is unblocked, it will be executed to set the time at which it will unblock its downstream dependencies.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`tick_once()` doctest graph (it reminds you to take a nap in a couple of seconds;)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         tick_once(
...>           :schedule_a_nap,
...>           [:name],
...>           fn %{name: _name} ->
...>             # This function is to return the time (in epoch seconds) at which
...>             # its downstream dependencies should be unblocked.
...>             in_two_seconds = System.system_time(:second) + 2
...>             {:ok, in_two_seconds}
...>           end
...>         ),
...>         compute(
...>           :nap_time,
...>           [:name, :schedule_a_nap],
...>           fn %{name: name, schedule_a_nap: _time_to_take_a_nap} ->
...>             {:ok, "It's time to take a nap, #{name}!"}
...>           end
...>         )
...>       ]
...>     )
iex> execution =
...>     graph
...>     |> Journey.start()
...>     |> Journey.set(:name, "Mario")
iex> execution |> Journey.values() |> Map.get(:name)
"Mario"
iex> # This is only needed in a test, to simulate the background processing that happens in non-tests automatically.
iex> background_sweeps_task = Journey.Scheduler.Background.Periodic.start_background_sweeps_in_test(execution.id)
iex> {:ok, "It's time to take a nap, Mario!", _} = execution |> Journey.get(:nap_time, wait: :any)
iex> Journey.Scheduler.Background.Periodic.stop_background_sweeps_in_test(background_sweeps_task)

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

Returning {:ok, 0} indicates that the schedule should not trigger downstream computations. This might be useful for conditional scheduling - when the function determines that scheduling is not needed based on the current state.

Options

• :f_on_save — see f_on_save callbacks. On success, result is {:ok, schedule_time} (the unix-epoch second returned by f_compute).

Conditional Scheduling Example

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "conditional scheduling example",
...>       "v1.0.0",
...>       [
...>         input(:user_name),
...>         input(:wants_reminder),
...>         tick_once(
...>           :schedule_reminder,
...>           [:user_name, :wants_reminder],
...>           fn %{wants_reminder: wants_reminder} ->
...>             if wants_reminder do
...>               # Schedule for 2 seconds from now
...>               {:ok, System.system_time(:second) + 2}
...>             else
...>               # Don't schedule - user opted out
...>               {:ok, 0}
...>             end
...>           end
...>         ),
...>         compute(
...>           :send_reminder,
...>           [:user_name, :schedule_reminder],
...>           fn %{user_name: name} ->
...>             {:ok, "Reminder for #{name}"}
...>           end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start()
iex> background_sweeps_task = Journey.Scheduler.Background.Periodic.start_background_sweeps_in_test(execution.id)
iex> # User doesn't want reminders - schedule returns 0
iex> execution = Journey.set(execution, :user_name, "Mario")
iex> execution = Journey.set(execution, :wants_reminder, false)
iex> {:ok, 0, _} = Journey.get(execution, :schedule_reminder, wait: :any)
iex> # Reminder is never sent because schedule returned 0
iex> :timer.sleep(3_000)
iex> Journey.get(execution, :send_reminder)
{:error, :not_set}
iex> Journey.Scheduler.Background.Periodic.stop_background_sweeps_in_test(background_sweeps_task)

Creates a graph node that declares its readiness at a specific time, time after time.

Once this node is unblocked, it will be repeatedly computed, to set the time at which it will unblock its downstream dependencies.

This is useful for triggering recurring tasks, such as sending reminders or notifications.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`tick_recurring()` doctest graph (it issues 'reminders' every few seconds)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         tick_recurring(
...>           :schedule_a_reminder,
...>           [:name],
...>           fn _ ->
...>             soon = System.system_time(:second) + 2
...>             {:ok, soon}
...>           end
...>         ),
...>         compute(
...>           :send_a_reminder,
...>           [:name, :schedule_a_reminder],
...>           fn %{name: name} = v ->
...>             reminder_count = Map.get(v, :send_a_reminder, 0) + 1
...>             IO.puts("[#{System.system_time(:second)}] #{name}, here is your scheduled reminder # #{reminder_count}.")
...>             {:ok, reminder_count}
...>           end
...>         )
...>       ]
...>     )
iex> execution =
...>     graph
...>     |> Journey.start()
...>     |> Journey.set(:name, "Mario")
iex> execution |> Journey.values() |> Map.get(:name)
"Mario"
iex> # This is only needed in a test, to simulate the background processing that happens in non-tests automatically.
iex> background_sweeps_task = Journey.Scheduler.Background.Periodic.start_background_sweeps_in_test(execution.id)
iex> # Wait for initial reminders
iex> {:ok, count1, _} = Journey.get(execution, :send_a_reminder, wait: :any)
iex> count1 >= 1
true
iex> # Wait for more reminders to verify recurring behavior
iex> execution = Journey.load(execution)
iex> {:ok, count2, _} = Journey.get(execution, :send_a_reminder, wait: :newer)
iex> count2 > count1
true
iex> execution = Journey.load(execution)
iex> {:ok, count3, _} = Journey.get(execution, :send_a_reminder, wait: :newer)
iex> count3 > count2
true
iex> Journey.Scheduler.Background.Periodic.stop_background_sweeps_in_test(background_sweeps_task)

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

Returning {:ok, 0} indicates that the schedule should not trigger downstream computations.

Options

• :f_on_save — see f_on_save callbacks. On success, result is {:ok, schedule_time} (the unix-epoch second returned by f_compute for one scheduled fire).

Computation Retention

Recurring nodes accumulate computation records over time. Use :keep_latest_completed_computations to limit retention. See compute/4 for details.

tick_recurring(:schedule, deps, &next_time/1,
  keep_latest_completed_computations: 200
)