fix clustering problems with singleton processes

config/dev.exs

@@ -16,6 +16,30 @@ config :teiserver, Teiserver.SpringTcpServer,
   heartbeat_interval: nil,
   heartbeat_timeout: nil
 
+raw_http_port = System.get_env("TEISERVER_HTTP_PORT", "4002") |> String.to_integer()
+
+# Secondary dev nodes should disable the asset watchers, otherwise both nodes
+# race to write the same asset output files (TEISERVER_WATCHERS=off)
+watchers =
+  if System.get_env("TEISERVER_WATCHERS") == "off" do
+    []
+  else
+    [
+      esbuild: {Esbuild, :install_and_run, [:teiserver, ~w(--sourcemap=inline --watch)]},
+      tailwind: {Tailwind, :install_and_run, [:teiserver, ~w(--watch)]},
+      dark_sass: {
+        DartSass,
+        :install_and_run,
+        [:dark, ~w(--embed-source-map --source-map-urls=absolute --watch)]
+      },
+      light_sass: {
+        DartSass,
+        :install_and_run,
+        [:light, ~w(--embed-source-map --source-map-urls=absolute --watch)]
+      }
+    ]
+  end
+
 # For development, we disable any cache and enable
 # debugging and code reloading.
 #
@@ -24,23 +48,10 @@ config :teiserver, Teiserver.SpringTcpServer,
 # with webpack to recompile .js and .css sources.
 config :teiserver, TeiserverWeb.Endpoint,
   url: [host: "localhost"],
-  http: [ip: {0, 0, 0, 0}, port: 4000],
-  watchers: [
-    esbuild: {Esbuild, :install_and_run, [:teiserver, ~w(--sourcemap=inline --watch)]},
-    tailwind: {Tailwind, :install_and_run, [:teiserver, ~w(--watch)]},
-    dark_sass: {
-      DartSass,
-      :install_and_run,
-      [:dark, ~w(--embed-source-map --source-map-urls=absolute --watch)]
-    },
-    light_sass: {
-      DartSass,
-      :install_and_run,
-      [:light, ~w(--embed-source-map --source-map-urls=absolute --watch)]
-    }
-  ]
+  http: [ip: {0, 0, 0, 0}, port: raw_http_port],
+  watchers: watchers
 
