tutorial

How to Monitor DeltaCRDT with Vigilmon

Monitor DeltaCRDT health, delta sync rates, CRDT state size, and cross-node convergence in your Elixir distributed application with Vigilmon heartbeat and HTTP monitors.

How to Monitor DeltaCRDT with Vigilmon

DeltaCRDT is a pure Elixir implementation of delta-state Conflict-free Replicated Data Types. It lets you build eventually-consistent distributed data structures — counters, sets, maps, and registers — that merge correctly without coordination between nodes. Each replica maintains its own state and propagates compact deltas rather than full state, making it bandwidth-efficient for large distributed clusters.

When DeltaCRDT state fails to converge, your distributed application silently diverges. One node believes a value is 1,200; another believes it is 800. A key added on node A never appears on node B. These failures don't raise exceptions — your processes are alive, your HTTP endpoints return 200, and your supervisor trees are green. The data is just wrong.

Vigilmon heartbeat monitors let you verify that DeltaCRDT state is converging across your cluster at each sync cycle, not just that the processes are alive.

Why Monitor DeltaCRDT?

DeltaCRDT failures are almost always data consistency failures, not process failures:

  • Delta sync stall — the periodic delta sync task (:sync_interval) may be misconfigured or the underlying transport may drop messages; deltas accumulate without propagating
  • State divergence after network partition — each partition continues accepting mutations; when the partition heals, the merge must reconcile all diverged deltas; this merge backlog can cause temporary state bloat
  • Garbage collection lag — DeltaCRDT uses causal contexts to track which deltas have been applied; if garbage collection falls behind, the causal context grows and serialization becomes expensive
  • Process crash without state transfer — a node crash before state is replicated to a peer means the delta is lost; depending on your CRDT type, this may be unrecoverable without application-level replay
  • Neighbor list staleness — DeltaCRDT requires explicit configuration of which neighbors to sync with; if your cluster membership changes (nodes join or leave) but the neighbor list is not updated, new nodes never receive deltas

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Delta sync call frequency | Whether the sync timer is firing at the expected rate | | CRDT state byte size | Growing size indicates GC lag or accumulation of unmerged deltas | | Sync neighbor count | Whether the configured neighbor set matches the live cluster | | Cross-node state hash | Whether replicas have converged to the same value | | Merge duration | Time spent processing incoming deltas; spikes indicate large state or CPU pressure | | GC runs per minute | Whether causal context cleanup is keeping up |

Step 1: Add DeltaCRDT to Your Application

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

