How to Monitor Exceptional with Vigilmon

Exceptional is a railway-oriented error handling library for Elixir. It provides pipe-friendly operators — ~>, >>>, and ||| — that thread {:ok, value} and {:error, reason} tagged tuples through transformation pipelines without nesting case statements. Instead of deeply nested conditionals, your business logic reads as a left-to-right chain: each step receives an unwrapped value on success and short-circuits to pass the error downstream on failure.

Error handling pipelines that look clean can still be dangerous when they silently swallow failures. A step that produces {:error, :not_found} propagates through every downstream pipe operator untouched — useful behavior, but invisible unless you instrument it. Without observability, you cannot tell whether your railway tracks are routing correctly, which steps fail most often, or whether an error type you assumed was transient is actually growing.

Vigilmon gives you the visibility layer: track error types by pipeline stage, alert on unexpected error rates, and catch regressions before users do.

Why Monitor Exceptional Pipelines?

Railway-oriented programming makes the happy path obvious and the error path explicit — but errors still need to be observed:

• Silent short-circuits — an {:error, reason} entering a pipeline skips all downstream steps without logging, making failures invisible without instrumentation
• Error type drift — error reasons returned by dependencies change between library versions; unmonitored pipelines silently start returning unexpected error types
• Unhandled error accumulation — pipelines that handle most errors with ||| fallbacks may accumulate a class of errors that no fallback covers
• Performance regression in error paths — a step that starts producing errors causes all pipeline continuations to be skipped, masking the fact that the pipeline now takes a different (and possibly slower) code path
• Missing boundary validation — Exceptional makes it easy to compose pipelines across module boundaries; without telemetry, you can't see where in the chain an error originated

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Pipeline success rate by pipeline name | What fraction of executions reach the final {:ok, value} | | Error count by step and reason | Which steps produce which error types | | Error rate over time | Whether a specific error reason is growing or seasonal | | Pipeline duration (success vs. error) | Whether error paths have different latency profiles | | Fallback activation rate | How often ||| fallback operators are invoked | | Unhandled error count | Errors that escape pipelines without a matching fallback |

Step 1: Add Exceptional to Your Project

# mix.exs
defp deps do
  [
    {:exceptional, "~> 2.1"}
  ]
end

Basic railway pipeline:

# lib/my_app/order_pipeline.ex
defmodule MyApp.OrderPipeline do
  use Exceptional

  def process(order_params) do
    order_params
    ~> validate_params()
    ~> fetch_customer()
    ~> check_inventory()
    ~> create_order()
    ~> send_confirmation()
  end

  defp validate_params(params) when is_map(params), do: {:ok, params}
  defp validate_params(_), do: {:error, :invalid_params}

  defp fetch_customer({:ok, %{customer_id: id} = params}) do
    case MyApp.Customers.get(id) do
      nil -> {:error, {:not_found, :customer, id}}
      customer -> {:ok, Map.put(params, :customer, customer)}
    end
  end
  defp fetch_customer(error), do: error

  defp check_inventory({:ok, params}) do
    if MyApp.Inventory.available?(params.product_id, params.quantity) do
      {:ok, params}
    else
      {:error, :insufficient_inventory}
    end
  end
  defp check_inventory(error), do: error

  defp create_order({:ok, params}), do: MyApp.Orders.insert(params)
  defp create_order(error), do: error

  defp send_confirmation({:ok, order}) do
    MyApp.Mailer.send_order_confirmation(order)
    {:ok, order}
  end
  defp send_confirmation(error), do: error
end

Step 2: Instrument Pipeline Execution with Telemetry

Exceptional does not emit telemetry natively. Wrap your pipelines in a measurement helper that records success, error type, and duration:

