tutorial

How to Monitor TypeCheck with Vigilmon

Track runtime type contract violations, typespec check failures, and performance overhead of TypeCheck in your Elixir application, and use Vigilmon to alert when type errors indicate data corruption or interface contract drift.

How to Monitor TypeCheck with Vigilmon

TypeCheck is a runtime type checking library for Elixir that validates values against your declared typespecs at the moment they are used, not just at compile time. Where Dialyzer performs static analysis and finds potential issues at compile time, TypeCheck verifies actual runtime values — catching malformed data from external APIs, database rows, or user input that Dialyzer cannot see. When a value violates its declared type, TypeCheck raises a detailed error describing exactly which constraint failed and why.

Runtime type validation is only useful if you know when it fires. A type contract violation that happens once an hour in a background job may be silent for days without monitoring. When TypeCheck starts raising on data that previously conformed — because an upstream API changed its response shape, or because new code paths produce values with unexpected structure — you need to know immediately.

Vigilmon closes that gap: track violation rates by type and call site, alert when violation patterns indicate interface drift, and catch data contract breaks before they cause downstream failures.

Why Monitor TypeCheck at Runtime?

TypeCheck violations in production are a class of error distinct from logic bugs:

  • Interface contract drift — an external service changes its response schema; TypeCheck violations flag this immediately rather than letting malformed data corrupt your database or crash a downstream consumer
  • Data migration side effects — schema migrations that alter existing rows can cause TypeCheck violations in read paths that previously passed — violations surface exactly which old records have the wrong shape
  • Performance overhead visibility — TypeCheck validation adds CPU cost at every check point; without metrics, you can't see whether type checking is impacting latency in hot code paths
  • Violation clustering — a spike in violations may pinpoint a single bad actor (one endpoint, one background job, one upstream) rather than a systemic problem
  • Gradual type erosion — some violations happen rarely and only on edge-case inputs; without counters, you never know they exist

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Violation count by type name | Which type contracts are failing most often | | Violation count by call site (module/function) | Where in your code violations occur | | Violation rate over time | Whether violations are increasing, stable, or isolated | | Check duration distribution | Performance overhead of type checking per code path | | Unique violation signatures | Whether many distinct inputs are failing or one bad input repeats | | Violation-to-call ratio | What fraction of type-checked calls actually violate the contract |

Step 1: Add TypeCheck to Your Project

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

Define types and checked functions:

# lib/my_app/user.ex
defmodule MyApp.User do
  use TypeCheck

  @type! t :: %__MODULE__{
    id: pos_integer(),
    email: String.t(),
    role: :admin | :user | :guest,
    created_at: DateTime.t()
  }

  defstruct [:id, :email, :role, :created_at]
end

# lib/my_app/accounts.ex
defmodule MyApp.Accounts do
  use TypeCheck

  @spec! get_user(pos_integer()) :: {:ok, MyApp.User.t()} | {:error, :not_found}
  def get_user(id) do
    case MyApp.Repo.get(MyApp.User, id) do
      nil -> {:error, :not_found}
      user -> {:ok, user}
    end
  end

  @spec! create_user(String.t(), :admin | :user | :guest) :: {:ok, MyApp.User.t()} | {:error, term()}
  def create_user(email, role) do
    %MyApp.User{email: email, role: role, created_at: DateTime.utc_now()}
    |> MyApp.Repo.insert()
  end
end

Step 2: Capture TypeCheck Violations with Telemetry

TypeCheck raises TypeCheck.TypeError on violations. Wrap your type-checked calls to capture these:

# lib/my_app/type_check_monitor.ex
defmodule MyApp.TypeCheckMonitor do
  @moduledoc """
  Captures TypeCheck.TypeError exceptions and emits telemetry before re-raising.
  Attach this handler to any process that calls type-checked functions.
  """

  require Logger

  @doc """
  Execute a function and emit telemetry for any TypeCheck violations.
  Re-raises the TypeError so normal error handling still applies.
  """
  def checked(type_name, call_site, fun) do
    start = System.monotonic_time()
    result = fun.()
    duration = System.monotonic_time() - start

    :telemetry.execute(
      [:my_app, :type_check, :call],
      %{duration: duration, count: 1},
      %{type: type_name, call_site: call_site, result: :ok}
    )

    result
  rescue
    e in TypeCheck.TypeError ->
      duration = System.monotonic_time() - start

      violation_summary = summarize_violation(e)

      Logger.warning("TypeCheck violation",
        type: type_name,
        call_site: call_site,
        violation: violation_summary,
        message: Exception.message(e)
      )

      :telemetry.execute(
        [:my_app, :type_check, :violation],
        %{duration: duration, count: 1},
        %{
          type: type_name,
          call_site: call_site,
          violation_kind: violation_summary.kind
        }
      )

      reraise e, __STACKTRACE__
  end

  defp summarize_violation(%TypeCheck.TypeError{} = e) do
    message = Exception.message(e)

    kind =
      cond do
        String.contains?(message, "is not a") -> :wrong_type
        String.contains?(message, "does not match") -> :pattern_mismatch
        String.contains?(message, "is not one of") -> :invalid_enum
        true -> :unknown
      end

    %{kind: kind, raw: message}
  end
