diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 94ce33cd..22ce4dc5 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -22,7 +22,7 @@ defmodule Microwaveprop.Application do # config logs a warning and skips startup so the app still boots. aprs_repo_child_spec(), {Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]}, - {Phoenix.PubSub, name: Microwaveprop.PubSub}, + pubsub_child_spec(), # Partitioned Task.Supervisor — callers route heavy `async_stream` # work through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, # self()}}` so concurrent workers (recalibrator, HRRR/GEFS range @@ -101,6 +101,36 @@ defmodule Microwaveprop.Application do :ok end + # PubSub adapter selector. When `:valkey_url` is configured (prod via + # VALKEY_URL env var), broadcasts ride Phoenix.PubSub.Redis on the + # shared Valkey instance — same backend that GridCache uses, so both + # cross-pod features go through one connection target. Decouples + # PubSub fan-out from libcluster's Erlang distribution: a flapping + # BEAM connection (e.g., the OOM-cascade scenario from 2026-05-03) + # no longer takes broadcasts down with it. + # + # When unset, default to Phoenix.PubSub.PG2 (in-process / Erlang + # distribution). Dev/test boot without Valkey running. + defp pubsub_child_spec do + case Application.get_env(:microwaveprop, :valkey_url) do + nil -> + {Phoenix.PubSub, name: Microwaveprop.PubSub} + + "" -> + {Phoenix.PubSub, name: Microwaveprop.PubSub} + + url -> + # node_name must be unique per pod or Redis-adapter loopback + # filtering breaks (every node would discard its own broadcasts + # mistakenly identifying them as another node's). Pod IP is + # the simplest cluster-wide-unique identifier; falls back to + # Erlang's node() name in non-K8s environments. + node_name = System.get_env("POD_IP") || to_string(node()) + + {Phoenix.PubSub, name: Microwaveprop.PubSub, adapter: Phoenix.PubSub.Redis, url: url, node_name: node_name} + end + end + # Returns Microwaveprop.AprsRepo child spec if configured (has a :url # or :database key), otherwise logs a warning and returns nil so the # supervisor children list can filter it out. APRS calibration is an diff --git a/mix.exs b/mix.exs index 1f5991b6..8ede37a6 100644 --- a/mix.exs +++ b/mix.exs @@ -107,6 +107,12 @@ defmodule Microwaveprop.MixProject do # Talks to the Valkey/Redis instance at VALKEY_URL; falls back to # the legacy ETS implementation when unset (dev/test). {:redix, "~> 1.5"}, + # Redis-backed Phoenix.PubSub adapter. Used in prod when + # VALKEY_URL is set so broadcasts go through Valkey instead of + # libcluster's Erlang distribution; dev/test stay on the default + # PG2 adapter (in-process). Wires up via application.ex's + # PubSub child spec, not a config block. + {:phoenix_pubsub_redis, "~> 3.0"}, {:ecto_psql_extras, "~> 0.8"}, {:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false}, {:live_table, "~> 0.4.1"}, diff --git a/mix.lock b/mix.lock index 3bc9f305..19a88199 100644 --- a/mix.lock +++ b/mix.lock @@ -52,10 +52,12 @@ "phoenix_live_reload": {:hex, :phoenix_live_reload, "1.6.2", "b18b0773a1ba77f28c52decbb0f10fd1ac4d3ae5b8632399bbf6986e3b665f62", [:mix], [{:file_system, "~> 0.2.10 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.4", [hex: :phoenix, repo: "hexpm", optional: false]}], "hexpm", "d1f89c18114c50d394721365ffb428cce24f1c13de0467ffa773e2ff4a30d5b9"}, "phoenix_live_view": {:hex, :phoenix_live_view, "1.1.28", "8a8e123d018025f756605a2fb02a4854f0d3cd7b207f710fef1fd5d9d72d0254", [:mix], [{:igniter, ">= 0.6.16 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:lazy_html, "~> 0.1.0", [hex: :lazy_html, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.6.15 or ~> 1.7.0 or ~> 1.8.0-rc", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.3 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.15", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "24faad535b65089642c3a7d84088109dc58f49c1f1c5a978659855d643466353"}, "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.2.0", "ff3a5616e1bed6804de7773b92cbccfc0b0f473faf1f63d7daf1206c7aeaaa6f", [:mix], [], "hexpm", "adc313a5bf7136039f63cfd9668fde73bba0765e0614cba80c06ac9460ff3e96"}, + "phoenix_pubsub_redis": {:hex, :phoenix_pubsub_redis, "3.1.1", "54919406b0b6401e0f643e26fd3b814bc3460f3fd3a32aa9c3a473db89e9dca7", [:mix], [{:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:poolboy, "~> 1.5.1 or ~> 1.6", [hex: :poolboy, repo: "hexpm", optional: false]}, {:redix, "~> 1.0", [hex: :redix, repo: "hexpm", optional: false]}], "hexpm", "e339963f649e7cf52bd352e55fab7bebb42b65b501e43499dd6eb56890703c32"}, "phoenix_template": {:hex, :phoenix_template, "1.0.4", "e2092c132f3b5e5b2d49c96695342eb36d0ed514c5b252a77048d5969330d639", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "2c0c81f0e5c6753faf5cca2f229c9709919aba34fab866d3bc05060c9c444206"}, "plug": {:hex, :plug, "1.19.1", "09bac17ae7a001a68ae393658aa23c7e38782be5c5c00c80be82901262c394c0", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "560a0017a8f6d5d30146916862aaf9300b7280063651dd7e532b8be168511e62"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "polaris": {:hex, :polaris, "0.1.0", "dca61b18e3e801ecdae6ac9f0eca5f19792b44a5cb4b8d63db50fc40fc038d22", [:mix], [{:nx, "~> 0.5", [hex: :nx, repo: "hexpm", optional: false]}], "hexpm", "13ef2b166650e533cb24b10e2f3b8ab4f2f449ba4d63156e8c569527f206e2c2"}, + "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"}, "postgrex": {:hex, :postgrex, "0.22.0", "fb027b58b6eab1f6de5396a2abcdaaeb168f9ed4eccbb594e6ac393b02078cbd", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a68c4261e299597909e03e6f8ff5a13876f5caadaddd0d23af0d0a61afcc5d84"}, "prom_ex": {:hex, :prom_ex, "1.11.0", "1f6d67f2dead92224cb4f59beb3e4d319257c5728d9638b4a5e8ceb51a4f9c7e", [:mix], [{:absinthe, ">= 1.7.0", [hex: :absinthe, repo: "hexpm", optional: true]}, {:broadway, ">= 1.1.0", [hex: :broadway, repo: "hexpm", optional: true]}, {:ecto, ">= 3.11.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:finch, "~> 0.18", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, ">= 2.10.0", [hex: :oban, repo: "hexpm", optional: true]}, {:octo_fetch, "~> 0.4", [hex: :octo_fetch, repo: "hexpm", optional: false]}, {:peep, "~> 3.0", [hex: :peep, repo: "hexpm", optional: false]}, {:phoenix, ">= 1.7.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, ">= 0.20.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, ">= 1.16.0", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 2.6.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, ">= 1.0.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}, {:telemetry_metrics_prometheus_core, "~> 1.2", [hex: :telemetry_metrics_prometheus_core, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 1.1", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "76b074bc3730f0802978a7eb5c7091a65473eaaf07e99ec9e933138dcc327805"}, "ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"},