Add distributed Erlang clustering for single APRS-IS connection

Implement leader election to ensure only one APRS-IS connection across
multiple Kubernetes replicas. This prevents duplicate packet processing
and respects APRS-IS usage policies.

Key changes:
- Add leader election using :global registry
- Create connection manager for dynamic APRS-IS management
- Implement packet distribution from leader to all nodes
- Add Kubernetes headless service for node discovery
- Configure DNS-based clustering with libcluster
- Update deployment to support 3 replicas with clustering

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

CLAUDE.md

@@ -127,6 +127,7 @@ The application supports Kubernetes deployment with manifests in `k8s/` director
 The app is deployed in a k3s cluster with the following structure:
 - **App name**: `aprs`
 - **Namespace**: `aprs`
+- **K8s manifests location**: `~/dev/infra/clusters/aprs/`
 
 Common kubectl commands for debugging:
 ```bash
@@ -150,4 +151,37 @@ kubectl rollout status deployment/aprs -n aprs
 
 # Execute commands in the pod
 kubectl exec -it deployment/aprs -n aprs -- /app/bin/aprsme remote
-```
+
+# Check cluster membership
+kubectl exec -it <pod-name> -n aprs -- /app/bin/aprsme eval "Node.list()"
+
+# Check leader status
+kubectl exec -it <pod-name> -n aprs -- /app/bin/aprsme eval "Aprsme.Cluster.LeaderElection.is_leader?()"
+```
+
+### Clustering Architecture
+
+The application uses distributed Erlang clustering to ensure only one APRS-IS connection across multiple replicas:
+
+1. **Leader Election**: Uses `:global` registry for distributed leader election
+   - Only the elected leader maintains the APRS-IS connection
+   - Automatic failover when leader goes down
+   - Leader election managed by `Aprsme.Cluster.LeaderElection`
+
+2. **Connection Management**: 
+   - `Aprsme.Cluster.ConnectionManager` starts/stops APRS-IS based on leadership
+   - Uses `DynamicSupervisor` to manage connection lifecycle
+   - Prevents duplicate connections and packet processing
+
+3. **Kubernetes Configuration**:
+   - Headless service (`aprs-headless`) for node discovery
+   - DNS-based clustering via libcluster
+   - Environment variables:
+     - `CLUSTER_ENABLED=true` - Enables clustering
+     - `RELEASE_NODE` - Erlang node name
+     - `RELEASE_COOKIE` - Erlang distribution cookie
+
+4. **Deployment**:
+   - Default replicas: 3 (configurable in `aprs-deployment.yaml`)
+   - Only leader processes APRS packets
+   - All nodes serve web traffic

config/runtime.exs

@@ -46,6 +46,9 @@ if config_env() == :prod do
   host = System.get_env("PHX_HOST") || "example.com"
   port = String.to_integer(System.get_env("PORT") || "4000")
 
+  # Configure clustering based on environment
+  cluster_enabled = System.get_env("CLUSTER_ENABLED", "false") == "true"
+
   # ## Configuring the mailer
   #
   # Configure Resend for email delivery
@@ -116,6 +119,8 @@ if config_env() == :prod do
       "https://tile.openstreetmap.org"
     ]
 
+  config :aprsme, :cluster_enabled, cluster_enabled
+
   # Optional: Set the default "from" email address
   config :aprsme,
     default_from_email: "w5isp@aprs.me"
@@ -134,19 +139,35 @@ if config_env() == :prod do
   config :hammer,
     backend: {Hammer.Backend.ETS, [expiry_ms: 60_000 * 60 * 4, cleanup_interval_ms: 60_000 * 10]}
 
-  # Configure libcluster for Dokku clustering
+  # Configure libcluster topology based on environment
-  config :libcluster,
+  if cluster_enabled do
-    topologies: [
+    config :libcluster,
-      dokku: [
+      topologies: [
-        strategy: Cluster.Strategy.Gossip,
+        kubernetes: [
-        config: [
+          strategy: Cluster.Strategy.Kubernetes.DNS,
-          port: 45_892,
+          config: [
-          if_addr: "0.0.0.0",
+            service: "aprs-headless",
-          multicast_addr: "230.1.1.1",
+            application_name: "aprs",
-          multicast_ttl: 1
+            namespace: System.get_env("POD_NAMESPACE", "aprs"),
+            polling_interval: 10_000
+          ]
         ]
       ]
-    ]
+  else
+    # Keep existing Dokku configuration for non-Kubernetes deployments
+    config :libcluster,
+      topologies: [
+        dokku: [
+          strategy: Cluster.Strategy.Gossip,
+          config: [
+            port: 45_892,
+            if_addr: "0.0.0.0",
+            multicast_addr: "230.1.1.1",
+            multicast_ttl: 1
+          ]
+        ]
+      ]
+  end
 
   #
   # For this example you need include a HTTP client required by Swoosh API client.

lib/aprsme/application.ex

@@ -16,8 +16,6 @@ defmodule Aprsme.Application do
       config: %{metadata: [:file, :line]}
     })
 
-    topologies = Application.get_env(:libcluster, :topologies) || []
-
     children = [
       # Start the Telemetry supervisor
       AprsmeWeb.Telemetry,
@@ -49,7 +47,8 @@ defmodule Aprsme.Application do
       # Start a worker by calling: Aprsme.Worker.start_link(arg)
       # {Aprsme.Worker, arg}
       {Registry, keys: :duplicate, name: Registry.PubSub, partitions: System.schedulers_online()},
-      {Cluster.Supervisor, [topologies, [name: Aprsme.ClusterSupervisor]]},
+      # Start clustering
+      Aprsme.Cluster.Topology,
       # Start Oban for background jobs
       {Oban, :aprsme |> Application.get_env(Oban, []) |> Keyword.put(:queues, default: 10, maintenance: 2)},
       Aprsme.Presence,
@@ -58,6 +57,7 @@ defmodule Aprsme.Application do
       Aprsme.PacketPipelineSupervisor
     ]
 
+    children = maybe_add_cluster_components(children)
     children = maybe_add_is_supervisor(children, Application.get_env(:aprsme, :env))
     children = maybe_add_aprs_connection(children, Application.get_env(:aprsme, :env))
 
@@ -99,10 +99,30 @@ defmodule Aprsme.Application do
       :ok
   end
 
+  defp maybe_add_cluster_components(children) do
+    if Application.get_env(:aprsme, :cluster_enabled, false) do
+      children ++
+        [
+          # Dynamic supervisor for processes managed by cluster leader
+          Aprsme.DynamicSupervisor,
+          # Leader election process
+          Aprsme.Cluster.LeaderElection,
+          # Connection manager that starts/stops APRS-IS based on leadership
+          Aprsme.Cluster.ConnectionManager,
+          # Packet receiver for distributed packets on non-leader nodes
+          Aprsme.Cluster.PacketReceiver
+        ]
+    else
+      children
+    end
+  end
+
   defp maybe_add_is_supervisor(children, env) do
     disable_connection = Application.get_env(:aprsme, :disable_aprs_connection, false)
+    cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
 
-    if env in [:prod, :dev] and not disable_connection do
+    # Only add Is.IsSupervisor directly if clustering is disabled
+    if env in [:prod, :dev] and not disable_connection and not cluster_enabled do
       children ++ [Aprsme.Is.IsSupervisor]
     else
       children

lib/aprsme/cluster/connection_manager.ex

@@ -0,0 +1,87 @@
+defmodule Aprsme.Cluster.ConnectionManager do
+  @moduledoc """
+  Manages the APRS-IS connection based on cluster leadership.
+  Only starts the connection when this node is elected as leader.
+  """
+  use GenServer
+
+  alias Aprsme.Is.IsSupervisor
+
+  require Logger
+
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_opts) do
+    # Subscribe to leadership changes
+    Phoenix.PubSub.subscribe(Aprsme.PubSub, "cluster:leadership")
+
+    # Check initial leadership state
+    Process.send_after(self(), :check_initial_state, 1000)
+
+    {:ok, %{connection_started: false}}
+  end
+
+  @impl true
+  def handle_info(:check_initial_state, state) do
+    if Aprsme.Cluster.LeaderElection.is_leader?() do
+      Logger.info("This node is the leader, starting APRS-IS connection")
+      start_aprs_connection()
+      {:noreply, %{state | connection_started: true}}
+    else
+      Logger.info("This node is not the leader, APRS-IS connection will not be started")
+      {:noreply, state}
+    end
+  end
+
+  @impl true
+  def handle_info({:leadership_change, node, became_leader}, state) do
+    cond do
+      became_leader and node == node() and not state.connection_started ->
+        Logger.info("This node became the leader, starting APRS-IS connection")
+        start_aprs_connection()
+        {:noreply, %{state | connection_started: true}}
+
+      not became_leader and node == node() and state.connection_started ->
+        Logger.info("This node lost leadership, stopping APRS-IS connection")
+        stop_aprs_connection()
+        {:noreply, %{state | connection_started: false}}
+
+      true ->
+        {:noreply, state}
+    end
+  end
+
+  defp start_aprs_connection do
+    # Start the APRS-IS connection supervisor if not already started
+    case DynamicSupervisor.start_child(
+           Aprsme.DynamicSupervisor,
+           {IsSupervisor, []}
+         ) do
+      {:ok, _pid} ->
+        Logger.info("APRS-IS connection started successfully")
+
+      {:error, {:already_started, _pid}} ->
+        Logger.info("APRS-IS connection was already running")
+
+      {:error, reason} ->
+        Logger.error("Failed to start APRS-IS connection: #{inspect(reason)}")
+    end
+  end
+
+  defp stop_aprs_connection do
+    # Find and terminate the APRS-IS connection supervisor
+    children = DynamicSupervisor.which_children(Aprsme.DynamicSupervisor)
+
+    Enum.each(children, fn
+      {_, pid, _, [IsSupervisor]} when is_pid(pid) ->
+        Logger.info("Stopping APRS-IS connection supervisor")
+        DynamicSupervisor.terminate_child(Aprsme.DynamicSupervisor, pid)
+
+      _ ->
+        :ok
+    end)
+  end
+end

lib/aprsme/cluster/leader_election.ex

@@ -0,0 +1,106 @@
+defmodule Aprsme.Cluster.LeaderElection do
+  @moduledoc """
+  Manages leader election for APRS-IS connection using distributed Erlang.
+  Only the elected leader will maintain the APRS-IS connection.
+  """
+  use GenServer
+
+  require Logger
+
+  @election_key {:aprs_is_leader, __MODULE__}
+  @check_interval 5_000
+
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def is_leader? do
+    GenServer.call(__MODULE__, :is_leader?)
+  end
+
+  def current_leader do
+    GenServer.call(__MODULE__, :current_leader)
+  end
+
+  @impl true
+  def init(_opts) do
+    Logger.info("Starting leader election process")
+
+    # Schedule initial election
+    Process.send_after(self(), :attempt_election, 100)
+
+    # Schedule periodic checks
+    Process.send_after(self(), :check_leadership, @check_interval)
+
+    {:ok, %{is_leader: false, leader_node: nil}}
+  end
+
+  @impl true
+  def handle_info(:attempt_election, state) do
+    case :global.register_name(@election_key, self(), &resolve_conflict/3) do
+      :yes ->
+        Logger.info("Elected as APRS-IS connection leader on node #{node()}")
+        notify_leadership_change(true)
+        {:noreply, %{state | is_leader: true, leader_node: node()}}
+
+      :no ->
+        leader_pid = :global.whereis_name(@election_key)
+        leader_node = if leader_pid, do: node(leader_pid)
+        Logger.info("Not elected as leader. Current leader is on node #{inspect(leader_node)}")
+        {:noreply, %{state | is_leader: false, leader_node: leader_node}}
+    end
+  end
+
+  @impl true
+  def handle_info(:check_leadership, state) do
+    # Re-attempt election if we're not leader
+    if not state.is_leader do
+      Process.send_after(self(), :attempt_election, 100)
+    end
+
+    # Schedule next check
+    Process.send_after(self(), :check_leadership, @check_interval)
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_call(:is_leader?, _from, state) do
+    {:reply, state.is_leader, state}
+  end
+
+  @impl true
+  def handle_call(:current_leader, _from, state) do
+    {:reply, state.leader_node, state}
+  end
+
+  @impl true
+  def terminate(reason, state) do
+    if state.is_leader do
+      Logger.info("Leader stepping down due to: #{inspect(reason)}")
+      :global.unregister_name(@election_key)
+      notify_leadership_change(false)
+    end
+
+    :ok
+  end
+
+  # Conflict resolution - prefer the process that's been running longer
+  defp resolve_conflict(_name, pid1, pid2) do
+    info1 = Process.info(pid1, [:registered_name, :current_function])
+    info2 = Process.info(pid2, [:registered_name, :current_function])
+
+    Logger.debug("Resolving leader conflict between #{inspect(info1)} and #{inspect(info2)}")
+
+    # Keep the first registered process
+    pid1
+  end
+
+  defp notify_leadership_change(became_leader) do
+    Phoenix.PubSub.broadcast(
+      Aprsme.PubSub,
+      "cluster:leadership",
+      {:leadership_change, node(), became_leader}
+    )
+  end
+end

lib/aprsme/cluster/packet_distributor.ex

@@ -0,0 +1,38 @@
+defmodule Aprsme.Cluster.PacketDistributor do
+  @moduledoc """
+  Distributes APRS packets from the leader node to all cluster members.
+  This ensures all nodes can serve real-time updates via LiveView while
+  only the leader maintains the APRS-IS connection.
+  """
+  require Logger
+
+  @pubsub_topic "cluster:packets"
+
+  def distribute_packet(packet) do
+    # Only distribute if we're the leader
+    if Aprsme.Cluster.LeaderElection.is_leader?() do
+      # Broadcast to all nodes including self
+      Phoenix.PubSub.broadcast(
+        Aprsme.PubSub,
+        @pubsub_topic,
+        {:distributed_packet, packet}
+      )
+
+      Logger.debug("Distributed packet #{packet.raw} to cluster")
+    end
+  end
+
+  def subscribe do
+    Phoenix.PubSub.subscribe(Aprsme.PubSub, @pubsub_topic)
+  end
+
+  def handle_distributed_packet({:distributed_packet, packet}) do
+    # Broadcast to local LiveView clients
+    Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
+
+    # Update packet store for LiveView
+    AprsmeWeb.MapLive.PacketStore.store_packet(packet)
+
+    Logger.debug("Received distributed packet on node #{node()}")
+  end
+end

lib/aprsme/cluster/packet_receiver.ex

@@ -0,0 +1,41 @@
+defmodule Aprsme.Cluster.PacketReceiver do
+  @moduledoc """
+  Receives distributed packets from the cluster leader on non-leader nodes.
+  Ensures all nodes can serve real-time updates even though only the leader
+  processes APRS packets.
+  """
+  use GenServer
+
+  alias Aprsme.Cluster.PacketDistributor
+
+  require Logger
+
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_opts) do
+    # Subscribe to distributed packets
+    PacketDistributor.subscribe()
+
+    Logger.info("Started packet receiver on node #{node()}")
+
+    {:ok, %{}}
+  end
+
+  @impl true
+  def handle_info({:distributed_packet, packet}, state) do
+    # Only process if we're not the leader (leader already processed locally)
+    if !Aprsme.Cluster.LeaderElection.is_leader?() do
+      PacketDistributor.handle_distributed_packet({:distributed_packet, packet})
+    end
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+end

lib/aprsme/cluster/topology.ex

@@ -0,0 +1,43 @@
+defmodule Aprsme.Cluster.Topology do
+  @moduledoc """
+  Cluster topology configuration for different environments.
+  """
+
+  def child_spec(opts) do
+    cluster_config = config()
+
+    if cluster_config do
+      {Cluster.Supervisor, [cluster_config, opts]}
+    else
+      # Return a no-op spec when clustering is disabled
+      %{
+        id: __MODULE__,
+        start: {__MODULE__, :start_link, [opts]},
+        type: :worker
+      }
+    end
+  end
+
+  def start_link(_opts), do: :ignore
+
+  defp config do
+    if Application.get_env(:aprsme, :cluster_enabled, false) do
+      kubernetes_config()
+    end
+  end
+
+  defp kubernetes_config do
+    [
+      kubernetes: [
+        strategy: Cluster.Strategy.Kubernetes,
+        config: [
+          mode: :dns,
+          kubernetes_node_basename: "aprs",
+          kubernetes_selector: "app=aprs",
+          kubernetes_namespace: "aprs",
+          polling_interval: 10_000
+        ]
+      ]
+    ]
+  end
+end

lib/aprsme/dynamic_supervisor.ex

@@ -0,0 +1,16 @@
+defmodule Aprsme.DynamicSupervisor do
+  @moduledoc """
+  Dynamic supervisor for managing processes that may be started/stopped at runtime.
+  Used primarily for the APRS-IS connection which is managed by cluster leadership.
+  """
+  use DynamicSupervisor
+
+  def start_link(init_arg) do
+    DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_init_arg) do
+    DynamicSupervisor.init(strategy: :one_for_one)
+  end
+end

lib/aprsme/packet_consumer.ex

@@ -251,6 +251,8 @@ defmodule Aprsme.PacketConsumer do
   # Broadcast packets asynchronously to avoid blocking
   defp broadcast_packets_async(packets) do
     Task.start(fn ->
+      cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
+
       Enum.each(packets, fn packet_attrs ->
         # Convert back to a format suitable for broadcasting
         packet = %{
@@ -265,7 +267,13 @@ defmodule Aprsme.PacketConsumer do
           comment: packet_attrs[:comment]
         }
 
-        Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
+        if cluster_enabled do
+          # Use cluster distributor to broadcast to all nodes
+          Aprsme.Cluster.PacketDistributor.distribute_packet(packet)
+        else
+          # Normal single-node broadcasting
+          Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
+        end
       end)
     end)
   end