Arrea (Arrea v2.2.0)

Copy Markdown View Source

Main facade of the Arrea orchestrator.

Responsible for:

Architecture

┌─────────────────────────────────────────────────────────┐
│                      Arrea                              │
│                    (Facade)                             │
└─────────────────────────┬───────────────────────────────┘
                         │
┌────────────────────────▼───────────────────────────────┐
│              Arrea.Leader (GenServer)                   │
│         Coordinates execution, manages workers           │
│         Emits {:leader_event, event} events           │
└─────────────────────────┬───────────────────────────────┘
                         │
┌────────────────────────▼───────────────────────────────┐
│       Arrea.WorkerSupervisor (DynamicSupervisor)       │
│              Ephemeral workers                          │
└─────────────────────────┬───────────────────────────────┘
                         │
┌────────────────────────▼───────────────────────────────┐
│              Arrea.Worker (GenServer)                  │
│              Executes individual tasks                │
└─────────────────────────────────────────────────────────┘

Arrea.Monitor — Worker lifecycle statistics
Arrea.CircuitBreaker — Fault tolerance

The supervision tree starts automatically when including Arrea as a dependency. There is no need to start Arrea.Supervisor manually.

Quick start

# Simple execution
{:ok, result} = Arrea.execute(fn -> :ok end)

# Parallel execution
{:ok, results} = Arrea.run([fn -> 1 end, fn -> 2 end], workers: 4)

# Subscribing to Leader events
:ok = Arrea.subscribe()

receive do
  {:leader_event, %{type: :worker_started} = event} ->
    IO.inspect(event)
  {:leader_event, %{type: :finished} = event} ->
    IO.inspect(event)
end

:ok = Arrea.unsubscribe()

Summary

Types

execution_option()

Functions

execute(cmd, opts \\ [])

Executes a single command synchronously.

max_workers()

Returns the maximum number of workers configured.

run(commands, opts \\ [])

Executes multiple commands in parallel.

run_sync(commands, opts \\ [])

Runs a list of commands or functions in parallel and waits for all to complete.

stats()

Returns the current statistics of the Engine.

subscribe()

Subscribes the current process to the Leader's semantic events.

unsubscribe()

Cancels the subscription of the current process to Leader events.

Types

execution_option()

@type execution_option() ::
  {:workers, non_neg_integer()}
  | {:timeout, non_neg_integer()}
  | {:retry, boolean()}
  | {:validate, boolean()}

Functions

execute(cmd, opts \\ [])

@spec execute(binary() | (-> term()), [execution_option()]) ::
  {:ok, Arrea.Result.t()} | {:error, Arrea.Error.t()}

Executes a single command synchronously.

Parameters

  • cmd — A binary (shell command) or a zero-arity function
  • opts — Additional options:
    • :timeout — Timeout in ms (default 30_000). Real timeout: cancels execution.
    • :retry — Whether to retry on failure
    • :shell — Shell to use (highest priority over config and environment)
    • :validate — Run safety validation before executing (default false). Set to true to reject dangerous commands (:rm -rf, :sudo, etc.) before they run. Disabled by default to preserve prior behaviour; trusted internal callers (e.g. Apero's info commands) can keep the default and pay no per-call cost.

Returns

  • {:ok, Arrea.Result.t()} — Success with result
  • {:error, Arrea.Error.t()} — Error with code and message

Examples

iex> Arrea.execute("echo hello")
{:ok, %Arrea.Result{success: true, data: %{stdout: "hello\n", ...}, failures: []}}

iex> Arrea.execute(fn -> :work end)
{:ok, %Arrea.Result{success: true, data: :work, failures: []}}

iex> Arrea.execute("rm -rf /tmp", validate: true)
{:error, %Arrea.Error{code: :validation_failed, message: "{:dangerous_command, \"rm -rf\"}"}}

max_workers()

@spec max_workers() :: non_neg_integer()

Returns the maximum number of workers configured.

Example

iex> Arrea.max_workers()
100

run(commands, opts \\ [])

@spec run([binary() | (-> term())], [execution_option()]) ::
  {:ok, Arrea.Result.t()} | {:error, Arrea.Error.t()}

Executes multiple commands in parallel.

Parameters

  • commands — List of binaries or functions
  • opts — Options:
    • :workers — Maximum number of parallel workers (default max_workers())
    • :timeout — Total timeout in ms

Returns

  • {:ok, Arrea.Result.t()} — With batch_id to correlate events
  • {:error, Arrea.Error.t()} — If all failed or no workers are available

Example

iex> {:ok, result} = Arrea.run([fn -> 1 end, fn -> 2 end, fn -> 3 end], workers: 2)
iex> result.data.batch_id
"batch_..."

run_sync(commands, opts \\ [])

@spec run_sync(
  [binary() | (-> term())],
  keyword()
) :: [map()]

Runs a list of commands or functions in parallel and waits for all to complete.

Thin facade over the internal parallel engine so consumers do not need to reach into Arrea.Parallel (which is marked @moduledoc false).

Options

  • :workers — Number of parallel workers (default: 4)
  • :timeout — Default timeout per command in ms (default: 30_000)
  • :ordered — Return results in input order (default: true)

Example

iex> Arrea.run_sync([fn -> 1 end, fn -> 2 end])
[{:ok, %{result: 1, exit_code: 0}}, {:ok, %{result: 2, exit_code: 0}}]

stats()

@spec stats() :: {:ok, map()} | {:error, :monitor_unavailable}

Returns the current statistics of the Engine.

The statistics are provided by Arrea.Monitor, which tracks the lifecycle of all workers started under the Leader.

Example

iex> {:ok, stats} = Arrea.stats()
iex> Map.keys(stats)
[:active_workers, :completed_tasks, :failed_tasks, :total_workers]

subscribe()

@spec subscribe() :: :ok

Subscribes the current process to the Leader's semantic events.

Received messages have the shape {:leader_event, event} where event is a map with at least the :type key.

Common event types:

  • %{type: :worker_started, worker_id: id}
  • %{type: :progress, worker_id: id, percent: float, ...}
  • %{type: :finished, worker_id: id}
  • %{type: :error, worker_id: id, reason: term}
  • %{type: :result, worker_id: id, data: term}

To unsubscribe, call unsubscribe/0.

Example

:ok = Arrea.subscribe()

receive do
  {:leader_event, %{type: :finished, worker_id: id}} ->
    IO.puts("Worker #{id} finished")
end

unsubscribe()

@spec unsubscribe() :: :ok

Cancels the subscription of the current process to Leader events.

Example

:ok = Arrea.unsubscribe()