# lib/my_app/distributed_store.ex
defmodule MyApp.DistributedStore do
  use Supervisor

  def start_link(opts), do: Supervisor.start_link(__MODULE__, opts, name: __MODULE__)

  def init(_opts) do
    neighbors = Application.get_env(:my_app, :crdt_neighbors, [])

    children = [
      {DeltaCrdt, [
        crdt: DeltaCrdt.AWLWWMap,
        name: MyApp.CrdtStore,
        sync_interval: 200,
        max_sync_size: :infinite,
        neighbours: neighbors
      ]}
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end
end

# lib/my_app/application.ex
children = [
  MyApp.DistributedStore,
  # ...
]

Step 2: Build a CRDT Health Check

# lib/my_app/crdt_health.ex
defmodule MyApp.CrdtHealth do
  @probe_key "__health_probe__"

  def check do
    pid = Process.whereis(MyApp.CrdtStore)

    if is_nil(pid) do
      %{status: :error, reason: "CrdtStore process not found"}
    else
      memory = Process.info(pid, :memory) |> elem(1)
      state = DeltaCrdt.to_map(MyApp.CrdtStore)
      entry_count = map_size(state)

      write_result = probe_write()

      %{
        status: if(write_result == :ok, do: :ok, else: :error),
        pid: inspect(pid),
        memory_bytes: memory,
        entry_count: entry_count,
        write_probe: write_result
      }
    end
  rescue
    e -> %{status: :error, reason: inspect(e)}
  end

  defp probe_write do
    DeltaCrdt.put(MyApp.CrdtStore, @probe_key, :erlang.system_time(:millisecond))
    Process.sleep(50)
    result = DeltaCrdt.get(MyApp.CrdtStore, @probe_key)

    if is_integer(result) do
      DeltaCrdt.delete(MyApp.CrdtStore, @probe_key)
      :ok
    else
      :write_not_readable
    end
  rescue
    e -> {:error, inspect(e)}
  end
end

Step 3: Check Cross-Node Convergence

For multi-node deployments, verify that all nodes have converged to the same state:

# lib/my_app/crdt_consistency.ex
defmodule MyApp.CrdtConsistency do
  @convergence_key "__convergence_probe__"
  @tolerance_ms 5_000

  def check do
    nodes = [Node.self() | Node.list()]

    if length(nodes) < 2 do
      {:ok, :single_node}
    else
      probe_value = :erlang.system_time(:millisecond)
      DeltaCrdt.put(MyApp.CrdtStore, @convergence_key, probe_value)

      Process.sleep(500)

      remote_values =
        Enum.map(Node.list(), fn node ->
          case :rpc.call(node, DeltaCrdt, :get, [MyApp.CrdtStore, @convergence_key], 3_000) do
            {:badrpc, reason} -> {:error, reason}
            value -> {:ok, value}
          end
        end)

      errors = Enum.filter(remote_values, fn {tag, _} -> tag == :error end)
      values = Enum.filter(remote_values, fn {tag, _} -> tag == :ok end)
               |> Enum.map(fn {:ok, v} -> v end)

      all_match = Enum.all?(values, fn v -> v == probe_value end)

      DeltaCrdt.delete(MyApp.CrdtStore, @convergence_key)

      cond do
        errors != [] -> {:degraded, %{errors: errors, replicated: length(values), total: length(nodes)}}
        not all_match -> {:degraded, %{reason: :values_diverged, values: values}}
        true -> {:ok, %{nodes: length(nodes), all_converged: true}}
      end
    end
  end
end

Step 4: Expose a Health Endpoint

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

  def index(conn, _params) do
    crdt = MyApp.CrdtHealth.check()
    consistency = MyApp.CrdtConsistency.check()

    crdt_ok = crdt.status == :ok
    consistency_ok = elem(consistency, 0) == :ok

    status = if crdt_ok and consistency_ok, do: 200, else: 503

    conn
    |> put_status(status)
    |> json(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      node: Node.self() |> to_string(),
      crdt: crdt,
      consistency: format_consistency(consistency)
    })
  end

  defp format_consistency({:ok, :single_node}), do: %{status: "ok", mode: "single_node"}
  defp format_consistency({:ok, data}), do: Map.merge(%{status: "ok"}, data)
  defp format_consistency({:degraded, data}), do: Map.merge(%{status: "degraded"}, data)
end

# lib/my_app_web/router.ex
scope "/", MyAppWeb do
  get "/health", HealthController, :index
end

Step 5: Emit Telemetry Metrics

# lib/my_app/crdt_poller.ex
defmodule MyApp.CrdtPoller do
  use GenServer
  require Logger

  @interval :timer.seconds(30)

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

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

  def handle_info(:poll, state) do
    report()
    schedule()
    {:noreply, state}
  end

  defp report do
    pid = Process.whereis(MyApp.CrdtStore)

    if pid do
      memory = Process.info(pid, :memory) |> elem(1)
      entry_count = DeltaCrdt.to_map(MyApp.CrdtStore) |> map_size()
      neighbor_count = length(Application.get_env(:my_app, :crdt_neighbors, []))

      :telemetry.execute(
        [:my_app, :crdt, :state],
        %{memory_bytes: memory, entry_count: entry_count, neighbor_count: neighbor_count},
        %{node: Node.self()}
      )
    else
      Logger.warning("CrdtPoller: MyApp.CrdtStore not running")
    end
  rescue
    e -> Logger.warning("CrdtPoller error: #{inspect(e)}")
  end

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

