feat(pubsub): route Phoenix.PubSub through Valkey when configured
Switches the PubSub adapter to Phoenix.PubSub.Redis backed by the shared Valkey instance (same backend GridCache uses), via the new phoenix_pubsub_redis dep. Falls back to the default PG2 / Erlang- distribution adapter when VALKEY_URL is unset, so dev/test boot unchanged. Decouples PubSub fan-out from libcluster's BEAM connection. A flapping Erlang distribution (the OOM-cascade scenario from 2026-05-03 surfaced this — libcluster warnings dominated the logs while pods CrashLoopBackOff'd) no longer takes broadcasts with it, so the rest of the cluster keeps publishing while the bad pod recovers. node_name uses POD_IP for cluster-wide uniqueness, falling back to Erlang's node() name in non-K8s environments. Required by the Redis adapter so loopback filtering doesn't drop a pod's own broadcasts as if they came from a peer.
This commit is contained in:
parent
9edaee9344
commit
e53a34dd75
3 changed files with 39 additions and 1 deletions
|
|
@ -22,7 +22,7 @@ defmodule Microwaveprop.Application do
|
||||||
# config logs a warning and skips startup so the app still boots.
|
# config logs a warning and skips startup so the app still boots.
|
||||||
aprs_repo_child_spec(),
|
aprs_repo_child_spec(),
|
||||||
{Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]},
|
{Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]},
|
||||||
{Phoenix.PubSub, name: Microwaveprop.PubSub},
|
pubsub_child_spec(),
|
||||||
# Partitioned Task.Supervisor — callers route heavy `async_stream`
|
# Partitioned Task.Supervisor — callers route heavy `async_stream`
|
||||||
# work through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor,
|
# work through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor,
|
||||||
# self()}}` so concurrent workers (recalibrator, HRRR/GEFS range
|
# self()}}` so concurrent workers (recalibrator, HRRR/GEFS range
|
||||||
|
|
@ -101,6 +101,36 @@ defmodule Microwaveprop.Application do
|
||||||
:ok
|
:ok
|
||||||
end
|
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
|
# Returns Microwaveprop.AprsRepo child spec if configured (has a :url
|
||||||
# or :database key), otherwise logs a warning and returns nil so the
|
# or :database key), otherwise logs a warning and returns nil so the
|
||||||
# supervisor children list can filter it out. APRS calibration is an
|
# supervisor children list can filter it out. APRS calibration is an
|
||||||
|
|
|
||||||
6
mix.exs
6
mix.exs
|
|
@ -107,6 +107,12 @@ defmodule Microwaveprop.MixProject do
|
||||||
# Talks to the Valkey/Redis instance at VALKEY_URL; falls back to
|
# Talks to the Valkey/Redis instance at VALKEY_URL; falls back to
|
||||||
# the legacy ETS implementation when unset (dev/test).
|
# the legacy ETS implementation when unset (dev/test).
|
||||||
{:redix, "~> 1.5"},
|
{: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"},
|
{:ecto_psql_extras, "~> 0.8"},
|
||||||
{:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false},
|
{:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false},
|
||||||
{:live_table, "~> 0.4.1"},
|
{:live_table, "~> 0.4.1"},
|
||||||
|
|
|
||||||
2
mix.lock
2
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_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_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": {: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"},
|
"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": {: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"},
|
"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"},
|
"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"},
|
"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"},
|
"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"},
|
"ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"},
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue