GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

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

⭐

Log in to mark this repository.

One realtime control per feed URL, not per kind

b6d9a98 · Gabriel Morell · 2026-08-25 21:44

15 files +195 -102
Message 
{commit_body(@commit)}

Files changed

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+68 −36
@@ -342,10 +342,6 @@ defmodule RoomGtfs.Worker.RT do
342 342 # A source polling every three minutes can happily use data up to three
343 343 # minutes old, and several sources sharing one URL -- which is the whole point
344 344 # of the regional feed -- then fetch it once between them instead of once each.
345 − defp cached_get(state, kind, url) do
346 − FeedCache.get(url, max(rt_period_ms(state, kind) - :timer.seconds(5), :timer.seconds(1)))
347 − end
348 −
349 345 defp record_health(state, kind, url, result, took) do
350 346 RoomGtfs.FeedHealth.record(state.id, state.inst && state.inst.name, kind, url, result, took)
351 347 end
@@ -766,15 +762,17 @@ defmodule RoomGtfs.Worker.RT do
766 762 # Gated here rather than by rescheduling the Periodic, which cannot be changed
767 763 # once started and would have to be torn down and rebuilt whenever the config
768 764 # is refreshed -- every four seconds, as it happens.
769 − defp due?(state, kind) do
770 − case state |> Map.get(:rt_polled_at, %{}) |> Map.get(kind) do
765 + defp due?(state, url, period) do
766 + case state |> Map.get(:rt_polled_at, %{}) |> Map.get(url) do
771 767 nil -> true
772 − at -> System.monotonic_time(:millisecond) - at >= rt_period_ms(state, kind)
768 + at -> System.monotonic_time(:millisecond) - at >= period
773 769 end
774 770 end
775 771
776 − defp mark_polled(state, kind) do
777 − polled = state |> Map.get(:rt_polled_at, %{}) |> Map.put(kind, System.monotonic_time(:millisecond))
772 + defp mark_polled(state, url) do
773 + polled =
774 + state |> Map.get(:rt_polled_at, %{}) |> Map.put(url, System.monotonic_time(:millisecond))
775 +
778 776 Map.put(state, :rt_polled_at, polled)
779 777 end
780 778
@@ -797,7 +795,9 @@ defmodule RoomGtfs.Worker.RT do
797 795 defp do_update_realtime(state) do
798 796 state =
799 797 if state.inst.enabled do
800 − Enum.reduce([:sa, :tu, :vp], state, &fetch_kind/2)
798 + state.inst.config
799 + |> rt_groups()
800 + |> Enum.reduce(state, &fetch_group/2)
801 801 else
802 802 bcast(state.id, :disabled)
803 803 state
@@ -806,52 +806,84 @@ defmodule RoomGtfs.Worker.RT do
806 806 {:noreply, state}
807 807 end
808 808
809 − # Where each kind comes from. A kind with no URL of its own falls back to the
810 − # combined feed, so an agency publishing everything in one message -- the MTA
811 − # subway feeds are 67 trip updates, 45 vehicle positions and an alert in a
812 − # single URL -- names it once instead of three times. Specific beats general,
813 − # so a combined feed plus a dedicated alerts feed works too.
814 − defp rt_url(config, :sa), do: Map.get(config, :url_rt_sa) || Map.get(config, :url_rt_shared)
815 − defp rt_url(config, :tu), do: Map.get(config, :url_rt_tu) || Map.get(config, :url_rt_shared)
816 − defp rt_url(config, :vp), do: Map.get(config, :url_rt_vp) || Map.get(config, :url_rt_shared)
809 + @doc """
810 + The kinds this source reads, grouped by the URL they come from.
817 811
818 − defp fetch_kind(kind, state) do
819 − url = rt_url(state.inst.config, kind)
812 + A URL is the unit of scheduling, not a kind. Where three kinds have three
813 + feeds that is the same thing; where one feed carries all three -- the MTA's
814 + subway feeds -- it very much is not. Polling those kinds independently meant
815 + fetching the same URL on the fastest of their three clocks and then handing
816 + the slower kinds whatever the cache still held, so vehicle positions could be
817 + minutes stale while trip updates from the very same message were seconds old.
818 + Nothing was saved by it: the request had already been made.
820 819
821 − cond do
822 − is_nil(url) -> state
823 − not due?(state, kind) -> state
824 − true -> do_fetch_kind(kind, url, mark_polled(state, kind))
820 + Grouped, one fetch updates everything that came in it.
821 + """
822 + def rt_groups(config) do
823 + [:tu, :vp, :sa]
824 + |> Enum.map(fn kind -> {kind, rt_url(config, kind)} end)
825 + |> Enum.reject(fn {_kind, url} -> url in [nil, ""] end)
826 + |> Enum.reduce([], fn {kind, url}, acc ->
827 + case List.keyfind(acc, url, 0) do
828 + nil -> acc ++ [{url, [kind]}]
829 + {^url, kinds} -> List.keyreplace(acc, url, 0, {url, kinds ++ [kind]})
830 + end
831 + end)
832 + end
833 +
834 + defp fetch_group({url, kinds}, state) do
835 + # The fastest kind in the group wins: the URL is fetched that often, and
836 + # every kind in it is that fresh. Only reachable by hand-editing the config,
837 + # since the form offers one control per group.
838 + period = kinds |> Enum.map(&rt_period_ms(state, &1)) |> Enum.min()
839 +
840 + if due?(state, url, period) do
841 + state |> mark_polled(url) |> do_fetch_group(url, kinds, period)
842 + else
843 + state
825 844 end
826 845 end
827 846
828 − defp do_fetch_kind(kind, url, state) do
847 + defp do_fetch_group(state, url, kinds, period) do
829 848 started = System.monotonic_time(:millisecond)
830 − result = cached_get(state, kind, url)
849 + result = FeedCache.get(url, max(period - :timer.seconds(5), :timer.seconds(1)))
831 850 took = System.monotonic_time(:millisecond) - started
832 851
833 852 case result do
834 853 {:ok, feed} when is_struct(feed, TransitRealtime.FeedMessage) ->
835 − # Narrowed before it is recorded, so health reports what this kind
836 − # actually got. Recording the raw feed would have every kind of a
837 − # combined source claim the same entity count -- exactly the reading
838 − # that means "the same URL is in all three fields by mistake", and
839 − # would say it about a correct configuration.
840 − feed = feed |> for_this_agency(state) |> slice(entity_field(kind))
841 − record_health(state, kind, url, {:ok, feed}, took)
842 − store_feed(state, kind, feed)
854 + feed = for_this_agency(feed, state)
855 +
856 + Enum.reduce(kinds, state, fn kind, acc ->
857 + # Narrowed before it is recorded, so health reports what this kind
858 + # actually got. Recording the whole message would have every kind of a
859 + # combined feed claim the same entity count -- exactly the reading
860 + # that means "the same URL is in all three fields by mistake", and
861 + # would say it about a correct configuration.
862 + sliced = slice(feed, entity_field(kind))
863 + record_health(acc, kind, url, {:ok, sliced}, took)
864 + store_feed(acc, kind, sliced)
865 + end)
843 866
844 867 {:error, error} ->
845 − record_health(state, kind, url, result, took)
868 + Enum.each(kinds, &record_health(state, &1, url, result, took))
846 869
847 870 Logger.info(
848 − "failed to fetch gtfs-rt url[#{kind}] for '#{state.inst.name}', reason: #{inspect(error)}"
871 + "failed to fetch gtfs-rt url#{inspect(kinds)} for '#{state.inst.name}', reason: #{inspect(error)}"
849 872 )
850 873
851 874 state
852 875 end
853 876 end
854 877
878 + # Where each kind comes from. A kind with no URL of its own falls back to the
879 + # combined feed, so an agency publishing everything in one message -- the MTA
880 + # subway feeds are 67 trip updates, 45 vehicle positions and an alert in a
881 + # single URL -- names it once instead of three times. Specific beats general,
882 + # so a combined feed plus a dedicated alerts feed works too.
883 + defp rt_url(config, :sa), do: Map.get(config, :url_rt_sa) || Map.get(config, :url_rt_shared)
884 + defp rt_url(config, :tu), do: Map.get(config, :url_rt_tu) || Map.get(config, :url_rt_shared)
885 + defp rt_url(config, :vp), do: Map.get(config, :url_rt_vp) || Map.get(config, :url_rt_shared)
886 +
855 887 defp entity_field(:sa), do: :alert
856 888 defp entity_field(:tu), do: :trip_update
857 889 defp entity_field(:vp), do: :vehicle
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/form_component.ex
+114 −53
@@ -89,57 +89,113 @@ defmodule RoomSanctumWeb.SourceLive.FormComponent do
89 89 end
90 90
91 91 @doc """
92 − Requests an hour the three intervals add up to, counting each distinct URL
93 − once.
92 + The kinds a source reads, grouped by the URL they come from -- one control per
93 + group, because a URL is what actually gets scheduled.
94 94
95 − A source with one combined feed polls one URL however many kinds read from
96 − it, so counting per kind would overstate it threefold -- and the number is
97 − only worth showing because some feeds are metered, where an overstatement is
98 − as unhelpful as no number at all.
95 + Three sliders over one combined feed were a lie: the URL is fetched on the
96 + fastest of the three clocks, so setting service alerts slower than trip
97 + updates saved no request at all, it just handed the slower kinds a staler copy
98 + of a message that had already arrived. One group, one interval, everything in
99 + it equally fresh.
100 +
101 + Delegates, so the form and the poller cannot disagree about what a group is.
102 + """
103 + def rt_groups(config) when is_map(config), do: RoomGtfs.Worker.RT.rt_groups(config)
104 + def rt_groups(_config), do: []
105 +
106 + @doc """
107 + Names a group by what it carries: the kinds when it is some of them, and
108 + nothing more specific than "Realtime" when one feed carries all three.
99 109 """
100 − def rt_requests_per_hour(config, idxs) when is_map(config) do
101 − for {kind, _label} <- rt_period_kinds(), reduce: %{} do
102 − acc ->
103 − case rt_url_for(config, kind) do
104 − nil -> acc
105 − url -> Map.update(acc, url, rt_idx_to_secs(idxs[kind]), &min(&1, rt_idx_to_secs(idxs[kind])))
106 − end
110 + def rt_group_label(kinds) do
111 + case Enum.sort(kinds) do
112 + [:sa, :tu, :vp] -> "Realtime (combined feed)"
113 + sorted -> sorted |> Enum.map(&kind_title/1) |> Enum.join(", ")
107 114 end
108 − |> Enum.map(fn {_url, secs} -> div(3600, secs) end)
109 − |> Enum.sum()
110 115 end
111 116
112 − def rt_requests_per_hour(_config, _idxs), do: 0
117 + defp kind_title(:tu), do: "Trip updates"
118 + defp kind_title(:vp), do: "Vehicle positions"
119 + defp kind_title(:sa), do: "Service alerts"
113 120
114 − defp rt_url_for(config, kind) do
115 − specific =
116 − case kind do
117 − :tu -> Map.get(config, :url_rt_tu)
118 − :vp -> Map.get(config, :url_rt_vp)
119 − :sa -> Map.get(config, :url_rt_sa)
120 − end
121 + @doc "Form key for a group's slider, e.g. `rt_idx_g_sa-tu-vp`."
122 + def rt_group_key(kinds), do: "rt_idx_g_" <> (kinds |> Enum.sort() |> Enum.join("-"))
123 +
124 + @doc """
125 + The index a group's slider sits at. The kinds in a group are written together,
126 + so they agree; the fastest wins if a config was edited by hand.
127 + """
128 + def rt_group_idx(config, kinds) do
129 + kinds
130 + |> Enum.map(fn kind -> rt_config_seconds(config, kind) end)
131 + |> Enum.reject(&is_nil/1)
132 + |> case do
133 + [] -> rt_secs_to_idx(nil)
134 + seconds -> rt_secs_to_idx(Enum.min(seconds))
135 + end
136 + end
121 137
122 − blank(specific) || blank(Map.get(config, :url_rt_shared))
138 + def rt_config_seconds(config, kind) when is_map(config) do
139 + Map.get(config, String.to_existing_atom("rt_period_#{kind}")) ||
140 + Map.get(config, "rt_period_#{kind}")
123 141 end
124 142
125 − defp blank(nil), do: nil
126 − defp blank(""), do: nil
127 − defp blank(value), do: value
143 + def rt_config_seconds(_config, _kind), do: nil
128 144
129 − defp convert_rt_period_params(source_params) do
130 − Enum.reduce(rt_period_kinds(), source_params, fn {kind, _label}, params ->
131 − case get_in(params, ["config", "rt_idx_#{kind}"]) do
132 − nil ->
133 − params
145 + defp to_seconds(value) when is_integer(value), do: value
146 + defp to_seconds(value) when is_binary(value), do: String.to_integer(value)
134 147
135 − idx ->
136 − params
137 − |> put_in(["config", "rt_period_#{kind}"], rt_idx_to_secs(idx))
138 − |> update_in(["config"], &Map.delete(&1, "rt_idx_#{kind}"))
139 − end
148 + @doc """
149 + Requests an hour these intervals add up to.
150 +
151 + One term per group rather than per kind, because a group is one URL: three
152 + kinds arriving in one feed cost one request, and counting them separately
153 + would treat a combined feed as three times more expensive than it is. The
154 + number is only worth showing because some feeds are metered, and an
155 + overstatement there is as unhelpful as no number at all.
156 + """
157 + def rt_requests_per_hour(config, idxs) when is_map(config) and is_map(idxs) do
158 + config
159 + |> rt_groups()
160 + |> Enum.map(fn {_url, kinds} ->
161 + idx = Map.get(idxs, kinds) || rt_group_idx(config, kinds)
162 + div(3600, rt_idx_to_secs(idx))
163 + end)
164 + |> Enum.sum()
165 + end
166 +
167 + def rt_requests_per_hour(_config, _idxs), do: 0
168 +
169 + # One slider writes every kind in its group, so the stored per-kind values
170 + # remain the source of truth and a config edited by hand still works.
171 + @doc false
172 + # Public only so it can be exercised directly: it is the step that turns what
173 + # the form submits into what the schema stores, and getting it wrong writes a
174 + # poll interval nobody asked for.
175 + def convert_rt_period_params(%{"config" => config} = source_params) when is_map(config) do
176 + config
177 + |> Map.keys()
178 + |> Enum.filter(&String.starts_with?(&1, "rt_idx_g_"))
179 + |> Enum.reduce(source_params, fn key, params ->
180 + secs = rt_idx_to_secs(get_in(params, ["config", key]))
181 +
182 + kinds =
183 + key
184 + |> String.replace_prefix("rt_idx_g_", "")
185 + |> String.split("-", trim: true)
186 +
187 + params
188 + |> then(fn p ->
189 + Enum.reduce(kinds, p, fn kind, acc ->
190 + put_in(acc, ["config", "rt_period_#{kind}"], secs)
191 + end)
192 + end)
193 + |> update_in(["config"], &Map.delete(&1, key))
140 194 end)
141 195 end
142 196
197 + def convert_rt_period_params(source_params), do: source_params
198 +
143 199 # GitHub poll-interval slider: index → label/seconds
144 200 @github_poll_steps [
145 201 {"15s", 15},
@@ -353,19 +409,8 @@ defmodule RoomSanctumWeb.SourceLive.FormComponent do
353 409 |> Enum.map(&{&1.name, &1.id})
354 410 run_period_idx = seconds_to_idx(source.meta && source.meta.run_period)
355 411
356 − rt_idxs =
357 − Map.new(rt_period_kinds(), fn {kind, _label} ->
358 − secs =
359 − case source.config do
360 − %RoomSanctum.Configuration.Configs.GTFS{} = cfg ->
361 − Map.get(cfg, String.to_existing_atom("rt_period_#{kind}"))
362 −
363 − _other ->
364 − nil
365 − end
366 −
367 − {kind, rt_secs_to_idx(secs)}
368 − end)
412 + rt_groups = rt_groups(source.config)
413 + rt_idxs = Map.new(rt_groups, fn {_url, kinds} -> {kinds, rt_group_idx(source.config, kinds)} end)
369 414
370 415 gh_secs =
371 416 case source.config do
@@ -399,6 +444,7 @@ defmodule RoomSanctumWeb.SourceLive.FormComponent do
399 444 |> assign(:mailbox_sel, mailbox_sel)
400 445 |> assign(:tint_opts, RoomSanctum.Tints.all())
401 446 |> assign(:run_period_idx, run_period_idx)
447 + |> assign(:rt_groups, rt_groups)
402 448 |> assign(:rt_idxs, rt_idxs)
403 449 |> assign(:rt_requests, rt_requests_per_hour(source.config, rt_idxs))
404 450 |> assign(:github_poll_idx, github_secs_to_idx(gh_secs))
@@ -489,15 +535,30 @@ defmodule RoomSanctumWeb.SourceLive.FormComponent do
489 535 # Read back from the converted params rather than the changeset: a slider
490 536 # the user has just dragged is in the params, and its label has to move with
491 537 # it or the control reads as broken.
538 + rt_groups = rt_groups(config_for_estimate)
539 +
492 540 rt_idxs =
493 − Map.new(rt_period_kinds(), fn {kind, _label} ->
494 − secs = get_in(source_params, ["config", "rt_period_#{kind}"])
495 − {kind, rt_secs_to_idx(secs)}
541 + Map.new(rt_groups, fn {_url, kinds} ->
542 + # From the converted params, not the changeset: a slider just dragged is
543 + # in the params, and its label has to move with it.
544 + seconds =
545 + kinds
546 + |> Enum.map(&get_in(source_params, ["config", "rt_period_#{&1}"]))
547 + |> Enum.reject(&is_nil/1)
548 +
549 + idx =
550 + case seconds do
551 + [] -> rt_group_idx(config_for_estimate, kinds)
552 + values -> values |> Enum.map(&to_seconds/1) |> Enum.min() |> rt_secs_to_idx()
553 + end
554 +
555 + {kinds, idx}
496 556 end)
497 557
498 558 {:noreply,
499 559 socket
500 560 |> assign(:run_period_idx, run_period_idx)
561 + |> assign(:rt_groups, rt_groups)
501 562 |> assign(:rt_idxs, rt_idxs)
502 563 |> assign(:rt_requests, rt_requests_per_hour(config_for_estimate, rt_idxs))
503 564 |> assign(:github_poll_idx, gh_idx)
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/form_component.html.heex
+13 −13
@@ -89,23 +89,23 @@
89 89 </span>
90 90 </label>
91 91 </div>
92 − <%= for {kind, title} <- RoomSanctumWeb.SourceLive.FormComponent.rt_period_kinds() do %>
92 + <%= for {_url, kinds} <- @rt_groups do %>
93 93 <div class="form-control mt-2">
94 94 <label class="label">
95 95 <span class="label-text">
96 − <%= title %> interval &mdash;
96 + <%= RoomSanctumWeb.SourceLive.FormComponent.rt_group_label(kinds) %> interval &mdash;
97 97 <span class="font-bold">
98 − <%= RoomSanctumWeb.SourceLive.FormComponent.rt_period_label(@rt_idxs[kind]) %>
98 + <%= RoomSanctumWeb.SourceLive.FormComponent.rt_period_label(@rt_idxs[kinds]) %>
99 99 </span>
100 100 </span>
101 101 </label>
102 102 <input type="range"
103 − name={"source[config][rt_idx_#{kind}]"}
104 − id={"source_config_rt_idx_#{kind}"}
103 + name={"source[config][#{RoomSanctumWeb.SourceLive.FormComponent.rt_group_key(kinds)}]"}
104 + id={"source_config_#{RoomSanctumWeb.SourceLive.FormComponent.rt_group_key(kinds)}"}
105 105 min="0"
106 106 max={length(RoomSanctumWeb.SourceLive.FormComponent.rt_period_steps()) - 1}
107 107 step="1"
108 − value={@rt_idxs[kind]}
108 + value={@rt_idxs[kinds]}
109 109 class="range range-sm range-secondary" />
110 110 <div class="flex justify-between text-xs px-1 mt-1">
111 111 <%= for {label, _secs} <- RoomSanctumWeb.SourceLive.FormComponent.rt_period_steps() do %>
@@ -114,15 +114,15 @@
114 114 </div>
115 115 </div>
116 116 <% end %>
117 − <p class="text-xs text-base-content/60 mt-1">
118 − The three kinds go stale at different rates: trip updates in seconds, service
119 − alerts a few times a day. Feeds sharing one URL are fetched once between them.
117 + <p :if={@rt_groups != []} class="text-xs text-base-content/60 mt-1">
118 + One control per feed URL, since a URL is what gets fetched. Kinds arriving in
119 + the same feed are refreshed together &mdash; scheduling them apart would fetch
120 + that feed on the fastest of their clocks and then hand the others a staler copy
121 + of the message that had already arrived.
120 122 </p>
121 − <div class="text-xs mt-2 opacity-80">
123 + <div :if={@rt_groups != []} class="text-xs mt-2 opacity-80">
122 124 <i class="fa-solid fa-fw fa-gauge-high"></i>
123 − Roughly
124 − <span class="font-bold"><%= @rt_requests %></span>
125 − requests an hour, counting each distinct URL once.
125 + Roughly <span class="font-bold"><%= @rt_requests %></span> requests an hour.
126 126 511.org allows 60 on a default token.
127 127 </div>
128 128 <div class="form-control">

Parents: 3b86b77