Step 6: Create Monitors in Vigilmon

HTTP monitor for the health endpoint:

  1. Sign in at vigilmon.online
  2. Click New Monitor → HTTP
  3. URL: https://your-app.example.com/health
  4. Interval: 60 seconds
  5. Alert condition: non-200 or body contains "status":"degraded"

Heartbeat monitor for CRDT liveness:

# lib/my_app/crdt_heartbeat.ex
defmodule MyApp.CrdtHeartbeat do
  use GenServer

  @interval :timer.minutes(2)

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

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

  def handle_info(:ping, state) do
    check = MyApp.CrdtHealth.check()

    if check.status == :ok do
      url = System.get_env("VIGILMON_CRDT_HEARTBEAT_URL")

      if url do
        :httpc.request(:get, {String.to_charlist(url), []}, [{:timeout, 5_000}], [])
      end
    end

    schedule()
    {:noreply, state}
  end

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

In Vigilmon, create a Heartbeat monitor with a 5-minute expected interval. Set VIGILMON_CRDT_HEARTBEAT_URL in your release environment.

Step 7: Alerting

In Vigilmon, go to Notifications → New Channel and configure:

Slack for CRDT degradation:

🔴 DOWN: DeltaCRDT — MyApp
Monitor: /health returning crdt.status = "degraded"
Node: myapp@10.0.0.1
Action: Check CrdtStore process, neighbor list, and network partition status

PagerDuty for heartbeat miss — a missed heartbeat means the CRDT store process has crashed or write/read is failing; this warrants immediate attention as all distributed state updates are being dropped.

Common DeltaCRDT Issues and Fixes

Stale neighbor list after cluster resize:

# Update neighbors dynamically when cluster membership changes
def handle_info({:nodeup, node}, state) do
  new_neighbors = [node | state.neighbors]
  DeltaCrdt.set_neighbours(MyApp.CrdtStore, new_neighbors)
  {:noreply, %{state | neighbors: new_neighbors}}
end

def handle_info({:nodedown, node}, state) do
  new_neighbors = List.delete(state.neighbors, node)
  DeltaCrdt.set_neighbours(MyApp.CrdtStore, new_neighbors)
  {:noreply, %{state | neighbors: new_neighbors}}
end

Large state after partition reconciliation:

# From IEx, check current state size
iex> MyApp.CrdtStore |> DeltaCrdt.to_map() |> map_size()
# If far above expected, check for deleted-but-not-GC'd entries
# Reduce sync_interval temporarily to speed up delta propagation

Slow merge on large maps:

# Prefer AWLWWMap for mutable key/value data;
# use GSet for append-only sets to avoid tombstone accumulation
{DeltaCrdt, [crdt: DeltaCrdt.GSet, name: MyApp.AppendStore, sync_interval: 100]}

What You've Built

| What | How | |------|-----| | CRDT process health | Verify CrdtStore process alive and write/read cycle works | | Cross-node convergence check | RPC probe verifying all nodes converged on written value | | Structured health endpoint | JSON health response with CRDT and consistency status | | State size monitoring | Telemetry on entry count and memory per 30s | | Liveness heartbeat | GenServer pinging Vigilmon only when CRDT is healthy | | Real-time alerting | Vigilmon HTTP monitor + PagerDuty on heartbeat miss |

DeltaCRDT makes coordination-free distributed state possible. Vigilmon makes sure the eventual consistency is actually converging.

Next Steps

  • Add per-CRDT-type monitors if your application uses multiple CRDT instances for different data domains
  • Alert on neighbor count dropping below the expected cluster size to catch node siloing early
  • Use Vigilmon response time history to track health endpoint latency as CRDT state grows
  • Set up a status page combining CRDT health with your application's other distributed components

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