# lib/my_app/pipeline_telemetry.ex
defmodule MyApp.PipelineTelemetry do
  @moduledoc """
  Wraps Exceptional pipelines with telemetry measurements.
  Call `measure/3` instead of invoking your pipeline function directly.
  """

  require Logger

  @doc """
  Execute a pipeline function and emit telemetry for the result.

  ## Example
      PipelineTelemetry.measure(:order, fn -> OrderPipeline.process(params) end)
  """
  def measure(pipeline_name, fun) do
    start = System.monotonic_time()
    result = fun.()
    duration = System.monotonic_time() - start

    case result do
      {:ok, _value} ->
        :telemetry.execute(
          [:my_app, :pipeline, :complete],
          %{duration: duration, count: 1},
          %{pipeline: pipeline_name, result: :ok}
        )

      {:error, reason} ->
        error_class = classify_error(reason)

        Logger.warning("Pipeline error",
          pipeline: pipeline_name,
          reason: inspect(reason),
          error_class: error_class
        )

        :telemetry.execute(
          [:my_app, :pipeline, :complete],
          %{duration: duration, count: 1},
          %{pipeline: pipeline_name, result: :error, error_class: error_class}
        )

      other ->
        :telemetry.execute(
          [:my_app, :pipeline, :complete],
          %{duration: duration, count: 1},
          %{pipeline: pipeline_name, result: :unexpected}
        )

        Logger.error("Pipeline returned unexpected shape",
          pipeline: pipeline_name,
          result: inspect(other)
        )
    end

    result
  end

  defp classify_error({reason, _context}) when is_atom(reason), do: reason
  defp classify_error(reason) when is_atom(reason), do: reason
  defp classify_error(_), do: :unknown
end

Update your pipeline call sites:

defmodule MyApp.OrdersController do
  def create(conn, params) do
    result = MyApp.PipelineTelemetry.measure(:order, fn ->
      MyApp.OrderPipeline.process(params)
    end)

    case result do
      {:ok, order} -> json(conn, %{order_id: order.id})
      {:error, reason} -> send_resp(conn, 422, inspect(reason))
    end
  end
end

Step 3: Instrument Individual Steps

For granular visibility, emit telemetry at each pipeline step:

# lib/my_app/instrumented_order_pipeline.ex
defmodule MyApp.InstrumentedOrderPipeline do
  use Exceptional

  def process(order_params) do
    order_params
    ~> step(:validate, &validate_params/1)
    ~> step(:fetch_customer, &fetch_customer/1)
    ~> step(:check_inventory, &check_inventory/1)
    ~> step(:create_order, &create_order/1)
    ~> step(:send_confirmation, &send_confirmation/1)
  end

  defp step(name, fun) do
    fn input ->
      start = System.monotonic_time()
      result = fun.(input)
      duration = System.monotonic_time() - start

      outcome = if match?({:ok, _}, result), do: :ok, else: :error

      :telemetry.execute(
        [:my_app, :pipeline, :step],
        %{duration: duration, count: 1},
        %{pipeline: :order, step: name, outcome: outcome}
      )

      result
    end
  end

  defp validate_params(params) when is_map(params), do: {:ok, params}
  defp validate_params(_), do: {:error, :invalid_params}

  defp fetch_customer(%{customer_id: id} = params) do
    case MyApp.Customers.get(id) do
      nil -> {:error, {:not_found, :customer, id}}
      customer -> {:ok, Map.put(params, :customer, customer)}
    end
  end

  defp check_inventory(params) do
    if MyApp.Inventory.available?(params.product_id, params.quantity) do
      {:ok, params}
    else
      {:error, :insufficient_inventory}
    end
  end

  defp create_order(params), do: MyApp.Orders.insert(params)

  defp send_confirmation(order) do
    MyApp.Mailer.send_order_confirmation(order)
    {:ok, order}
  end
end

Step 4: Define Telemetry Metrics

# lib/my_app/telemetry.ex
def metrics do
  [
    # Pipeline-level outcomes
    counter("my_app.pipeline.complete.count",
      tags: [:pipeline, :result, :error_class]
    ),
    distribution("my_app.pipeline.complete.duration",
      unit: {:native, :millisecond},
      reporter_options: [buckets: [5, 25, 100, 500, 2000, 10_000]],
      tags: [:pipeline, :result]
    ),

    # Per-step outcomes
    counter("my_app.pipeline.step.count",
      tags: [:pipeline, :step, :outcome]
    ),
    distribution("my_app.pipeline.step.duration",
      unit: {:native, :millisecond},
      reporter_options: [buckets: [1, 5, 25, 100, 500]],
      tags: [:pipeline, :step, :outcome]
    )
  ]
