How to Monitor Reactor with Vigilmon

Reactor is the declarative, asynchronous workflow and saga orchestration engine used by Ash Framework. It models complex multi-step business processes as directed graphs — with concurrency, compensation (rollback), and structured error handling built in. When a workflow step hangs, a compensation chain fails, or the orchestration GenServer crashes, your business process is stuck mid-flight with no way to recover automatically.

External monitoring catches what BEAM supervision and Reactor's internal retry logic can't surface to operators. This tutorial covers:

• A health endpoint that exposes workflow orchestration status
• HTTP uptime monitoring with Vigilmon
• Heartbeat monitoring for long-running Reactor workflows
• Slack alerts and a status page

Step 1: Add Reactor to your project

# mix.exs
defp deps do
  [
    {:reactor, "~> 0.10"},
    {:req, "~> 0.4"}   # for heartbeat pings
  ]
end

Define a simple Reactor workflow:

defmodule MyApp.Workflows.OrderPlacement do
  use Reactor

  input :order_params
  input :customer_id

  step :validate_order, MyApp.Steps.ValidateOrder do
    argument :params, input(:order_params)
  end

  step :charge_payment, MyApp.Steps.ChargePayment do
    argument :order, result(:validate_order)
    argument :customer_id, input(:customer_id)
  end

  step :notify_warehouse, MyApp.Steps.NotifyWarehouse do
    argument :order, result(:validate_order)
    argument :payment, result(:charge_payment)
  end

  return :notify_warehouse
end

Implement a step with compensation:

defmodule MyApp.Steps.ChargePayment do
  use Reactor.Step

  @impl true
  def run(%{order: order, customer_id: customer_id}, _context, _options) do
    case MyApp.Payments.charge(customer_id, order.total) do
      {:ok, charge} -> {:ok, charge}
      {:error, reason} -> {:error, reason}
    end
  end

  @impl true
  def compensate(%{customer_id: customer_id} = _arguments, charge, _context, _options) do
    # Roll back the charge if a later step fails
    case MyApp.Payments.refund(charge.id) do
      {:ok, _} -> :ok
      {:error, reason} -> {:error, reason}
    end
  end
end

Step 2: Build a health endpoint for Reactor orchestration

# lib/my_app_web/plugs/health_check.ex
defmodule MyAppWeb.Plugs.HealthCheck do
  import Plug.Conn
  require Logger

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health"} = conn, _opts) do
    checks = run_checks()
    status = if checks.status == "ok", do: 200, else: 503

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(checks))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp run_checks do
    reactor = reactor_check()
    overall = reactor.status == "ok"

    %{
      status: if(overall, do: "ok", else: "degraded"),
      reactor: reactor
    }
  end

  defp reactor_check do
    start = System.monotonic_time(:millisecond)

    # Run a probe workflow — lightweight no-op to verify Reactor is operational
    result =
      case Reactor.run(MyApp.Workflows.HealthProbe, %{}) do
        {:ok, _result} ->
          elapsed = System.monotonic_time(:millisecond) - start
          {:ok, elapsed}

        {:error, reason} ->
          {:error, inspect(reason)}
      end

    case result do
      {:ok, latency_ms} ->
        %{status: "ok", probe_latency_ms: latency_ms}

      {:error, reason} ->
        Logger.error("[Reactor] health check failed: #{reason}")
        %{status: "error", reason: reason}
    end
  end
end

Define the probe workflow:

defmodule MyApp.Workflows.HealthProbe do
  use Reactor

  step :probe, MyApp.Steps.NoOp do
  end

  return :probe
end

defmodule MyApp.Steps.NoOp do
  use Reactor.Step

  @impl true
  def run(_arguments, _context, _options), do: {:ok, :ok}
end

Register the plug:

# lib/my_app_web/endpoint.ex
plug MyAppWeb.Plugs.HealthCheck
plug MyAppWeb.Router

Test it:

mix phx.server
curl http://localhost:4000/health | jq .
# {
#   "status": "ok",
#   "reactor": {
#     "status": "ok",
#     "probe_latency_ms": 3
#   }
# }

Step 3: Track in-flight workflow counts

For long-running production workflows, expose counts of active and failed workflows in your health response:

defp run_checks do
  reactor = reactor_check()
  workflows = workflow_stats()

  overall = reactor.status == "ok" and workflows.status == "ok"

  %{
    status: if(overall, do: "ok", else: "degraded"),
    reactor: reactor,
    workflows: workflows
  }
end

defp workflow_stats do
  # Query your workflow state table if you persist Reactor state
  stats =
    MyApp.Repo.one(
      from w in MyApp.WorkflowState,
        select: %{
          total: count(w.id),
          running: filter(count(w.id), w.status == "running"),
          failed: filter(count(w.id), w.status == "failed")
        }
    )

  if stats.failed > 0 do
    %{status: "degraded", stats: stats, reason: "#{stats.failed} workflows in failed state"}
  else
    %{status: "ok", stats: stats}
  end
end

Step 4: Monitor the health endpoint with Vigilmon

1. Sign up at vigilmon.online
2. Click New Monitor → HTTP
3. Enter https://yourdomain.com/health
4. Set check interval: 1 minute
5. Enable keyword check: body must contain "status":"ok"
6. Save