end

Usage in your code:

defmodule MyApp.UserService do
  def process_external_user(raw_data) do
    MyApp.TypeCheckMonitor.checked(MyApp.User, "UserService.process_external_user", fn ->
      user = struct!(MyApp.User, raw_data)
      MyApp.Accounts.create_user(user.email, user.role)
    end)
  end
end

Step 3: Configure TypeCheck Error Reporting

For non-critical paths where you want to observe violations without crashing, use TypeCheck's overrides to replace raises with logging:

# config/runtime.exs
if config_env() == :prod do
  config :type_check,
    enable_runtime_checks: true
end

# lib/my_app/soft_type_check.ex
defmodule MyApp.SoftTypeCheck do
  @moduledoc """
  Soft type checking for non-critical paths.
  Logs violations and emits telemetry but returns the unchecked value
  instead of raising, allowing the call to proceed.
  """

  require Logger

  def check(type_module, value, call_site) do
    case TypeCheck.conforms(value, type_module.t()) do
      :ok ->
        :telemetry.execute(
          [:my_app, :type_check, :soft_check],
          %{count: 1},
          %{type: type_module, call_site: call_site, result: :ok}
        )

        {:ok, value}

      {:error, error} ->
        Logger.warning("Soft TypeCheck violation (non-raising)",
          type: type_module,
          call_site: call_site,
          error: inspect(error)
        )

        :telemetry.execute(
          [:my_app, :type_check, :soft_check],
          %{count: 1},
          %{type: type_module, call_site: call_site, result: :violation}
        )

        {:violation, value}
    end
  end
end

Step 4: Define Telemetry Metrics

# lib/my_app/telemetry.ex
def metrics do
  [
    # Total TypeCheck calls (pass + fail)
    counter("my_app.type_check.call.count",
      tags: [:type, :call_site, :result]
    ),

    # Duration of each type check
    distribution("my_app.type_check.call.duration",
      unit: {:native, :microsecond},
      reporter_options: [buckets: [10, 50, 100, 500, 1000, 5000]],
      tags: [:type, :call_site]
    ),

    # Violations (raised TypeErrors)
    counter("my_app.type_check.violation.count",
      tags: [:type, :call_site, :violation_kind]
    ),

    # Soft check outcomes (non-raising observation mode)
    counter("my_app.type_check.soft_check.count",
      tags: [:type, :call_site, :result]
    )
  ]
end

Step 5: Track Violation Rate per Type

Build a rolling violation rate detector to spot type contract erosion:

