neiam /apollos_crib

It's Apollo's Crib

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

Survive a node that is shutting down underneath a query

c59c3a0 · Gabriel Morell · 2026-09-01 20:58

8 files +95 -11

Message

{commit_body(@commit)}

Files changed

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex

+26 −0

@@ -88,6 +88,9 @@ defmodule RoomGtfs.Worker do

88	88	\|> via_tuple()
89	89	\|> GenServer.call({:query_realtime, trips, stop}, 5_000)
90	90	\|> Enum.map(&slim_entity/1)
	91	+ rescue
	92	+ # As above: a stopped Registry raises rather than exiting.
	93	+ ArgumentError -> []
91	94	catch
92	95	:exit, _ -> []
93	96	end

@@ -132,6 +135,12 @@ defmodule RoomGtfs.Worker do

132	135	"gtfs-rt#{name}"
133	136	\|> via_tuple()
134	137	\|> GenServer.call({:query_alerts, stop, route_ids}, 10_000)
	138	+ rescue
	139	+ # A Registry that has stopped -- which is what a shutting-down node
	140	+ # looks like -- raises from the lookup before any call happens, where a
	141	+ # dead process would have exited. Catching only the exit left every
	142	+ # in-flight caller to crash on the way out of a deploy.
	143	+ ArgumentError -> []
135	144	catch
136	145	:exit, _ -> []
137	146	end

@@ -149,6 +158,12 @@ defmodule RoomGtfs.Worker do

149	158	"gtfs-rt#{name}"
150	159	\|> via_tuple()
151	160	\|> GenServer.call(:current_alerts, 10_000)
	161	+ rescue
	162	+ # A Registry that has stopped -- which is what a shutting-down node
	163	+ # looks like -- raises from the lookup before any call happens, where a
	164	+ # dead process would have exited. Catching only the exit left every
	165	+ # in-flight caller to crash on the way out of a deploy.
	166	+ ArgumentError -> []
152	167	catch
153	168	:exit, _ -> []
154	169	end

@@ -162,6 +177,9 @@ defmodule RoomGtfs.Worker do

162	177	:miss ->
163	178	try do
164	179	"gtfs-rt#{name}" \|> via_tuple() \|> GenServer.call(:query_vehicle_positions, 5_000)
	180	+ rescue
	181	+ # As above: a stopped Registry raises rather than exiting.
	182	+ ArgumentError -> []
165	183	catch
166	184	:exit, _ -> []
167	185	end

@@ -304,6 +322,8 @@ defmodule RoomGtfs.Worker do

304	322	_val ->
305	323	try do
306	324	get_current_vehicle_positions(id, trips, @occupancy_timeout_ms)
	325	+ rescue
	326	+ ArgumentError -> []
307	327	catch
308	328	:exit, _ -> []
309	329	end

@@ -670,6 +690,12 @@ defmodule RoomGtfs.Worker.RT do

670	690	"gtfs-rt#{name}"
671	691	\|> via_tuple()
672	692	\|> GenServer.call(:stats)
	693	+ rescue
	694	+ # A Registry that has stopped -- which is what a shutting-down node
	695	+ # looks like -- raises from the lookup before any call happens, where a
	696	+ # dead process would have exited. Catching only the exit left every
	697	+ # in-flight caller to crash on the way out of a deploy.
	698	+ ArgumentError -> %{rt_sa: :unavailable, rt_tu: :unavailable, rt_vp: :unavailable}
673	699	catch
674	700	:exit, _ -> %{rt_sa: :unavailable, rt_tu: :unavailable, rt_vp: :unavailable}
675	701	end

modified apps/room_gtfs/lib/room_gtfs/rt_index.ex

+24 −7

@@ -78,7 +78,7 @@ defmodule RoomGtfs.RTIndex do

78	78	\|> Enum.map(&{{source_id, :vp, &1.trip_id}, &1})
79	79
80	80	replace(source_id, :vp, rows)
81		− :ets.insert(@table, {{source_id, :vp_all}, vehicles})
	81	+ if exists?(), do: :ets.insert(@table, {{source_id, :vp_all}, vehicles})
82	82	:ok
83	83	end
84	84

@@ -140,9 +140,9 @@ defmodule RoomGtfs.RTIndex do

140	140	end
141	141
142	142	def vehicles(source_id) do
143		− case :ets.lookup(@table, {source_id, :vp_all}) do
	143	+ case exists?() and :ets.lookup(@table, {source_id, :vp_all}) do
144	144	[{_k, vehicles}] -> {:ok, vehicles}
145		− [] -> :miss
	145	+ _otherwise -> :miss
146	146	end
147	147	end
148	148

@@ -150,16 +150,33 @@ defmodule RoomGtfs.RTIndex do

150	150	# Whether this source has ever stored a feed of this kind. Distinct from
151	151	# having no rows: a feed can legitimately arrive carrying nothing.
152	152	def indexed?(source_id, kind) do
153		− :ets.member(@table, {source_id, {:present, kind}})
	153	+ exists?() and :ets.member(@table, {source_id, {:present, kind}})