The keyword check is critical here: if Reactor can't start a workflow step, your endpoint returns HTTP 200 but with "status":"degraded". The keyword match catches stuck orchestration states before customers see incomplete orders.

Step 5: Heartbeat monitoring for background workflow runners

Long-running background workflows need heartbeat checks to confirm they're making progress:

# lib/my_app/workers/order_workflow_runner.ex
defmodule MyApp.Workers.OrderWorkflowRunner do
  use GenServer
  require Logger

  @interval :timer.minutes(5)

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  def init(state) do
    schedule()
    {:ok, state}
  end

  def handle_info(:run, state) do
    run_pending_workflows()
    schedule()
    {:noreply, state}
  end

  defp run_pending_workflows do
    pending = MyApp.Repo.all(from w in MyApp.WorkflowState, where: w.status == "pending")

    Enum.each(pending, fn workflow ->
      case Reactor.run(MyApp.Workflows.OrderPlacement, workflow.inputs) do
        {:ok, result} ->
          MyApp.WorkflowState.mark_complete(workflow, result)

        {:error, reason} ->
          Logger.error("[OrderWorkflow] workflow #{workflow.id} failed: #{inspect(reason)}")
          MyApp.WorkflowState.mark_failed(workflow, reason)
      end
    end)

    ping_vigilmon()
  end

  defp ping_vigilmon do
    url = Application.get_env(:my_app, :vigilmon)[:heartbeat_url]

    if url do
      case Req.get(url, receive_timeout: 5_000) do
        {:ok, %{status: 200}} -> :ok
        err -> Logger.warning("[OrderWorkflowRunner] heartbeat ping failed: #{inspect(err)}")
      end
    end
  end

  defp schedule, do: Process.send_after(self(), :run, @interval)
end

Add to your supervision tree:

# lib/my_app/application.ex
children = [
  MyApp.Repo,
  MyAppWeb.Endpoint,
  MyApp.Workers.OrderWorkflowRunner
]

In Vigilmon, create a Heartbeat Monitor:

1. Click New Monitor → Heartbeat
2. Set expected interval: 15 minutes
3. Copy the ping URL
4. Set it as VIGILMON_HEARTBEAT_URL in your environment

If the workflow runner crashes or gets stuck on a long-running step, the heartbeat stops — and Vigilmon alerts you while workflows queue up.

Step 6: Telemetry for step-level monitoring

Reactor emits Telemetry events at the step level. Attach handlers to catch slow or failing steps:

defmodule MyApp.ReactorTelemetry do
  require Logger

  def setup do
    :telemetry.attach_many(
      "reactor-steps",
      [
        [:reactor, :step, :stop],
        [:reactor, :step, :error]
      ],
      &handle_event/4,
      nil
    )
  end

  def handle_event([:reactor, :step, :stop], measurements, meta, _config) do
    duration_ms = System.convert_time_unit(measurements.duration, :native, :millisecond)

    if duration_ms > 10_000 do
      Logger.warning("[Reactor] slow step #{meta.step_name} in #{meta.reactor}: #{duration_ms}ms")
    end
  end

  def handle_event([:reactor, :step, :error], _measurements, meta, _config) do
    Logger.error("[Reactor] step #{meta.step_name} in #{meta.reactor} failed: #{inspect(meta.reason)}")
  end
end

Call MyApp.ReactorTelemetry.setup/0 from your application start.

Step 7: Slack alerts and status page

Slack alerts:

1. In Vigilmon, go to Notifications → New Channel → Slack
2. Paste your Slack incoming webhook URL
3. Enable the channel on your Reactor health monitor and workflow heartbeat

When the orchestration probe fails:

🔴 DOWN: yourdomain.com/health
Keyword "status":"ok" not found
Detected: US-East
1 minute ago

When the workflow runner heartbeat misses:

🔴 HEARTBEAT MISSED: Order Workflow Runner
Expected ping within 15 minutes — none received
Last seen: 18 minutes ago

Status page:

1. Go to Status Pages → New Status Page
2. Add your health monitor and heartbeat monitor
3. Share with your team for incident coordination

What you've built

| What | How | |------|-----| | Reactor probe | Lightweight no-op workflow execution check | | Uptime monitoring | Vigilmon HTTP monitor → /health | | Keyword check | Vigilmon keyword match on "status":"ok" | | Workflow state exposure | Failed/running workflow counts in health response | | Background runner monitoring | Heartbeat GenServer + Vigilmon heartbeat monitor | | Step-level telemetry | Reactor Telemetry events + slow step logging | | Slack alerts | Vigilmon Slack notification channel | | Status page | Vigilmon public status page |

Reactor orchestrates your complex business processes reliably. Vigilmon tells you when the orchestrator — or the workflows it's driving — silently stops progressing.

Next steps

• Expose per-workflow-type failure rates in your health response so you can alert when a specific step type (e.g. payment charging) has elevated error rates
• Track compensation chain success rates separately — a failed compensate is worse than a failed forward step
• Use Vigilmon's response time history to detect when workflow probe latency trends upward, which can indicate resource contention on the BEAM scheduler

Get started free at vigilmon.online.