Add proper cluster awareness via libcluster

Author: mbklein

app/Dockerfile

@@ -50,7 +50,7 @@ LABEL edu.northwestern.library.app=meadow \
   edu.northwestern.library.stage=runtime
 RUN apt update -qq && apt install -y curl git jq libssl-dev libncurses5-dev
 ENV LANG=C.UTF-8
-EXPOSE 4000 4369
+EXPOSE 4000 4369 9100-9155
 COPY --from=build /app/_build/prod/rel/meadow /app
 WORKDIR /app
 RUN npm install -g @anthropic-ai/claude-code \

app/config/config.exs

@@ -115,6 +115,10 @@ config :meadow, Meadow.Pipeline,
     Actions.FileSetComplete
   ]
 
+config :mime, :types, %{
+  "text/event-stream" => ["event-stream"]
+}
+
 config :ueberauth, Ueberauth, providers: [nusso: {Ueberauth.Strategy.NuSSO, []}]
 
 import_config("#{Mix.env()}.exs")

app/lib/meadow/application.ex

@@ -4,16 +4,25 @@ defmodule Meadow.Application do
   @moduledoc false
 
   use Application
+  use Retry
   alias Meadow.Application.Children
   alias Meadow.Config.Runtime
 
   require Logger
+  require WaitForIt
 
   def start(_type, _args) do
     result =
       Supervisor.start_link(
         [
-          {DynamicSupervisor, max_restarts: 4096, strategy: :one_for_one, name: Meadow.Supervisor}
+          {DynamicSupervisor,
+           max_restarts: 4096, strategy: :one_for_one, name: Meadow.Supervisor},
+          {Horde.Registry, name: Meadow.HordeRegistry, keys: :unique, members: :auto},
+          {Horde.DynamicSupervisor,
+           name: Meadow.HordeSupervisor,
+           distribution_strategy: Horde.UniformQuorumDistribution,
+           strategy: :one_for_one,
+           members: :auto}
         ],
         max_restarts: 4096,
         strategy: :one_for_one,
@@ -26,22 +35,48 @@ defmodule Meadow.Application do
       Runtime.configure!()
     end
 
+    if System.get_env("CLUSTER_ENABLED") == "true" do
+      Logger.info("Starting libcluster")
+      topologies = Application.get_env(:libcluster, :topologies, [])
+
+      DynamicSupervisor.start_child(
+        Meadow.Supervisor,
+        {Cluster.Supervisor, [topologies, [name: Meadow.ClusterSupervisor]]}
+      )
+
+      Logger.info("Waiting for cluster")
+
+      WaitForIt.case_wait Horde.Cluster.members(Meadow.HordeSupervisor),
+        timeout: :timer.seconds(10),
+        interval: :timer.seconds(1) do
+        nodes when length(nodes) > 1 ->
+          Logger.info("Cluster formed with nodes: #{inspect(nodes)}")
+      else
+        Logger.warning("Timeout waiting for cluster formation. Continuing with single node.")
+      end
+
+      Logger.info("Waiting for quorum")
+      Horde.DynamicSupervisor.wait_for_quorum(Meadow.HordeSupervisor, :timer.seconds(10))
+    end
+
+    Logger.info("Starting Anubis MCP server")
+    Application.ensure_all_started(:anubis_mcp)
+
     unless System.get_env("MEADOW_NO_REPO") do
       DynamicSupervisor.start_child(Meadow.Supervisor, Meadow.Repo)
       DynamicSupervisor.start_child(Meadow.Supervisor, Meadow.Repo.Indexing)
       Meadow.Repo.wait_for_connection()
     end
 
-    base_children = [
+    [
       {Phoenix.PubSub, [name: Meadow.PubSub, adapter: Phoenix.PubSub.PG2]},
       Meadow.Telemetry,
       {Registry, keys: :unique, name: Meadow.TaskRegistry}
     ]
+    |> start_children()
 
-    children = base_children ++ Children.specs()
-
-    children
-    |> Enum.each(&DynamicSupervisor.start_child(Meadow.Supervisor, &1))
+    Children.specs()
+    |> start_children()
 
     :telemetry.attach(
       "reorder-file-sets-stop-handler",
@@ -53,10 +88,49 @@ defmodule Meadow.Application do
     result
   end
 
+  def start_children(children) do
+    retry with: exponential_backoff() |> randomize() |> cap(10_000) |> Stream.take(10) do
+      children
+      |> Enum.each(fn spec ->
+        if distributed?(spec) do
+          Horde.DynamicSupervisor.start_child(Meadow.HordeSupervisor, spec)
+        else
+          DynamicSupervisor.start_child(Meadow.Supervisor, spec)
+        end
+        |> case do
+          {:ok, _pid} ->
+            :ok
+
+          {:error, {:already_started, _pid}} ->
+            Logger.warning("Not starting #{inspect(spec)}: already started")
+            :ok
+
+          {:error, reason} ->
+            Logger.error("Failed to start child #{inspect(spec)}: #{inspect(reason)}")
+        end
+      end)
+    end
+  end
+
   # Tell Phoenix to update the endpoint configuration
   # whenever the application is updated.
   def config_change(changed, _new, removed) do
     MeadowWeb.Endpoint.config_change(changed, removed)
     :ok
   end
+
+  defp spec_name(child_spec) do
+    case child_spec do
+      %{start: {_mod, _fun, [args]}} when is_list(args) -> Keyword.get(args, :name)
+      %{start: {_mod, _fun, args}} when is_list(args) -> Keyword.get(args, :name)
+      {_mod, opts} when is_list(opts) -> Keyword.get(opts, :name)
+      %{id: {:via, Horde.Registry, _} = name} -> name
+      _ -> nil
+    end
+  end
+
+  defp distributed?(child_spec), do: spec_name(child_spec) |> via_horde?()
+
+  defp via_horde?({:via, Horde.Registry, _}), do: true
+  defp via_horde?(_), do: false
 end

app/lib/meadow/application/caches.ex

@@ -3,38 +3,39 @@ defmodule Meadow.Application.Caches do
   Cache specs for Meadow.Application
   """
   require Cachex.Spec
+  import Cachex.Spec
 
   def specs do
     [
       cache_spec(:global_cache, Meadow.Cache),
       cache_spec(
         :chat_conversation_cache,
         Meadow.Cache.Chat.Conversations,
-        expiration: Cachex.Spec.expiration(default: :timer.hours(6)),
+        expiration: expiration(default: :timer.hours(6)),
         stats: true
       ),
       cache_spec(
         :coded_term_cache,
         Meadow.Cache.CodedTerms,
-        expiration: Cachex.Spec.expiration(default: :timer.hours(6)),
+        expiration: expiration(default: :timer.hours(6)),
         stats: true
       ),
       cache_spec(
         :controlled_term_cache,
         Meadow.Cache.ControlledTerms,
-        expiration: Cachex.Spec.expiration(default: :timer.hours(6)),
+        expiration: expiration(default: :timer.hours(6)),
         stats: true
       ),
       cache_spec(
         :preservation_check_job_cache,
         Meadow.Cache.PreservationChecks,
-        expiration: Cachex.Spec.expiration(default: :timer.hours(6)),
+        expiration: expiration(default: :timer.hours(6)),
         stats: true
       ),
       cache_spec(
         :aws_credentials_cache,
         Meadow.Cache.AWS.Credentials,
-        expiration: Cachex.Spec.expiration(default: :timer.hours(6)),
+        expiration: expiration(default: :timer.hours(6)),
         stats: true
       )
     ]
@@ -57,8 +58,9 @@ defmodule Meadow.Application.Caches do
   end
 
   defp cache_spec(id, name, args \\ []) do
+    args = Keyword.put_new(args, :router, router(module: Cachex.Router.Ring, options: [monitor: true]))
     %{
-      id: id,
+      id: [id, Node.self()] |> Enum.join("_") |> String.to_atom(),
       start: {Cachex, :start_link, [name, args]},
       type: :supervisor
     }

app/lib/meadow/application/children.ex

@@ -17,10 +17,9 @@ defmodule Meadow.Application.Children do
       "csv_update_driver" => Meadow.CSVMetadataUpdateDriver,
       "database_listeners" => [
         {WalEx.Supervisor, Application.get_env(:meadow, WalEx)},
-        {Meadow.Events.Works.Arks.Processor,
-         token_count: 100, interval: 1_000, replenish_count: 10}
+        {Meadow.Events.Works.Arks.Processor, token_count: 100, interval: 1_000, replenish_count: 10} |> distributed()
       ],
-      "scheduler" => Meadow.Scheduler,
+      "scheduler" => Meadow.Scheduler |> distributed(),
       "work_creator" => [Meadow.Ingest.WorkCreator, Meadow.Ingest.WorkRedriver]
     }
   end
@@ -43,8 +42,8 @@ defmodule Meadow.Application.Children do
         MeadowWeb.Endpoint,
         {Absinthe.Subscription, MeadowWeb.Endpoint},
         MeadowWeb.Subscription,
-        Anubis.Server.Registry,
-        {MeadowWeb.MCP.Server, transport: :streamable_http}
+        MeadowWeb.MCP.GlobalRegistry,
+        {MeadowWeb.MCP.Server, transport: :streamable_http, registry: MeadowWeb.MCP.GlobalRegistry} |> distributed()
       ],
       "web.notifiers" => [
         {Meadow.Ingest.Progress, interval: Config.progress_ping_interval()}
@@ -144,4 +143,26 @@ defmodule Meadow.Application.Children do
         {Plug.Cowboy, scheme: :http, plug: plug, options: [port: port]}
     end
   end
+
+  def distributed(%{start: {mod, fun, [args]}} = spec) do
+    %{spec | start: {mod, fun, [via_horde(Keyword.put_new(args, :name, mod))]}}
+  end
+
+  def distributed(%{start: {mod, fun, args}} = spec) do
+    %{spec | start: {mod, fun, via_horde(Keyword.put_new(args, :name, mod))}}
+  end
+
+  def distributed({mod, args}) do
+    {mod, via_horde(Keyword.put_new(args, :name, mod))}
+  end
+
+  def distributed(mod), do: {mod, via_horde([name: mod])}
+
+  defp via_horde(args) do
+    Keyword.update(args, :name, nil, fn name ->
+      if is_atom(name),
+        do: {:via, Horde.Registry, {Meadow.HordeRegistry, name}},
+        else: name
+    end)
+  end
 end

app/lib/meadow/config/runtime/dev.ex

@@ -97,6 +97,11 @@ defmodule Meadow.Config.Runtime.Dev do
       callback_port: 3001,
       ssl_port: 3001
 
+    config :libcluster,
+      topologies: [
+        ecs: [strategy: Cluster.Strategy.LocalEpmd]
+      ]
+
     if prefix = System.get_env("DEV_PREFIX") do
       config :meadow,
         dc_api: [

app/lib/meadow/config/runtime/prod.ex

@@ -3,6 +3,8 @@ defmodule Meadow.Config.Runtime.Prod do
   Load and apply Meadow's runtime configuration for the production environment
   """
 
+  import Meadow.Config.Secrets
+
   def configure! do
     import Meadow.Config.Helper
 
@@ -12,5 +14,31 @@ defmodule Meadow.Config.Runtime.Prod do
       url: [host: System.get_env("MEADOW_HOSTNAME", "example.com"), port: 443],
       cache_static_manifest: "priv/static/cache_manifest.json",
       server: true
+
+    redis_host = get_secret(:redis, ["address"])
+    redis_port = get_secret(:redis, ["port"])
+
+    # config :anubis_mcp, :session_store,
+    #   enabled: true,
+    #   adapter: Anubis.Server.Session.Store.Redis,
+    #   redis_url: "redis://#{redis_host}:#{redis_port}",
+    #   pool_size: 10,
+    #   ttl: :timer.minutes(30),
+    #   namespace: "meadow:mcp:sessions",
+    #   connection_name: :anubis_redis
+
+    if System.get_env("CLUSTER_ENABLED") == "true" do
+      config :libcluster,
+        topologies: [
+          ecs: [
+            strategy: Cluster.Strategy.DNSPoll,
+            config: [
+              polling_interval: 5_000,
+              query: System.get_env("CLUSTER_DNS_NAME"),
+              node_basename: "meadow"
+            ]
+          ]
+        ]
+    end
   end
 end

app/lib/meadow/config/secrets.ex

@@ -16,6 +16,7 @@ defmodule Meadow.Config.Secrets do
     meadow: "config/meadow",
     meadow_pipeline: "config/meadow-pipeline",
     nusso: "infrastructure/nusso",
+    redis: "infrastructure/cache",
     wildcard_ssl: "config/wildcard_ssl"
   }
 

app/lib/meadow/data/index_batcher.ex

@@ -52,18 +52,27 @@ defmodule Meadow.Data.IndexBatcher do
   def clear(schema), do: dispatch(schema, :clear)
 
   defp dispatch(schema, message) do
-    case Process.whereis(:"#{schema}_batcher") do
-      nil -> :noop
-      pid -> GenServer.call(pid, message, :infinity)
+    case Horde.Registry.whereis_name({Meadow.HordeRegistry, registry_name(schema)}) do
+      pid when is_pid(pid) -> GenServer.call(pid, message, :infinity)
+      _ -> :noop
     end
   end
 
   def child_spec(schema, opts \\ []) do
     opts =
-      [schema: schema, version: 2, name: :"#{schema.__schema__(:source)}_batcher"]
+      [schema: schema, version: 2, name: registry_name(schema)]
       |> Keyword.merge(opts)
 
     %{id: Module.concat(__MODULE__, schema), start: {__MODULE__, :start_link, [opts]}}
+    |> Meadow.Application.Children.distributed()
+  end
+
+  defp registry_name(source) when is_binary(source), do: :"#{source}_batcher"
+
+  defp registry_name(schema) do
+    if Code.ensure_loaded?(schema) and function_exported?(schema, :__schema__, 1),
+      do: registry_name(schema.__schema__(:source)),
+      else: registry_name(to_string(schema))
   end
 
   def start_link(args \\ []) do