-config :teiserver, Teiserver.OAuth, issuer: "http://localhost:4000"
+config :teiserver, Teiserver.OAuth, issuer: "http://localhost:#{raw_http_port}"
 
 config :teiserver, Teiserver,
   certs: [

config/runtime.exs

@@ -254,3 +254,37 @@ end
 
 config :teiserver, TeiserverWeb.Monitoring,
   port: Teiserver.ConfigHelpers.get_env("TEI_METRICS_SERVER_PORT", 4001, :int)
+
+# Each test partition gets its own database and ports so that
+# `mix test --partitions N` instances can run concurrently
+if config_env() == :test do
+  partition = System.get_env("MIX_TEST_PARTITION", "0")
+  partition_port = String.to_integer(partition) * 100
+
+  config :teiserver, Teiserver.Repo, database: "teiserver_test#{partition}"
+
+  http_port = 4002 + partition_port
+
+  config :teiserver, TeiserverWeb.Endpoint, http: [port: http_port]
+  config :teiserver, Teiserver.OAuth, issuer: "http://localhost:#{http_port}"
+
+  config :teiserver, TeiserverWeb.Monitoring,
+    port: Teiserver.ConfigHelpers.get_env("TEI_METRICS_SERVER_PORT", 4001, :int) + partition_port
+end
+
+# Cluster formation, requires the node to run in distributed mode
+# (--name/--sname). Topologies are ignored when not distributed.
+is_local? = config_env() == :dev
+is_prod? = config_env() == :prod
+
+if is_local? do
+  # Connects to every distributed node registered with the local epmd,
+  # useful for local development (see the justfile)
+  config :libcluster,
+    topologies: [teiserver: [strategy: Cluster.Strategy.LocalEpmd]]
+end
+
+if is_prod? do
+  # TODO: set libcluster topology for prod, e.g. Cluster.Strategy.Epmd with a
+  # static host list or Cluster.Strategy.DNSPoll for discovery
+end

justfile

@@ -0,0 +1,14 @@
+
+# List available recipes
+default:
+    @just --list
+
+# Run a dev cluster node, e.g. `just iex` or `just iex 2`
+# Instance 1: http 4002, spring tcp 8200, metrics 4001; each further
+# instance is offset by 100. Asset watchers only run on instance 1.
+node instance='1':
+    TEISERVER_HTTP_PORT=$((4002 + ({{instance}} - 1) * 100)) \
+    TEI_SPRING_TCP_PORT=$((8200 + ({{instance}} - 1) * 100)) \
+    TEI_METRICS_SERVER_PORT=$((4001 + ({{instance}} - 1) * 100)) \
+    {{ if instance == '1' { '' } else { 'TEISERVER_WATCHERS=off' } }} \
+    iex --name node{{instance}}@127.0.0.1 --cookie teiserver-dev -S mix phx.server

lib/teiserver/application.ex

@@ -25,9 +25,19 @@ defmodule Teiserver.Application do
     LoggerBackends.add({LoggerFileBackend, :info_log})
     Logger.add_handlers(:teiserver)
 
+    # Topologies are only usable when running in distributed mode, otherwise
+    # libcluster logs a warning on every reconnect attempt
+    cluster_topologies =
+      if Node.alive?() do
+        Application.get_env(:libcluster, :topologies, [])
+      else
+        []
+      end
+
     children =
       [
         Teiserver.PromEx,
+        {Cluster.Supervisor, [cluster_topologies, [name: Teiserver.ClusterSupervisor]]},
         # Migrations
         {Ecto.Migrator,
          repos: Application.fetch_env!(:teiserver, :ecto_repos),
@@ -78,8 +88,12 @@ defmodule Teiserver.Application do
         {Horde.Registry, [keys: :unique, members: :auto, name: Teiserver.LobbyRegistry]},
         {Horde.Registry, [keys: :unique, members: :auto, name: Teiserver.ClientRegistry]},
         {Horde.Registry, [keys: :unique, members: :auto, name: Teiserver.PartyRegistry]},
-        {Horde.Registry, [keys: :unique, members: :auto, name: Teiserver.QueueWaitRegistry]},
-        {Horde.Registry, [keys: :unique, members: :auto, name: Teiserver.QueueMatchRegistry]},
+
+        # Cluster-wide singletons (CoordinatorServer, MatchMonitorServer,
+        # AutomodServer, LobbyIdServer); children of a dead node are restarted
+        # on a surviving node
+        {Horde.DynamicSupervisor,
+         [strategy: :one_for_one, members: :auto, name: Teiserver.SingletonSupervisor]},
 
         # These are for tracking the number of servers on the local node
         {Registry, keys: :duplicate, name: Teiserver.LocalPoolRegistry},
@@ -221,6 +235,7 @@ defmodule Teiserver.Application do
   @impl Application
   @spec prep_stop(map()) :: map()
   def prep_stop(state) do
+    # TODO - PubSub doesn't work here because the endpoint is already stopped
     PubSub.broadcast(
       Teiserver.PubSub,
       "application",

lib/teiserver/battle/servers/match_monitor_server.ex

@@ -3,6 +3,7 @@ defmodule Teiserver.Battle.MatchMonitorServer do
   The server used to monitor the autohosts and get data from them
   """
 
+  alias Horde.DynamicSupervisor, as: HordeSupervisor
   alias Phoenix.PubSub
   alias Teiserver.Account
   alias Teiserver.Account.Auth
@@ -22,27 +23,34 @@ defmodule Teiserver.Battle.MatchMonitorServer do
   alias Teiserver.Telemetry
 
   use Plugins
-  use GenServer
+  use GenServer, restart: :transient
 
   require Logger
 
   import Teiserver.Helper.NumberHelper, only: [int_parse: 1]
 
   @spec do_start() :: :ok
   def do_start do
-    # Start the supervisor server
-    {:ok, _monitor_pid} =
-      DynamicSupervisor.start_child(Teiserver.Coordinator.DynamicSupervisor, {
+    # Start the singleton server, another node may have already done so
+    result =
+      HordeSupervisor.start_child(Teiserver.SingletonSupervisor, {
         Teiserver.Battle.MatchMonitorServer,
         name: Teiserver.Battle.MatchMonitorServer, data: %{}
       })
 
-    :ok
+    case result do
+      {:ok, _monitor_pid} -> :ok
+      {:error, {:already_started, _pid}} -> :ok
+    end
   end
 
   @spec start_link(list()) :: :ignore | {:error, any} | {:ok, pid}
   def start_link(opts) do
-    GenServer.start_link(__MODULE__, opts[:data], [])
+    GenServer.start_link(__MODULE__, opts[:data], name: via_tuple())
+  end
+
+  defp via_tuple do
+    {:via, Horde.Registry, {Teiserver.ServerRegistry, "MatchMonitorServer", :match_monitor}}
   end
 
   @spec get_match_monitor_userid() :: User.id() | nil
@@ -375,6 +383,12 @@ defmodule Teiserver.Battle.MatchMonitorServer do
     {:noreply, state}
   end
 
+  # Another node won the singleton registration during a Horde registry
+  # merge; step down cleanly so the :transient restart does not respawn us
+  def handle_info({:EXIT, _from, {:name_conflict, _key, _registry, _winning_pid}}, state) do
+    {:stop, :normal, state}
+  end
+
   # Catchall handle_info
   def handle_info(msg, state) do
     Logger.warning(
@@ -542,12 +556,6 @@ defmodule Teiserver.Battle.MatchMonitorServer do
     send(self(), :begin)
     Logger.metadata(request_id: "MatchMonitorServer")
 
-    Horde.Registry.register(
-      Teiserver.ServerRegistry,
-      "MatchMonitorServer",
-      :match_monitor
-    )
-
     {:ok, %{}}
   end
 

lib/teiserver/coordinator.ex

@@ -10,8 +10,9 @@ defmodule Teiserver.Coordinator do
   def start_coordinator do
     case get_coordinator_pid() do
       nil ->
-        # Start the supervisor server
-        DynamicSupervisor.start_child(Teiserver.Coordinator.DynamicSupervisor, {
+        # Start the singleton server, another node may win the race and
+        # start it first
+        Horde.DynamicSupervisor.start_child(Teiserver.SingletonSupervisor, {
           Teiserver.Coordinator.CoordinatorServer,
           name: Teiserver.Coordinator.CoordinatorServer, data: %{}
         })

lib/teiserver/coordinator/automod_server.ex

@@ -1,6 +1,7 @@
 defmodule Teiserver.Coordinator.AutomodServer do
   @moduledoc false
 
+  alias Horde.DynamicSupervisor, as: HordeSupervisor
   alias Phoenix.PubSub
   alias Teiserver.Account
   alias Teiserver.Account.Auth
@@ -9,7 +10,7 @@ defmodule Teiserver.Coordinator.AutomodServer do
   alias Teiserver.Config
   alias Teiserver.Coordinator
   alias Teiserver.Moderation
-  use GenServer
+  use GenServer, restart: :transient
   require Logger
   import Teiserver.Logging.Helpers, only: [add_audit_log: 4]
 
@@ -33,18 +34,28 @@ defmodule Teiserver.Coordinator.AutomodServer do
 
   @spec do_start() :: :ok
   defp do_start do
-    {:ok, _automod_pid} =
-      DynamicSupervisor.start_child(Teiserver.Coordinator.DynamicSupervisor, {
+    # Another node may have already started the singleton
+    result =
+      HordeSupervisor.start_child(Teiserver.SingletonSupervisor, {
         Teiserver.Coordinator.AutomodServer,
         name: Teiserver.Coordinator.AutomodServer, data: %{}
       })
 
+    case result do
+      {:ok, _automod_pid} -> :ok
+      {:error, {:already_started, _pid}} -> :ok
+    end
+
     :ok
   end
 
   @spec start_link(list()) :: :ignore | {:error, any} | {:ok, pid}
   def start_link(opts) do
-    GenServer.start_link(__MODULE__, opts[:data], [])
+    GenServer.start_link(__MODULE__, opts[:data], name: via_tuple())
+  end
+
+  defp via_tuple do
+    {:via, Horde.Registry, {Teiserver.ServerRegistry, "AutomodServer", :automod}}
   end
 
   def handle_info(:begin, state) do
@@ -110,6 +121,12 @@ defmodule Teiserver.Coordinator.AutomodServer do
 
   def handle_info(%{channel: "telemetry_user_properties"}, state), do: {:noreply, state}
 
+  # Another node won the singleton registration during a Horde registry
+  # merge; step down cleanly so the :transient restart does not respawn us
+  def handle_info({:EXIT, _from, {:name_conflict, _key, _registry, _winning_pid}}, state) do
+    {:stop, :normal, state}
+  end
+
   # Catchall handle_info
   def handle_info(msg, state) do
     Logger.error("AutoMod handle_info error. No handler for msg of #{Kernel.inspect(msg)}")
@@ -123,11 +140,7 @@ defmodule Teiserver.Coordinator.AutomodServer do
 
   @spec init(map()) :: {:ok, map()}
   def init(_opts) do
-    Horde.Registry.register(
-      Teiserver.ServerRegistry,
-      "AutomodServer",
-      :automod
-    )
+    Process.flag(:trap_exit, true)
 
     :timer.send_after(500, :begin)
     {:ok, %{}}

lib/teiserver/coordinator/coordinator_server.ex

@@ -19,7 +19,7 @@ defmodule Teiserver.Coordinator.CoordinatorServer do
   alias Teiserver.Room
   alias Teiserver.Telemetry
 
-  use GenServer
+  use GenServer, restart: :transient
 
   require Logger
 
@@ -32,7 +32,11 @@ defmodule Teiserver.Coordinator.CoordinatorServer do
 
   @spec start_link(list()) :: :ignore | {:error, any} | {:ok, pid}
   def start_link(opts) do
-    GenServer.start_link(__MODULE__, opts[:data], [])
+    GenServer.start_link(__MODULE__, opts[:data], name: via_tuple())
+  end
+
+  defp via_tuple do
+    {:via, Horde.Registry, {Teiserver.ServerRegistry, "CoordinatorServer", :coordinator}}
   end
 
   @impl GenServer
@@ -310,6 +314,12 @@ defmodule Teiserver.Coordinator.CoordinatorServer do
     {:noreply, state}
   end
 
+  # Another node won the singleton registration during a Horde registry
+  # merge; step down cleanly so the :transient restart does not respawn us
+  def handle_info({:EXIT, _from, {:name_conflict, _key, _registry, _winning_pid}}, state) do
+    {:stop, :normal, state}
+  end
+
   # Catchall handle_info
   def handle_info(msg, state) do
     Logger.error(
@@ -369,12 +379,6 @@ defmodule Teiserver.Coordinator.CoordinatorServer do
   def init(_opts) do
     Process.flag(:trap_exit, true)
 
-    Horde.Registry.register(
-      Teiserver.ServerRegistry,
-      "CoordinatorServer",
-      :coordinator
-    )
-
     send(self(), :begin)
     {:ok, %{client: %{}}}
   end