end

Step 5: Track Error Rates with a Rolling Window

Detect sudden increases in error rate that may indicate a broken dependency:

# lib/my_app/pipeline_error_rate.ex
defmodule MyApp.PipelineErrorRate do
  use GenServer
  require Logger

  @window_seconds 300
  @error_threshold 0.25
  @poll_interval :timer.seconds(30)

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

  def init(state) do
    :telemetry.attach(
      "pipeline-error-rate-tracker",
      [:my_app, :pipeline, :complete],
      &handle_event/4,
      nil
    )

    schedule()
    {:ok, state}
  end

  def handle_event(_event, %{count: 1}, %{pipeline: pipeline, result: result}, _) do
    GenServer.cast(__MODULE__, {:record, pipeline, result, System.system_time(:second)})
  end

  def handle_cast({:record, pipeline, result, ts}, state) do
    event = {pipeline, result, ts}
    {:noreply, %{state | events: [event | state.events]}}
  end

  def handle_info(:check, state) do
    cutoff = System.system_time(:second) - @window_seconds

    recent =
      state.events
      |> Enum.filter(fn {_, _, ts} -> ts > cutoff end)

    recent
    |> Enum.group_by(fn {pipeline, _, _} -> pipeline end)
    |> Enum.each(fn {pipeline, events} ->
      total = length(events)
      errors = Enum.count(events, fn {_, result, _} -> result == :error end)
      rate = if total > 0, do: errors / total, else: 0.0

      :telemetry.execute(
        [:my_app, :pipeline, :error_rate],
        %{rate: rate, total: total, errors: errors},
        %{pipeline: pipeline, window_seconds: @window_seconds}
      )

      if rate > @error_threshold and total > 10 do
        Logger.warning("High pipeline error rate",
          pipeline: pipeline,
          error_rate: Float.round(rate * 100, 1),
          total: total,
          errors: errors
        )
      end
    end)

    pruned = Enum.filter(recent, fn {_, _, ts} -> ts > cutoff end)
    schedule()
    {:noreply, %{state | events: pruned}}
  end

  defp schedule, do: Process.send_after(self(), :check, @poll_interval)
end

Step 6: Health Endpoint

# lib/my_app_web/controllers/health_controller.ex
defmodule MyAppWeb.HealthController do
  use MyAppWeb, :controller

  def index(conn, _params) do
    pipeline_check = run_pipeline_smoke_test()

    checks = %{
      pipeline: pipeline_check,
      database: check_db()
    }

    status = if pipeline_check.status == "ok", do: 200, else: 503

    conn
    |> put_status(status)
    |> json(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      checks: checks
    })
  end

  defp run_pipeline_smoke_test do
    probe = %{customer_id: :probe, product_id: :probe, quantity: 0}

    case MyApp.PipelineTelemetry.measure(:health_probe, fn ->
      # A probe that returns a controlled result without side effects
      MyApp.OrderPipeline.validate_probe(probe)
    end) do
      {:ok, _} -> %{status: "ok"}
      {:error, :probe_ok} -> %{status: "ok"}
      {:error, reason} -> %{status: "error", reason: inspect(reason)}
    end
  end

  defp check_db do
    case Ecto.Adapters.SQL.query(MyApp.Repo, "SELECT 1", []) do
      {:ok, _} -> %{status: "ok"}
      _ -> %{status: "error"}
    end
  rescue
    _ -> %{status: "error"}
  end
end

Step 7: Create Monitors in Vigilmon

HTTP monitor for the health endpoint:

1. Sign in at vigilmon.online
2. Click New Monitor → HTTP
3. Set URL to https://your-app.example.com/health
4. Set interval to 60 seconds
5. Alert condition: status 200 and body contains "pipeline":{"status":"ok"}

