neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

Count the Ankyra's subscribers where the answer actually lives

a971fe6 · Gabriel Morell · 2026-09-03 00:10

10 files +91 -39

Message

{commit_body(@commit)}

Files changed

modified apps/room_sanctum/lib/room_sanctum/workers/ankyra_worker.ex

+69 −39

@@ -21,8 +21,11 @@ defmodule RoomSanctum.Worker.Ankyra do

21	21	initial_delay: 10
22	22	)
23	23
	24	+ # Five seconds rather than two: this is an HTTP call to the management
	25	+ # API now, not a local queue declare, and how many phones are attached is
	26	+ # not a fast-moving number.
24	27	Periodic.start_link(
25		− every: :timer.seconds(2),
	28	+ every: :timer.seconds(5),
26	29	run: fn -> RoomSanctum.Worker.Ankyra.meta_check(opts[:id]) end,
27	30	initial_delay: 10
28	31	)

@@ -57,26 +60,15 @@ defmodule RoomSanctum.Worker.Ankyra do

57	60	\|> GenServer.cast(:meta_check)
58	61	end
59	62
60		− defp queue_from_client_id(client_id) do
61		− "mqtt-subscription-#{client_id}qos0"
62		− end
63		−
64	63	def topic(id), do: "ankyra:#{id}"
65	64
66	65	def handle_cast(:meta_check, %{ankyra: nil} = state), do: {:noreply, state}
67	66
68	67	def handle_cast(:meta_check, state) do
69	68	parent = self()
70		− client_ids = state.ankyra.client_ids \|\| []
71		−
72		− spawn(fn ->
73		− count =
74		− client_ids
75		− \|> Enum.map(&check_consumer_count(queue_from_client_id(&1)))
76		− \|> Enum.sum()
	69	+ username = state.ankyra.username
77	70
78		− send(parent, {:status_result, count})
79		− end)
	71	+ spawn(fn -> send(parent, {:status_result, count_mqtt_connections(username)}) end)
80	72
81	73	{:noreply, state}
82	74	end

@@ -122,32 +114,70 @@ defmodule RoomSanctum.Worker.Ankyra do

122	114	{:noreply, state}
123	115	end
124	116
125		− defp check_consumer_count(queue) do
126		− with {:ok, conn} <- AMQP.Application.get_connection(:default),
127		− {:ok, chan} <- AMQP.Channel.open(conn) do
128		− Process.unlink(chan.pid)
129		−
130		− result =
131		− try do
132		− case AMQP.Queue.declare(chan, queue, passive: true) do
133		− {:ok, %{consumer_count: n}} -> n
134		− _ -> 0
135		− end
136		− catch
137		− _, _ -> 0
138		− end
139		−
140		− if Process.alive?(chan.pid) do
141		− try do
142		− AMQP.Channel.close(chan)
143		− catch
144		− _, _ -> :ok
145		− end
146		− end
147		−
148		− result
	117	+ @doc """
	118	+ How many clients are attached to this Ankyra, or nil if we could not find out.
	119	+
	120	+ RabbitMQ will not answer this over AMQP. An MQTT subscriber's queue is
	121	+ exclusive to the connection that made it, so a passive `queue.declare` from
	122	+ ours is refused with `405 RESOURCE_LOCKED`; and since 3.12 a QoS 0
	123	+ subscription is a `rabbit_mqtt_qos0_queue`, a pseudo-queue whose stat is the
	124	+ constant `{ok, 0, 0}` with no consumer ever counted. Both roads led to zero,
	125	+ which is what the detail page reported while two clients sat there happily
	126	+ receiving.
	127	+
	128	+ The management API does know. Every connection carries the protocol it
	129	+ speaks and the user it authenticated as, and an Ankyra is a rabbit user --
	130	+ so MQTT connections under that username are exactly its clients. Counting
	131	+ connections rather than probing per-client queues also drops the dependency
	132	+ on `client_ids`, which this worker only re-reads once a minute.
	133	+
	134	+ nil rather than 0 when the API cannot be reached: "we do not know" and
	135	+ "nobody is listening" are different things, and the page draws them
	136	+ differently. Reporting the unreachable case as zero is the bug this replaces.
	137	+ """
	138	+ def count_mqtt_connections(username) do
	139	+ cfg = Application.get_env(:room_sanctum, :rabbit_mgmt, [])
	140	+
	141	+ with url when is_binary(url) <- cfg[:url],
	142	+ {:ok, %HTTPoison.Response{status_code: 200, body: body}} <- fetch_connections(url, cfg),
	143	+ {:ok, connections} when is_list(connections) <- Jason.decode(body) do
	144	+ vhost = cfg[:vhost] \|\| "/"
	145	+ Enum.count(connections, &mqtt_connection_for?(&1, username, vhost))