# lib/my_app/type_violation_tracker.ex
defmodule MyApp.TypeViolationTracker do
  use GenServer
  require Logger

  @window_seconds 600
  @alert_threshold 10
  @poll_interval :timer.seconds(60)

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

  def init(state) do
    :telemetry.attach(
      "type-violation-tracker",
      [:my_app, :type_check, :violation],
      &handle_violation/4,
      nil
    )

    schedule()
    {:ok, state}
  end

  def handle_violation(_event, %{count: 1}, %{type: type, violation_kind: kind}, _) do
    GenServer.cast(__MODULE__, {:record, type, kind, System.system_time(:second)})
  end

  def handle_cast({:record, type, kind, ts}, state) do
    entry = {type, kind, ts}
    {:noreply, %{state | violations: [entry | state.violations]}}
  end

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

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

    recent
    |> Enum.group_by(fn {type, _, _} -> type end)
    |> Enum.each(fn {type, entries} ->
      count = length(entries)

      :telemetry.execute(
        [:my_app, :type_check, :violation_rate],
        %{count: count},
        %{type: type, window_seconds: @window_seconds}
      )

      if count > @alert_threshold do
        Logger.error("TypeCheck violation spike",
          type: type,
          count: count,
          window_seconds: @window_seconds
        )
      end
    end)

    pruned = Enum.filter(state.violations, fn {_, _, ts} -> ts > cutoff end)
    schedule()
    {:noreply, %{state | violations: 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
    type_check_status = check_type_validation()

    checks = %{
      type_check: type_check_status,
      database: check_db()
    }

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

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

  defp check_type_validation do
    probe = %MyApp.User{
      id: 1,
      email: "health@probe.local",
      role: :user,
      created_at: DateTime.utc_now()
    }

    case MyApp.SoftTypeCheck.check(MyApp.User, probe, "health_check") do
      {:ok, _} -> %{status: "ok"}
      {:violation, _} -> %{status: "error", reason: "health probe violated User type contract"}
    end
  rescue
    e -> %{status: "error", reason: Exception.message(e)}
  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 "type_check":{"status":"ok"}

Heartbeat monitor for the type-checking subsystem:

# lib/my_app/type_check_heartbeat.ex
defmodule MyApp.TypeCheckHeartbeat do
  use GenServer

  @heartbeat_url System.get_env("VIGILMON_TYPECHECK_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 = %MyApp.User{
      id: 999,
      email: "heartbeat@probe.local",
      role: :guest,
      created_at: DateTime.utc_now()
    }

    case MyApp.SoftTypeCheck.check(MyApp.User, probe, "heartbeat") do
      {:ok, _} -> ping_vigilmon()
      {:violation, _} -> :skip
    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 TypeCheck violation spike:

⚠️ DEGRADED: TypeCheck Violations — MyApp
Type: MyApp.User
Violations: 23 in last 10 minutes (threshold: 10)
Action: Check User type contract — possible upstream API schema change

PagerDuty for critical type check failure:

Page on-call if User, Order, or Payment type violations exceed 50 per minute — this indicates data corruption risk.

Grafana alert for sustained violation rate:

- alert: TypeCheckViolationSpike
  expr: |
    sum(rate(my_app_type_check_violation_count[5m])) by (type) > 0.5
  for: 2m
  annotations:
    summary: "TypeCheck violations spiking for {{ $labels.type }} — possible interface contract break"

Common TypeCheck Issues and Fixes

Expensive type checks in hot paths:

# TypeCheck validation runs on every call — measure the overhead
# and move checks to boundary entry points rather than inner loops
defmodule MyApp.DataProcessor do
  def process_batch(items) when is_list(items) do
    # Validate once at the boundary
    Enum.each(items, fn item ->
      MyApp.SoftTypeCheck.check(MyApp.Item, item, "DataProcessor.process_batch")
    end)

    # Process without per-item type checking inside the loop
    Enum.map(items, &process_item/1)
  end
end

Violations from external API responses:

# Normalize API responses before type checking to avoid spurious violations
defmodule MyApp.ExternalAPI do
  def fetch_user(id) do
    with {:ok, raw} <- HTTPoison.get("https://api.example.com/users/#{id}"),
         {:ok, body} <- Jason.decode(raw.body),
         normalized = normalize_user(body),
         {:ok, user} <- MyApp.SoftTypeCheck.check(MyApp.User, normalized, "ExternalAPI.fetch_user") do
      {:ok, user}
    end
  end

  defp normalize_user(%{"id" => id, "email" => email, "role" => role}) do
    %MyApp.User{
      id: id,
      email: email,
      role: String.to_existing_atom(role),
      created_at: DateTime.utc_now()
    }
  rescue
    _ -> %{}
  end
end

Violations hiding behind catch-all rescue:

# Don't silence TypeCheck.TypeError in generic rescue blocks
def risky_call(data) do
  process(data)
rescue
  e in TypeCheck.TypeError ->
    # Re-emit telemetry, then decide: re-raise or return error tuple
    MyApp.TypeCheckMonitor.record_violation(e, "risky_call")
    {:error, :type_violation}
  e ->
    {:error, Exception.message(e)}
end

What You've Built

| What | How | |------|-----| | Instrumented type check wrapper | Telemetry on every check with duration and outcome | | Soft check mode | Non-raising type observation for non-critical paths | | Violation rate tracker | Rolling-window counter with GenServer alerting | | Structured health endpoint | JSON health with type validation probe result | | Liveness heartbeat | GenServer pinging Vigilmon after successful type check | | Real-time alerting | Vigilmon HTTP monitor + PagerDuty on violation spikes |

TypeCheck makes Elixir's type guarantees visible at runtime. Vigilmon makes those runtime violations visible to you — before malformed data propagates through your system and causes failures that are much harder to diagnose.

Next Steps

  • Use TypeCheck with Ecto changesets at database boundaries to enforce type contracts on all persisted data
  • Add per-type violation dashboards in Grafana showing violation kinds (wrong type, pattern mismatch, invalid enum) over time
  • Integrate TypeCheck violation telemetry with your distributed tracing system to correlate violations with specific request traces
  • Set up Vigilmon monitors for each external API integration to detect interface drift the moment an upstream schema changes

Get started free at vigilmon.online.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles