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.

Stream a source's stops in pages instead of capping them

5730d3d · Gabriel Morell · 2026-09-07 03:05

15 files +200 -12
Message 
{commit_body(@commit)}

Files changed

modified apps/room_sanctum/lib/room_sanctum/storage.ex
+18 −0
@@ -1057,6 +1057,24 @@ defmodule RoomSanctum.Storage do
1057 1057 |> Repo.all()
1058 1058 end
1059 1059
1060 + @doc """
1061 + A page of a source's stops, after `cursor`, in stop_id order.
1062 +
1063 + Keyset rather than offset: `OFFSET 400000` makes Postgres walk and discard
1064 + four hundred thousand rows to reach the last page, so a paged read gets
1065 + slower the further it goes and ends up costing more than the single query it
1066 + replaced. `stop_id > cursor` starts where the last page stopped.
1067 +
1068 + There has been a unique index on (source_id, stop_id) since this table was
1069 + created, which is exactly this access pattern -- so each page is an index
1070 + range scan of `limit` rows, whatever page it is.
1071 + """
1072 + def stops_after(source_id, cursor, limit) do
1073 + from(p in Stop, where: p.source_id == ^source_id, order_by: p.stop_id, limit: ^limit)
1074 + |> then(fn q -> if cursor, do: where(q, [p], p.stop_id > ^cursor), else: q end)
1075 + |> Repo.all()
1076 + end
1077 +
1060 1078 @doc """
1061 1079 The first `limit` stops of a source, for callers that cannot use all of them.
1062 1080
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/show.ex
+47 −7
@@ -4,6 +4,7 @@ defmodule RoomSanctumWeb.SourceLive.Show do
4 4
5 5 alias RoomSanctum.Configuration
6 6 alias RoomSanctum.Storage
7 + alias RoomSanctumWeb.SourceLive.StationStream
7 8 alias RoomSanctumWeb.SourceLive.Stations
8 9
9 10 @impl true
@@ -34,6 +35,7 @@ defmodule RoomSanctumWeb.SourceLive.Show do
34 35 # first tick, and a template reading an assign that is not there yet is
35 36 # a KeyError on the very first paint.
36 37 |> assign(:station_count, 0)
38 + |> assign(:stations_loading, false)
37 39 # Both are otherwise only set by the :update_sec tick 200ms in, so any
38 40 # view touching them could be rendered before they exist.
39 41 |> assign(:station_statuses, [])
@@ -78,11 +80,6 @@ defmodule RoomSanctumWeb.SourceLive.Show do
78 80 _ -> []
79 81 end
80 82
81 − stations = Stations.for_source(socket.assigns.source)
82 −
83 − # What the map is not showing. Counted rather than inferred from the list,
84 − # which cannot tell a source with exactly the cap from one with far more.
85 − station_count = Stations.count_for_source(socket.assigns.source)
86 83
87 84 # Add station status for GBFS sources
88 85 station_statuses = Stations.statuses_for_source(socket.assigns.source)
@@ -111,8 +108,6 @@ defmodule RoomSanctumWeb.SourceLive.Show do
111 108 |> assign(:queries, queries)
112 109 |> assign(:available_tints, available_tints)
113 110 |> assign(:free_bikes, free_bikes)
114 − |> assign(:stations, stations)
115 − |> assign(:station_count, station_count)
116 111 |> assign(:station_statuses, station_statuses)
117 112 |> assign(:route_types, route_types)
118 113 |> assign(:aircraft, aircraft)
@@ -121,6 +116,50 @@ defmodule RoomSanctumWeb.SourceLive.Show do
121 116 end
122 117
123 118 @impl true
119 + # Read the source's places in the background, folding each page in as it
120 + # lands. Only when connected: the dead render has no process to send to and
121 + # no client to paint, and starting one would read a quarter of a million rows
122 + # for a page nobody is looking at yet.
123 + defp stream_stations(socket, source) do
124 + if connected?(socket) and Stations.mappable?(source) do
125 + # Linked, so a reader whose page has gone dies with it rather than
126 + # finishing into a dead mailbox.
127 + {:ok, _pid} = StationStream.start_link(source: source, to: self())
128 +
129 + socket
130 + |> assign(:stations, [])
131 + |> assign(:stations_loading, true)
132 + |> assign(:station_count, Stations.count_for_source(source))
133 + else
134 + socket
135 + end
136 + end
137 +
138 + # A page of places, folded in as it lands.
139 + #
140 + # `batch ++ stations` rather than `stations ++ batch`: ++ walks its left
141 + # side, so appending onto the accumulated list costs more with every page and
142 + # a quarter of a million stops becomes quadratic. Prepending walks only the
143 + # new page, which is the same twenty thousand each time. The map does not
144 + # care which order its markers arrive in.
145 + def handle_info({:stations_batch, source_id, batch}, socket) do
146 + if source_id == socket.assigns.source_id do
147 + {:noreply, assign(socket, :stations, batch ++ socket.assigns.stations)}
148 + else
149 + # A page navigated away from mid-read; its reader is on its way out.
150 + {:noreply, socket}
151 + end
152 + end
153 +
154 + def handle_info({:stations_done, source_id, total}, socket) do
155 + if source_id == socket.assigns.source_id do
156 + {:noreply,
157 + socket |> assign(:stations_loading, false) |> assign(:station_count, total)}
158 + else
159 + {:noreply, socket}
160 + end
161 + end
162 +
124 163 def handle_info(:update_tester, socket) do
125 164 Process.send_after(self(), :update_tester, 2000)
126 165 case socket.assigns.tester_selected do
@@ -169,6 +208,7 @@ defmodule RoomSanctumWeb.SourceLive.Show do
169 208 socket
170 209 |> assign(:page_title, page_title(socket.assigns.live_action))
171 210 |> assign(:source, source)
211 + |> stream_stations(source)
172 212 |> assign(:source_id, source_id)
173 213 |> assign(:queries, queries)
174 214 |> assign(:available_tints, get_available_tints(queries))
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/show.html.heex
+16 −5
@@ -473,6 +473,13 @@ gl_available_projects={@gitlab_available_projects}
473 473 <%= cond do %>
474 474 <% @view_mode == :detail -> %>
475 475 <div class="px-6">
476 + <%!-- max_markers: loading every stop and drawing every
477 + stop are different questions. All of them are read,
478 + so lookups and counts are complete; the map draws a
479 + balanced sample, because half a million custom
480 + elements is a dead browser however well Leaflet
481 + renders them. The component says how many it
482 + dropped. --%>
476 483 <.query_geospatial_map
477 484 queries={filter_queries_by_tint(@queries, @tint)}
478 485 vehicle_positions={@vehicle_positions}
@@ -490,12 +497,16 @@ gl_available_projects={@gitlab_available_projects}
490 497 height="600px"
491 498 add_query_event="map-add-query"
492 499 queried_station_ids={Enum.map(@queries, &station_id_of/1)}
500 + max_markers={5000}
493 501 />
494 − <%!-- A national feed has stops by the hundred thousand.
495 − The map draws a bounded subset; saying so is better
496 − than a map that looks complete and is not. --%>
497 − <p :if={@station_count > length(@stations)} class="text-xs opacity-60 mt-1">
498 − Showing <%= length(@stations) %> of <%= @station_count %> stops.
502 + <%!-- A national feed has stops by the hundred thousand and
503 + they arrive in pages, so the map fills in rather than
504 + waiting. Saying how far along says why it is still
505 + moving -- otherwise a half-drawn map reads as a
506 + broken one. --%>
507 + <p :if={@stations_loading} class="text-xs opacity-60 mt-1">
508 + <i class="fa-solid fa-spinner fa-spin-pulse mr-1"></i>
509 + Loading stops — <%= length(@stations) %> of <%= @station_count %> so far.
499 510 </p>
500 511 </div>
501 512
added apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/station_stream.ex
+66 −0
@@ -0,0 +1,66 @@
1 +defmodule RoomSanctumWeb.SourceLive.StationStream do
2 + @moduledoc """
3 + Reads a source's places in pages and posts them to a LiveView as they arrive.
4 +
5 + A national GTFS feed has stops by the hundred thousand. Fetching them in one
6 + query took longer than the fifteen second checkout timeout, which killed the
7 + LiveView -- and it did it on a ten second timer, so the page crashed, came
8 + back, and crashed again.
9 +
10 + Paging fixes the part that was actually broken. No single query is long
11 + enough to time out, the connection is returned between pages rather than held
12 + across the whole read, and the page paints its first twenty thousand stops
13 + while the rest are still coming.
14 +
15 + ## Why a process rather than a loop in the LiveView
16 +
17 + A LiveView that loops is a LiveView that is not answering anything else --
18 + no clicks, no ticks, no navigation, until the last page lands. Reading in
19 + its own process leaves the page live throughout, and the batches arrive as
20 + ordinary messages it can choose to fold in.
21 +
22 + Linked to the LiveView on purpose: a reader whose page has gone is reading
23 + for nobody, and should go with it rather than finish a quarter of a million
24 + rows into a dead mailbox.
25 + """
26 + use GenServer, restart: :temporary
27 +
28 + alias RoomSanctumWeb.SourceLive.Stations
29 +
30 + # Big enough that a national feed is tens of queries rather than thousands,
31 + # small enough that one page is nowhere near the checkout timeout.
32 + @batch 20_000
33 +
34 + @doc """
35 + Start reading `source`, posting batches to `to`.
36 +
37 + Sends `{:stations_batch, source_id, stations}` per page and
38 + `{:stations_done, source_id, total}` at the end.
39 + """
40 + def start_link(opts) do
41 + GenServer.start_link(__MODULE__, Map.new(opts))
42 + end
43 +
44 + @impl true
45 + def init(%{source: source, to: to}) do
46 + {:ok, %{source: source, to: to, cursor: nil, sent: 0}, {:continue, :page}}
47 + end
48 +
49 + @impl true
50 + def handle_continue(:page, state) do
51 + case Stations.page(state.source, state.cursor, @batch) do
52 + {[], _cursor} ->
53 + send(state.to, {:stations_done, state.source.id, state.sent})
54 + {:stop, :normal, state}
55 +
56 + {stations, cursor} ->
57 + send(state.to, {:stations_batch, state.source.id, stations})
58 +
59 + # Straight on to the next page: `handle_continue` yields between them,
60 + # so this stays interruptible and the linked LiveView's exit still
61 + # takes it down mid-read.
62 + {:noreply, %{state | cursor: cursor, sent: state.sent + length(stations)},
63 + {:continue, :page}}
64 + end
65 + end
66 +end
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/stations.ex
+28 −0
@@ -41,6 +41,34 @@ defmodule RoomSanctumWeb.SourceLive.Stations do
41 41 @doc "The cap `for_source/1` applies to a single source's stops."
42 42 def map_cap, do: @map_cap
43 43
44 + @doc """
45 + One page of a source's places, and the cursor to ask for the next.
46 +
47 + `{[], :done}` means there are no more. Only GTFS pages -- it is the only one
48 + of these that runs to hundreds of thousands -- and everything else comes back
49 + whole on its first page, which is what `:done` on the second call says.
50 +
51 + See `RoomSanctumWeb.SourceLive.StationStream` for who calls this and why.
52 + """
53 + def page(source, cursor, limit)
54 +
55 + def page(_source, :done, _limit), do: {[], :done}
56 +
57 + def page(%{type: :gtfs, id: id}, cursor, limit) do
58 + case Storage.stops_after(id, cursor, limit) do
59 + [] ->
60 + {[], :done}
61 +
62 + stops ->
63 + # The cursor is the last stop_id of this page, so the next page starts
64 + # after it. Taken from the row rather than counted, because a short
65 + # page is the last page and its own last id is still where to resume.
66 + {Enum.map(stops, &stop_as_station/1), List.last(stops).stop_id}
67 + end
68 + end
69 +
70 + def page(source, _cursor, _limit), do: {for_source(source), :done}
71 +
44 72 @doc """
45 73 How many places a source actually has, capped or not.
46 74
added apps/room_sanctum/test/room_sanctum_web/station_stream_test.exs
+25 −0
@@ -0,0 +1,25 @@
1 +defmodule RoomSanctumWeb.StationStreamTest do
2 + use ExUnit.Case, async: true
3 +
4 + alias RoomSanctumWeb.SourceLive.Stations
5 +
6 + describe "paging" do
7 + test "a cursor of :done ends the read, whatever the source" do
8 + # The stream stops on an empty page, so this is what stops it. Without
9 + # it a source type with no paging would answer its whole list for ever.
10 + assert {[], :done} = Stations.page(%{type: :gtfs, id: 1}, :done, 20_000)
11 + assert {[], :done} = Stations.page(%{type: :gbfs, id: 1}, :done, 20_000)
12 + assert {[], :done} = Stations.page(%{type: :weather, id: 1}, :done, 20_000)
13 + end
14 +
15 + test "a source with no geography is one empty page rather than a loop" do
16 + assert {[], :done} = Stations.page(%{type: :weather, id: 1}, nil, 20_000)
17 + end
18 +
19 + test "the cap still exists for callers that draw many sources at once" do
20 + # The tint map draws every source sharing a colour; it has not been
21 + # taught to stream and must not fetch a country per source.
22 + assert Stations.map_cap() > 0
23 + end
24 + end
25 +end

Parents: 17836dc