149	146	else
150		− _ -> 0
	147	+ other ->
	148	+ Logger.debug("ankyra: could not read connections from management API: #{inspect(other)}")
	149	+ nil
151	150	end
152	151	end
	152	+
	153	+ # HTTPoison answers with a tuple for anything the far end does, but hackney
	154	+ # itself throws -- so a broker that is merely unreachable arrives here as an
	155	+ # exception, and unrescued it would kill the spawned checker with no status
	156	+ # ever sent and the badge stuck on "checking" forever.
	157	+ defp fetch_connections(url, cfg) do
	158	+ HTTPoison.get(
	159	+ String.trim_trailing(url, "/") <> "/api/connections?columns=protocol,user,vhost",
	160	+ [{"accept", "application/json"}],
	161	+ hackney: [basic_auth: {to_string(cfg[:username]), to_string(cfg[:password])}],
	162	+ timeout: 2_000,
	163	+ recv_timeout: 2_000
	164	+ )
	165	+ rescue
	166	+ error -> {:error, error}
	167	+ catch
	168	+ kind, reason -> {:error, {kind, reason}}
	169	+ end
	170	+
	171	+ # "MQTT 3-1-1" today, "MQTT 5-0" the moment a client asks for it -- the
	172	+ # version is not what is being asked about here.
	173	+ defp mqtt_connection_for?(
	174	+ %{"protocol" => protocol, "user" => user, "vhost" => on_vhost},
	175	+ username,
	176	+ vhost
	177	+ )
	178	+ when is_binary(protocol) do
	179	+ user == username and on_vhost == vhost and String.starts_with?(protocol, "MQTT")
	180	+ end
	181	+
	182	+ defp mqtt_connection_for?(_conn, _username, _vhost), do: false
153	183	end

modified apps/room_sanctum/mix.exs

+1 −0

@@ -98,6 +98,7 @@ defmodule RoomSanctum.MixProject do

98	98
99	99	# hermes
100	100	# {:room_hermes, in_umbrella: true}
	101	+ {:httpoison, "~> 1.8"},
101	102	{:amqp, "~> 4.1"},
102	103	{:amqp_client, "~> 4.2"},
103	104	{:rabbit_common, "~> 4.0"},

modified config/dev.exs

+9 −0

@@ -144,6 +144,15 @@ config :amqp,

144	144	default: [connection: :default]
145	145	]
146	146
	147	+# How many clients are attached to an Ankyra is a question only the management
	148	+# API answers -- see RoomSanctum.Worker.Ankyra.count_mqtt_connections/1 for why
	149	+# AMQP cannot. Same broker, its HTTP port.
	150	+config :room_sanctum, :rabbit_mgmt,
	151	+ url: "http://localhost:15672",
	152	+ username: "hermes",
	153	+ password: "hermes",
	154	+ vhost: "/"
	155	+
147	156	# Do not include metadata nor timestamps in development logs
148	157	config :logger, :console, format: "[$level] $message\n", level: :info
149	158

modified config/runtime.exs

+12 −0

@@ -134,6 +134,18 @@ if config_env() == :prod do

134	134	default: [connection: :default]
135	135	]
136	136
	137	+ # Counting an Ankyra's subscribers needs the management API rather than AMQP.
	138	+ # Credentials are the broker's own default user -- the operator writes it to
	139	+ # the rabbitmq-default-user secret and tags it administrator -- so this reuses
	140	+ # what RABBIT_URL is already built from rather than asking for a second
	141	+ # account. Left unset, the detail page says "checking" instead of lying about
	142	+ # having no subscribers.
	143	+ config :room_sanctum, :rabbit_mgmt,
	144	+ url: System.get_env("RABBIT_MGMT_URL"),
	145	+ username: System.get_env("AMQP_USER"),
	146	+ password: System.get_env("AMQP_PASS"),
	147	+ vhost: System.get_env("RABBIT_VHOST", "/")
	148	+
137	149	# ## Using releases
138	150	#
139	151	# If you are doing OTP releases, you need to instruct Phoenix

Parents: 43141c3