154	154	end
155	155
	156	+ @doc false
	157	+ # The table is not always there, and every access has to survive that.
	158	+ #
	159	+ # It is created by this app's supervisor and dies with it, so there are two
	160	+ # windows either side of a running node: before the supervisor has started,
	161	+ # and after it has stopped. The second is the one that bit -- during a
	162	+ # deploy, the old node's ETS table goes while LiveViews are still asking it
	163	+ # for arrivals, and `:ets.member/2` against a table that has gone raises
	164	+ # rather than answering. Every caller holding a query crashed on the way out.
	165	+ #
	166	+ # A missing table means nothing is indexed, which is a state the readers
	167	+ # already handle: they fall back to asking the worker.
	168	+ def exists?, do: :ets.whereis(@table) != :undefined
	169	+
156	170	# The marker is a two-element key on purpose: the per-trip rows are scanned
157	171	# with a three-element pattern for suffix matching, and a marker shaped like
158	172	# one of them comes back from that scan as a trip whose id is an atom.
159	173	defp replace(source_id, kind, rows) do
160		− :ets.match_delete(@table, {{source_id, kind, :_}, :_})
161		− :ets.insert(@table, rows)
162		− :ets.insert(@table, {{source_id, {:present, kind}}, true})
	174	+ if exists?() do
	175	+ :ets.match_delete(@table, {{source_id, kind, :_}, :_})
	176	+ :ets.insert(@table, rows)
	177	+ :ets.insert(@table, {{source_id, {:present, kind}}, true})
	178	+ end
	179	+
163	180	:ok
164	181	end
165	182

modified apps/room_gtfs/test/rt_index_test.exs

+45 −4

@@ -12,15 +12,29 @@ defmodule RoomGtfs.RTIndexTest do

12	12	alias RoomGtfs.RTIndex
13	13	alias TransitRealtime, as: T
14	14
	15	+ # The table's owner is linked to whoever started it, so in tests it does not
	16	+ # outlive the process that created it. Each test therefore makes sure the
	17	+ # table is there rather than assuming a previous one left it behind -- which
	18	+ # also makes the "table is gone" tests below safe to run in any order.
15	19	setup do
	20	+ ensure_table()
	21	+
	22	+ id = System.unique_integer([:positive])
	23	+
	24	+ on_exit(fn ->
	25	+ if :ets.whereis(RTIndex.table()) != :undefined do
	26	+ :ets.match_delete(RTIndex.table(), {{id, :_, :_}, :_})
	27	+ end
	28	+ end)
	29	+
	30	+ %{id: id}
	31	+ end
	32	+
	33	+ defp ensure_table do
16	34	if :ets.whereis(RTIndex.table()) == :undefined do
17	35	{:ok, _} = RTIndex.start_link()
18	36	Process.sleep(20)
19	37	end
20		−
21		− id = System.unique_integer([:positive])
22		− on_exit(fn -> :ets.match_delete(RTIndex.table(), {{id, :_, :_}, :_}) end)
23		− %{id: id}
24	38	end
25	39
26	40	defp entity(trip_id, stop_ids, attrs \\ []) do

@@ -124,6 +138,33 @@ defmodule RoomGtfs.RTIndexTest do

124	138	end
125	139	end
126	140
	141	+ describe "when the table is not there" do
	142	+ # It is created by the app's supervisor and dies with it, so there are two
	143	+ # windows either side of a running node. The second is the one that bit: on
	144	+ # a deploy the old node's table goes while LiveViews are still asking for
	145	+ # arrivals, and every caller holding a query crashed on the way out.
	146	+ setup do
	147	+ # The next test's setup puts it back.
	148	+ if :ets.whereis(RTIndex.table()) != :undefined do
	149	+ :ets.delete(RTIndex.table())
	150	+ end
	151	+
	152	+ :ok
	153	+ end
	154	+
	155	+ test "reads answer :miss rather than raising", %{id: id} do
	156	+ assert RTIndex.trip_updates(id, ["t1"], "s1") == :miss
	157	+ assert RTIndex.trip_updates_by_suffix(id, ["t1"], "s1") == :miss
	158	+ assert RTIndex.vehicles(id) == :miss
	159	+ assert RTIndex.vehicles(id, ["t1"]) == :miss
	160	+ end
	161	+
	162	+ test "writes are dropped rather than raising", %{id: id} do
	163	+ assert RTIndex.put_trip_updates(id, [entity("t1", ["s1"])]) == :ok
	164	+ assert RTIndex.put_vehicles(id, [%{vehicle_id: "v1", trip_id: "t1"}]) == :ok
	165	+ end
	166	+ end
	167	+
127	168	describe "vehicles" do
128	169	test "by trip, and all at once", %{id: id} do
129	170	vehicles = [

Parents: b85c005