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.

Run GTFS static imports one at a time, through Oban

ae13c56 · Gabriel Morell · 2026-08-24 04:43

14 files +255 -68
Message 
{commit_body(@commit)}

Files changed

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+125 −67
@@ -31,10 +31,16 @@ defmodule RoomGtfs.Worker do
31 31 |> GenServer.cast(:scheduled_static)
32 32 end
33 33
34 + @doc """
35 + Ask for a source's static feed to be reimported.
36 +
37 + This queues rather than starts. Every caller — the nightly scheduler and the
38 + button on the source page both — lands here, and `RoomGtfs.ImportJob` decides
39 + when the import actually runs; see that module for why. A source that already
40 + has an import queued or running does not get a second one.
41 + """
34 42 def update_static_data(name) do
35 − "gtfs#{name}"
36 − |> via_tuple()
37 − |> GenServer.cast(:update_static)
43 + RoomGtfs.ImportJob.enqueue(name)
38 44 end
39 45
40 46 def update_static_data(name, :str) do
@@ -726,17 +732,25 @@ defmodule RoomGtfs.Worker.Static do
726 732 end
727 733
728 734 defp bcast(id, file, complete, total) do
729 − Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, id, file, complete, total})
735 + Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, wire_id(id), file, complete, total})
730 736 end
731 737
732 738 defp bcast(id, :disabled) do
733 − Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, id, :disabled})
739 + Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, wire_id(id), :disabled})
734 740 end
735 741
736 742 defp bcast(id, :done) do
737 − Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, id, :done})
743 + Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, wire_id(id), :done})
738 744 end
739 745
746 + # The source page does `String.to_integer/1` on the id it receives, so these
747 + # messages have always carried a string — which they did by accident, because
748 + # the only caller was a GenServer whose name is the id as a string. The
749 + # importer is now handed an integer by the job that runs it, so the shape is
750 + # pinned here rather than left to whoever happens to call.
751 + defp wire_id(id) when is_binary(id), do: id
752 + defp wire_id(id) when is_integer(id), do: Integer.to_string(id)
753 +
740 754 def init(opts) do
741 755
742 756 # pgopts = RoomSanctum.Repo.config()
@@ -1238,81 +1252,125 @@ end
1238 1252 "gtfs_#{atom}"
1239 1253 end
1240 1254
1241 − @impl true
1242 − def handle_cast(:update_static, state) do
1243 − cfg = Configuration.get_source!(state.id)
1255 + @doc """
1256 + Import a source's static feed, start to finish, in the calling process.
1257 +
1258 + This is the slow half of GTFS: download a zip, then COPY eight files into
1259 + Postgres, of which stop_times runs to millions of rows and is preceded by a
1260 + delete of the millions already there. It is deliberately synchronous — the
1261 + caller is held for the whole import, which is what lets `RoomGtfs.ImportJob`
1262 + bound how many run at once. Casting this at a GenServer instead, as the
1263 + scheduler used to, means the cast returns immediately and nothing anywhere
1264 + knows how many imports are in flight.
1265 +
1266 + Returns `:ok` once the feed has been written, or `{:error, reason}` for a
1267 + failure that happened before anything was written — a download that failed, a
1268 + body that was not a zip. Those are safe to retry. A file that fails *during*
1269 + the load is logged and the import still finishes and stamps `last_run`, which
1270 + is long-standing behaviour: a retry would truncate and reload the seven files
1271 + that did work.
1272 + """
1273 + def import_static(id) do
1274 + cfg = Configuration.get_source!(id)
1244 1275
1245 − case cfg.enabled do
1246 − true ->
1247 − Logger.info("GTFS::#{state.id} updating static info")
1248 − bcast(state.id, :downloading, 1, 10)
1276 + if cfg.enabled do
1277 + do_import_static(id, cfg)
1278 + else
1279 + # Previously a `case cfg.enabled do true -> ... end`, which raised
1280 + # CaseClauseError on a disabled source. Harmless when it was a cast into
1281 + # a GenServer that restarted; as a queued job it would fail, retry and
1282 + # fail again.
1283 + Logger.info("GTFS::#{id} static import skipped, source is disabled")
1284 + :ok
1285 + end
1286 + end
1249 1287
1250 − case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
1251 − {:ok, result} ->
1252 − bcast(state.id, :extracting, 2, 10)
1288 + defp do_import_static(id, cfg) do
1289 + Logger.info("GTFS::#{id} updating static info")
1290 + bcast(id, :downloading, 1, 10)
1253 1291
1254 − case result.body |> Unzip.InMem.new() |> Unzip.new() do
1255 − {:ok, unzip} ->
1256 − files = Unzip.list_entries(unzip)
1257 −
1258 − case Enum.find(files, &(&1.file_name == "linked_datasets.txt")) do
1259 − nil ->
1260 − :ok
1261 −
1262 − entry ->
1263 − Unzip.file_stream!(unzip, entry.file_name)
1264 − |> Enum.to_list()
1265 − |> IO.iodata_to_binary()
1266 − |> parse_linked_datasets()
1267 − |> linked_dataset_urls()
1268 − |> then(&apply_linked_datasets(cfg, &1))
1269 − end
1292 + case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
1293 + {:ok, result} ->
1294 + bcast(id, :extracting, 2, 10)
1270 1295
1271 − try do
1272 − files
1273 − |> Enum.map(fn e ->
1274 − if Enum.member?(
1275 − [
1276 − "agency.txt",
1277 − "calendar.txt",
1278 − "directions.txt",
1279 − "routes.txt",
1280 − "stops.txt",
1281 − "stop_times.txt",
1282 − "trips.txt",
1283 − "shapes.txt",
1284 − ],
1285 − e.file_name
1286 − ) do
1287 −
1288 − bcast(state.id, file_to_atom(e.file_name), file_to_order(e.file_name), 10)
1289 −
1290 − Unzip.file_stream!(unzip, e.file_name)
1291 − |> write_file(file_to_atom(e.file_name), state.id, nil)
1292 − end
1293 − end)
1294 − rescue
1295 − e ->
1296 − Logger.error("GTFS::#{state.id} error during static import: #{inspect(e)}")
1297 − after
1298 − Configuration.update_source_meta(cfg, %{last_run: DateTime.utc_now()})
1299 − bcast(state.id, :done)
1300 − end
1296 + case result.body |> Unzip.InMem.new() |> Unzip.new() do
1297 + {:ok, unzip} ->
1298 + files = Unzip.list_entries(unzip)
1301 1299
1302 − {:error, term} ->
1303 − Logger.error(term)
1300 + case Enum.find(files, &(&1.file_name == "linked_datasets.txt")) do
1301 + nil ->
1302 + :ok
1303 +
1304 + entry ->
1305 + Unzip.file_stream!(unzip, entry.file_name)
1306 + |> Enum.to_list()
1307 + |> IO.iodata_to_binary()
1308 + |> parse_linked_datasets()
1309 + |> linked_dataset_urls()
1310 + |> then(&apply_linked_datasets(cfg, &1))
1304 1311 end
1305 − {:error, error} -> Logger.error(error)
1312 +
1313 + try do
1314 + files
1315 + |> Enum.map(fn e ->
1316 + if Enum.member?(
1317 + [
1318 + "agency.txt",
1319 + "calendar.txt",
1320 + "directions.txt",
1321 + "routes.txt",
1322 + "stops.txt",
1323 + "stop_times.txt",
1324 + "trips.txt",
1325 + "shapes.txt",
1326 + ],
1327 + e.file_name
1328 + ) do
1329 +
1330 + bcast(id, file_to_atom(e.file_name), file_to_order(e.file_name), 10)
1331 +
1332 + Unzip.file_stream!(unzip, e.file_name)
1333 + |> write_file(file_to_atom(e.file_name), id, nil)
1334 + end
1335 + end)
1336 + rescue
1337 + e ->
1338 + Logger.error("GTFS::#{id} error during static import: #{inspect(e)}")
1339 + after
1340 + Configuration.update_source_meta(cfg, %{last_run: DateTime.utc_now()})
1341 + bcast(id, :done)
1342 + end
1343 +
1344 + :ok
1345 +
1346 + {:error, term} ->
1347 + # `Logger.error/1` takes chardata, not a struct: passing the
1348 + # raw term raised here, inside the branch meant to report the
1349 + # error.
1350 + Logger.error("GTFS::#{id} static feed was not a readable zip: #{inspect(term)}")
1351 + {:error, {:unzip, term}}
1306 1352 end
1307 − end
1308 1353
1309 − {:noreply, state}
1354 + {:error, error} ->
1355 + Logger.error("GTFS::#{id} static feed download failed: #{inspect(error)}")
1356 + {:error, {:download, error}}
1357 + end
1310 1358 end
1311 1359
1312 1360 defp replace(string) do
1313 1361 String.replace(string, ~s("), "")
1314 1362 end
1315 1363
1364 + # Runs the import immediately, in this GenServer, bypassing the queue.
1365 + # Nothing in the app reaches this any more — `update_static_data/1` enqueues
1366 + # instead — but it is left as the way to force one feed from IEx without
1367 + # waiting behind whatever else is queued.
1368 + @impl true
1369 + def handle_cast(:update_static, state) do
1370 + import_static(state.id)
1371 + {:noreply, state}
1372 + end
1373 +
1316 1374 @impl true
1317 1375 def handle_cast(:update_static_old, state) do
1318 1376 cfg = Configuration.get_source!(state.id)
added apps/room_gtfs/lib/room_gtfs/import_job.ex
+113 −0
@@ -0,0 +1,113 @@
1 +defmodule RoomGtfs.ImportJob do
2 + @moduledoc """
3 + The queue in front of GTFS static imports.
4 +
5 + ## What this fixes
6 +
7 + A static import downloads a feed and COPYs eight files into Postgres. For a
8 + large agency that is a couple of million `gtfs_stop_times` rows, preceded by
9 + a delete of the couple of million already there. One at a time, that is a
10 + slow but unremarkable bulk load.
11 +
12 + The trouble was that nothing said "one at a time". Every GTFS source runs its
13 + own worker with its own scheduler, and every one of those schedulers checks
14 + at the same instant — `Periodic` firing on `%Time{hour: 0, minute: 0}` — so
15 + every feed that happened to be due began importing together. Four concurrent
16 + bulk loads against the same eight tables is what made Postgres miserable.
17 +
18 + Pool size was never a brake on this, which is worth knowing before assuming
19 + it was: `write_file/4` opens its own `Postgrex` connection per file rather
20 + than checking one out of the Ecto pool, so `pool_size` bounded none of it.
21 +
22 + The scheduling has not moved. Each worker still decides *whether* its feed is
23 + due, on its own `run_period` and `last_run`. This decides *when* the import
24 + it asked for actually runs, and the answer is "when the one before it has
25 + finished" — the `gtfs_import` queue is configured with a concurrency of 1 in
26 + `config/config.exs`, and that single number is the entire mechanism. Raising
27 + it is how you trade Postgres load for getting through the feeds faster.
28 +
29 + ## One import per source
30 +
31 + Uniqueness covers everything not yet finished — queued, scheduled, running,
32 + waiting to retry — so a source can have at most one import outstanding.
33 + Without it the midnight check firing twice inside its minute, or someone
34 + pressing the button on the source page while the nightly run is still
35 + queued, would enqueue a second import of a feed whose tables the first one is
36 + midway through filling. Both would then truncate and reload the same rows.
37 +
38 + Completed jobs do not block anything, so tomorrow's run enqueues normally.
39 +
40 + ## Where to watch it
41 +
42 + The `gtfs_import` queue reports through the same PromEx Oban plugin as every
43 + other queue, so the Background jobs row of the apollos-crib dashboard picks
44 + it up with no change — queue depth there is how long the feeds are waiting.
45 + Per-feed progress still broadcasts on the `"gtfs"` PubSub topic exactly as
46 + before; moving the work into a job process did not change that.
47 + """
48 +
49 + use Oban.Worker,
50 + queue: :gtfs_import,
51 + max_attempts: 3,
52 + unique: [
53 + keys: [:source_id],
54 + states: [:available, :scheduled, :executing, :retryable],
55 + period: :infinity
56 + ]
57 +
58 + require Logger
59 +
60 + @doc """
61 + Queue a static import for a source.
62 +
63 + Returns `{:ok, job}` whether or not it actually enqueued anything: a job that
64 + collides with one already outstanding for the same source comes back with
65 + `conflict?: true` rather than as an error, so callers do not have to care
66 + which it was.
67 + """
68 + def enqueue(source_id) do
69 + id = normalize_id(source_id)
70 +
71 + with {:ok, job} <- Oban.insert(new(%{source_id: id})) do
72 + # Tell the source page something happened. Pressing update used to start
73 + # the download immediately, so the progress bar moved within a second or
74 + # two; now the import may sit behind another feed for a while, and with
75 + # nothing broadcast until it starts the button looks broken.
76 + if not job.conflict? do
77 + Phoenix.PubSub.broadcast(
78 + RoomSanctum.PubSub,
79 + "gtfs",
80 + {:gtfs, Integer.to_string(id), :queued, 0, 10}
81 + )
82 + end
83 +
84 + {:ok, job}
85 + end
86 + end
87 +
88 + # Callers arrive with two different shapes: the scheduler holds the worker's
89 + # registry name, which is the source id as a string, while the source page
90 + # hands over an integer parsed from the URL. Normalising matters more than it
91 + # looks — uniqueness compares args, and `%{source_id: 1}` and
92 + # `%{source_id: "1"}` are different args, so the two paths would not see each
93 + # other and both would enqueue.
94 + defp normalize_id(id) when is_integer(id), do: id
95 + defp normalize_id(id) when is_binary(id), do: String.to_integer(id)
96 +
97 + @impl Oban.Worker
98 + # A large feed genuinely takes tens of minutes. The default is no timeout at
99 + # all, which at a concurrency of 1 means a wedged download holds the only
100 + # slot indefinitely and no other feed imports again until the node restarts.
101 + def timeout(_job), do: :timer.minutes(45)
102 +
103 + @impl Oban.Worker
104 + def perform(%Oban.Job{args: %{"source_id" => id}}) do
105 + RoomGtfs.Worker.Static.import_static(id)
106 + rescue
107 + Ecto.NoResultsError ->
108 + # Deleted between enqueue and execution. Retrying cannot bring it back,
109 + # so stop now rather than failing twice more first.
110 + Logger.info("GTFS::#{id} static import cancelled, source no longer exists")
111 + {:cancel, "source #{id} no longer exists"}
112 + end
113 +end
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/show.ex
+11 −0
@@ -1287,6 +1287,17 @@ defmodule RoomSanctumWeb.SourceLive.Show do
1287 1287 |> assign(:status_val, 0)
1288 1288 }
1289 1289
1290 + # GTFS imports run one at a time through Oban, so an update that was
1291 + # asked for has not necessarily started. Without this the button reads as
1292 + # doing nothing until the feed reaches the front of the queue.
1293 + :queued ->
1294 + {
1295 + :noreply,
1296 + socket
1297 + |> assign(:status, "Queued")
1298 + |> assign(:status_val, 0)
1299 + }
1300 +
1290 1301 _ ->
1291 1302 case complete == total do
1292 1303 true ->
modified config/config.exs
+6 −1
@@ -90,7 +90,12 @@ config :logger,
90 90 config :phoenix, :json_library, Jason
91 91 config :room_sanctum, Oban,
92 92 engine: Oban.Engines.Basic,
93 − queues: [default: 10, webhooks: 20, emails: 20],
93 + # gtfs_import is 1 on purpose. A GTFS static import bulk-loads millions of
94 + # rows through its own Postgrex connections, and several at once — which
95 + # is what happened when every source scheduled itself for midnight —
96 + # buries Postgres. See RoomGtfs.ImportJob. Raising this trades database
97 + # load for getting through the feeds sooner.
98 + queues: [default: 10, webhooks: 20, emails: 20, gtfs_import: 1],
94 99 repo: RoomSanctum.Repo
95 100 config :messenger, smtp_opts: [[port: 2525]]
96 101

Parents: 3f6ff12