Heartbeat monitor to confirm pipelines are processing:

# lib/my_app/pipeline_heartbeat.ex
defmodule MyApp.PipelineHeartbeat do
  use GenServer

  @heartbeat_url System.get_env("VIGILMON_PIPELINE_HEARTBEAT_URL")
  @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(:ping, state) do
    probe_result = MyApp.PipelineTelemetry.measure(:heartbeat_probe, fn ->
      {:ok, :pipeline_reachable}
    end)

    if match?({:ok, _}, probe_result) do
      ping_vigilmon()
    end

    schedule()
    {:noreply, state}
  end

  defp ping_vigilmon do
    if @heartbeat_url do
      :httpc.request(:get, {String.to_charlist(@heartbeat_url), []}, [{:timeout, 3000}], [])
    end
  end

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

Create a Heartbeat monitor in Vigilmon with an 8-minute expected interval.

Step 8: Alerting

In Vigilmon, configure Notifications → New Channel:

Slack for elevated pipeline error rate:

⚠️ DEGRADED: Pipeline Error Rate — MyApp
Pipeline: order
Error rate: 31% over last 5 minutes (threshold: 25%)
Action: Check order pipeline steps — validate, fetch_customer, check_inventory

PagerDuty for pipeline health endpoint returning 503:

Page on-call immediately if the health endpoint returns non-200 for two consecutive checks.

Grafana alert for step-level errors:

- alert: PipelineStepHighErrorRate
  expr: |
    sum(rate(my_app_pipeline_step_count{outcome="error"}[5m])) by (pipeline, step)
    /
    sum(rate(my_app_pipeline_step_count[5m])) by (pipeline, step)
    > 0.10
  for: 3m
  annotations:
    summary: "High error rate at step {{ $labels.step }} in {{ $labels.pipeline }} pipeline"

Common Exceptional Issues and Fixes

Untagged errors escaping pipelines:

# Wrap bare raise/rescue in railway style at system boundaries
def fetch_from_external(id) do
  {:ok, ExternalAPI.get!(id)}
rescue
  e in RuntimeError -> {:error, {:external_api, e.message}}
end

Error reason too broad for alerting:

# Instead of {:error, :failed}, be specific
{:error, {:payment_gateway, :timeout, gateway_response}}
# This lets you group alerts by gateway vs. validation vs. inventory

Missing fallback on optional steps:

# Use ||| to provide a default when a step is non-critical
order_params
~> enrich_with_recommendations()
||| fn _ -> {:ok, Map.put(order_params, :recommendations, [])} end
~> create_order()

Long pipelines hiding the first error:

# Log the error at the step level, not just at the pipeline boundary
defp fetch_customer(%{customer_id: id} = params) do
  case MyApp.Customers.get(id) do
    nil ->
      Logger.warning("Customer not found in order pipeline", customer_id: id)
      {:error, {:not_found, :customer, id}}
    customer ->
      {:ok, Map.put(params, :customer, customer)}
  end
end

What You've Built

| What | How | |------|-----| | Instrumented pipeline wrapper | Telemetry on every pipeline invocation with outcome and duration | | Per-step metrics | Counter and distribution for each step outcome | | Rolling error rate tracker | GenServer that alerts when error rate crosses threshold | | Structured health endpoint | JSON health response reporting pipeline status | | Liveness heartbeat | GenServer pinging Vigilmon after successful pipeline probe | | Real-time alerting | Vigilmon HTTP monitor + PagerDuty on health degradation |

Exceptional makes Elixir error handling readable and composable. Vigilmon makes those error flows visible — so you know exactly which step in which pipeline is failing, and you hear about it before your users do.

Next Steps

• Add distributed tracing (OpenTelemetry) to correlate pipeline executions with upstream request IDs
• Build per-pipeline Grafana dashboards showing step-by-step funnel conversion and error breakdown
• Use Vigilmon's status page to communicate pipeline degradation to internal stakeholders
• Combine Exceptional with :fuse circuit breakers so pipelines that hit external services trip automatically on exhaustion

Get started free at vigilmon.online.