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.

Stop the realtime worker reading a whole country every five minutes

a1ad701 · Gabriel Morell · 2026-09-08 16:23

15 files +152 -14
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/application.ex
+12 −1
@@ -31,7 +31,7 @@ defmodule RoomGtfs.Application do
31 31 # start callback, it is.
32 32 :ok = Protobuf.load_extensions()
33 33
34 − children = [static_pool(), RoomGtfs.FeedCache, RoomGtfs.RTIndex]
34 + children = [static_pool(), rt_pool(), RoomGtfs.FeedCache, RoomGtfs.RTIndex]
35 35
36 36 Supervisor.start_link(children, strategy: :one_for_one, name: RoomGtfs.Supervisor)
37 37 end
@@ -48,6 +48,17 @@ defmodule RoomGtfs.Application do
48 48 # Separate from hackney's default pool for the same reason. Everything else
49 49 # in this umbrella -- realtime feeds, GitHub, the weather -- shares that one,
50 50 # and a bulk transfer sitting in it starves requests that take milliseconds.
51 + # The realtime feeds' pool.
52 + #
53 + # Bigger than the static one and with a much shorter timeout: these are small
54 + # requests made often, where the static pool is a few enormous ones. Separate
55 + # from hackney's default so that a slow transit feed cannot exhaust the pool
56 + # every other client in this umbrella is also using -- which is what the
57 + # `:checkout_timeout`s on GitHub and the weather were.
58 + defp rt_pool do
59 + :hackney_pool.child_spec(:gtfs_rt, max_connections: 32, timeout: 20_000)
60 + end
61 +
51 62 defp static_pool do
52 63 :hackney_pool.child_spec(:gtfs_static, max_connections: 4, timeout: 300_000)
53 64 end
modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+64 −13
@@ -740,7 +740,18 @@ defmodule RoomGtfs.Worker.RT do
740 740 # A realtime feed is a protobuf of a few hundred kilobytes, fetched on a
741 741 # tick. Explicit rather than left to HTTPoison's five seconds, which is the
742 742 # right number here but should not be a coincidence.
743 − @rt_http [follow_redirect: true, recv_timeout: 10_000, timeout: 8_000]
743 + # A pool of its own, for the same reason the static downloads have one.
744 + # Realtime feeds are small but there are many of them and some are slow --
745 + # Germany's three sources poll a combined feed each, and a slow one holds its
746 + # connection for the full ten seconds. On hackney's shared default pool that
747 + # showed up as `:checkout_timeout` on unrelated things: GitHub, the weather,
748 + # and the German feeds starving each other.
749 + @rt_http [
750 + follow_redirect: true,
751 + recv_timeout: 10_000,
752 + timeout: 8_000,
753 + hackney: [pool: :gtfs_rt]
754 + ]
744 755 use GenServer
745 756 @registry :zeus
746 757
@@ -1123,33 +1134,73 @@ defmodule RoomGtfs.Worker.RT do
1123 1134 end
1124 1135 end
1125 1136
1126 − # Static stops for this source, as the fallback above needs them. Reloaded at
1127 − # most every few minutes: they only change when a feed is reimported, and a
1128 − # subway feed is 1,488 rows that would otherwise be read on every poll.
1129 − defp ensure_stops(state) do
1137 + # Positions for the stops this feed actually names, as the fallback above
1138 + # needs them.
1139 + #
1140 + # It used to read every stop the source had. That was 1,488 rows for a subway
1141 + # feed and half a million for a national one, on a five minute timer, per
1142 + # source -- a query slow enough to hold a database connection until it timed
1143 + # out, which starved the pool and crashed this worker on a loop.
1144 + #
1145 + # Only a vehicle that reports a stop and no coordinates needs a lookup at
1146 + # all, and there are rarely many of those. What is fetched is now bounded by
1147 + # what the feed asks about rather than by how large the country is.
1148 + #
1149 + # Still capped by the same five minutes, which is what picks up coordinates
1150 + # that changed in a reimport: past it the cache is dropped rather than added
1151 + # to.
1152 + defp ensure_stops(state, feed) do
1130 1153 now = System.monotonic_time(:millisecond)
1131 1154
1132 1155 # Matched rather than defaulted: the key is present from init with a value of
1133 1156 # nil, so Map.get/3's default never fired and the first poll of every worker
1134 1157 # arithmetic'd against nil.
1135 − stale? =
1158 + expired? =
1136 1159 case state.stops_at do
1137 1160 nil -> true
1138 1161 at -> now - at >= @stops_ttl
1139 1162 end
1140 1163
1141 − if stale? do
1142 − stops =
1143 − state.id
1144 − |> Storage.list_stops()
1145 − |> Map.new(fn stop -> {stop.stop_id, {stop.stop_lat, stop.stop_lon}} end)
1164 + known = if expired?, do: %{}, else: state.stops || %{}
1165 + wanted = feed |> stops_wanted() |> Enum.reject(&Map.has_key?(known, &1))
1146 1166
1147 − state |> Map.put(:stops, stops) |> Map.put(:stops_at, now)
1167 + if wanted == [] and not expired? do
1168 + state
1148 1169 else
1170 + loaded = Storage.stop_positions(state.id, wanted)
1171 +
1149 1172 state
1173 + |> Map.put(:stops, Map.merge(known, loaded))
1174 + |> Map.put(:stops_at, now)
1150 1175 end
1151 1176 end
1152 1177
1178 + @doc false
1179 + # Public only so a test can reach it; nothing outside calls this.
1180 + def stops_wanted_for_test(feed), do: stops_wanted(feed)
1181 +
1182 + # The stops a feed needs looked up: those named by a vehicle that did not say
1183 + # where it is. A vehicle carrying its own coordinates never consults these.
1184 + defp stops_wanted(feed) do
1185 + feed.entity
1186 + |> Enum.flat_map(fn entity ->
1187 + case entity.vehicle do
1188 + %{stop_id: stop_id} = vehicle when is_binary(stop_id) ->
1189 + if positioned?(vehicle), do: [], else: [stop_id]
1190 +
1191 + _ ->
1192 + []
1193 + end
1194 + end)
1195 + |> Enum.uniq()
1196 + end
1197 +
1198 + defp positioned?(%{position: %{latitude: lat, longitude: lon}})
1199 + when not is_nil(lat) and not is_nil(lon),
1200 + do: true
1201 +
1202 + defp positioned?(_vehicle), do: false
1203 +
1153 1204
1154 1205 def fetch_rt_url(url, headers \\ []) do
1155 1206 case HTTPoison.get(url, headers, @rt_http) do
@@ -1473,7 +1524,7 @@ defmodule RoomGtfs.Worker.RT do
1473 1524 end
1474 1525
1475 1526 defp store_feed(state, :vp, feed) do
1476 − state = state |> Map.put(:rt_vp, feed) |> ensure_stops()
1527 + state = state |> Map.put(:rt_vp, feed) |> ensure_stops(feed)
1477 1528 vehicles = vehicle_positions_from(feed, state.stops)
1478 1529 RoomGtfs.RTIndex.put_vehicles(state.id, vehicles)
1479 1530
added apps/room_gtfs/test/stops_wanted_test.exs
+50 −0
@@ -0,0 +1,50 @@
1 +defmodule RoomGtfs.StopsWantedTest do
2 + @moduledoc """
3 + What the realtime worker asks the database about.
4 +
5 + These are the inputs that decide the size of a query which, unbounded, took
6 + the worker down against a national feed.
7 + """
8 + use ExUnit.Case, async: true
9 +
10 + alias TransitRealtime.{FeedEntity, FeedMessage, Position, VehiclePosition}
11 +
12 + defp feed(vehicles) do
13 + %FeedMessage{
14 + header: %TransitRealtime.FeedHeader{gtfs_realtime_version: "2.0"},
15 + entity: Enum.map(vehicles, &%FeedEntity{id: "e", vehicle: &1})
16 + }
17 + end
18 +
19 + defp wanted(feed), do: :erlang.apply(RoomGtfs.Worker.RT, :stops_wanted_for_test, [feed])
20 +
21 + test "a vehicle that says where it is needs no lookup" do
22 + f = feed([%VehiclePosition{stop_id: "S1", position: %Position{latitude: 1.0, longitude: 2.0}}])
23 +
24 + assert wanted(f) == []
25 + end
26 +
27 + test "a vehicle that names a stop and no position needs that stop" do
28 + assert wanted(feed([%VehiclePosition{stop_id: "S1", position: nil}])) == ["S1"]
29 + end
30 +
31 + test "a vehicle with neither is not a lookup" do
32 + assert wanted(feed([%VehiclePosition{stop_id: nil, position: nil}])) == []
33 + end
34 +
35 + test "a stop named by many vehicles is asked for once" do
36 + vehicles = for _ <- 1..50, do: %VehiclePosition{stop_id: "S1", position: nil}
37 +
38 + assert wanted(feed(vehicles)) == ["S1"]
39 + end
40 +
41 + test "a position missing a coordinate is not a position" do
42 + f = feed([%VehiclePosition{stop_id: "S1", position: %Position{latitude: nil, longitude: 2.0}}])
43 +
44 + assert wanted(f) == ["S1"]
45 + end
46 +
47 + test "an empty feed asks for nothing" do
48 + assert wanted(feed([])) == []
49 + end
50 +end
modified apps/room_sanctum/lib/room_sanctum/storage.ex
+26 −0
@@ -1057,6 +1057,32 @@ defmodule RoomSanctum.Storage do
1057 1057 |> Repo.all()
1058 1058 end
1059 1059
1060 + @doc """
1061 + Just these stops of a source, as `%{stop_id => {lat, lon}}`.
1062 +
1063 + For the realtime worker, which needs a position for the handful of vehicles
1064 + that report which stop they are at and not where that is. It used to read
1065 + every stop the source had for that -- fine for a subway feed's 1,488 rows,
1066 + and a slow query holding a database connection for a national feed's half
1067 + million, which is what saturated the pool and took the worker down with it.
1068 +
1069 + Chunked, because `stop_id in (...)` with tens of thousands of ids is its own
1070 + kind of enormous query, and selected down to the three columns the caller
1071 + actually uses.
1072 + """
1073 + def stop_positions(source_id, stop_ids) do
1074 + stop_ids
1075 + |> Enum.chunk_every(5_000)
1076 + |> Enum.reduce(%{}, fn chunk, acc ->
1077 + from(p in Stop,
1078 + where: p.source_id == ^source_id and p.stop_id in ^chunk,
1079 + select: {p.stop_id, p.stop_lat, p.stop_lon}
1080 + )
1081 + |> Repo.all()
1082 + |> Enum.reduce(acc, fn {id, lat, lon}, inner -> Map.put(inner, id, {lat, lon}) end)
1083 + end)
1084 + end
1085 +
1060 1086 @doc """
1061 1087 A page of a source's stops, after `cursor`, in stop_id order.
1062 1088

Parents: 74f3450