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.

Bound how many of a vision's queries are out at once

b080da5 · Gabriel Morell · 2026-09-02 15:57

9 files +122 -21
Message 
{commit_body(@commit)}

Files changed

modified apps/room_sanctum/lib/room_sanctum/workers/vision_worker.ex
+46 −20
@@ -12,6 +12,20 @@ defmodule RoomSanctum.Worker.Vision do
12 12 # than on a round of them.
13 13 @query_timeout_ms 10_000
14 14
15 + # How many of a vision's queries may be out at once.
16 + #
17 + # Asking them all together looked like the obvious way to stop waiting on the
18 + # slowest, and it emptied the database pool: six visions of seventeen queries
19 + # each, every thirty seconds, against fifteen connections. Nothing could get
20 + # one inside the ten second cap, so every query overran and every panel fell
21 + # back to its previous answer -- a board that had stopped updating entirely,
22 + # reported as thirteen hundred overruns in five minutes.
23 + #
24 + # Two at a time per vision, with the rest queued behind them. The answers
25 + # still land one by one, which is what made this worth doing; they simply do
26 + # not all leave at once.
27 + @max_inflight 2
28 +
15 29
16 30 def start_link(opts) do
17 31 GenServer.start_link(__MODULE__, opts, name: via_tuple("vision" <> opts[:id]))
@@ -47,8 +61,10 @@ defmodule RoomSanctum.Worker.Vision do
47 61 # costs only its own panel.
48 62 data: %{},
49 63 data_at: %{},
50 − # ref => %{key:, task:, timer:} for the queries currently out.
51 − inflight: %{}
64 + # ref => %{key:, task:, timer:} for the queries currently out, and the
65 + # ones waiting for a slot.
66 + inflight: %{},
67 + pending: []
52 68 }}
53 69 end
54 70
@@ -93,28 +109,37 @@ defmodule RoomSanctum.Worker.Vision do
93 109 # backpressure the brutal kill was standing in for, and unlike the kill it
94 110 # does not destroy work in progress.
95 111 def handle_cast(:query_workers, state) do
96 − outstanding = state.inflight |> Map.values() |> MapSet.new(& &1.key)
97 −
98 − state =
99 − Enum.reduce(state.vision_q, state, fn q, acc ->
100 − key = {q.id, q.source.type}
112 + outstanding =
113 + state.inflight
114 + |> Map.values()
115 + |> Enum.map(& &1.key)
116 + |> Enum.concat(Enum.map(state.pending, &query_key/1))
117 + |> MapSet.new()
118 +
119 + # A query still out, or still queued, from the last round is not asked
120 + # again. That is the backpressure the brutal kill used to stand in for, and
121 + # unlike the kill it does not destroy work in progress.
122 + fresh = Enum.reject(state.vision_q, &MapSet.member?(outstanding, query_key(&1)))
123 +
124 + {:noreply, %{state | pending: state.pending ++ fresh} |> fill_slots()}
125 + end
101 126
102 − if MapSet.member?(outstanding, key) do
103 − Logger.debug("Vision #{state.id}: #{inspect(key)} still out, not asking again")
104 − acc
105 − else
106 − start_query(acc, q, key)
107 − end
108 − end)
127 + defp query_key(q), do: {q.id, q.source.type}
109 128
110 − {:noreply, state}
129 + defp fill_slots(state) do
130 + cond do
131 + map_size(state.inflight) >= @max_inflight -> state
132 + state.pending == [] -> state
133 + true -> state |> take_next() |> fill_slots()
134 + end
111 135 end
112 136
113 − defp start_query(state, q, key) do
137 + defp take_next(%{pending: [q | rest]} = state) do
114 138 task = Task.async(fn -> query_safely(q) end)
115 139 timer = Process.send_after(self(), {:query_overran, task.ref}, @query_timeout_ms)
116 140
117 − put_in(state.inflight[task.ref], %{key: key, task: task, timer: timer})
141 + %{state | pending: rest}
142 + |> put_in([:inflight, task.ref], %{key: query_key(q), task: task, timer: timer})
118 143 end
119 144
120 145 # Handle async task completion
@@ -134,7 +159,8 @@ defmodule RoomSanctum.Worker.Vision do
134 159 state
135 160 |> put_in([:data, key], result)
136 161 |> put_in([:data_at, key], DateTime.utc_now())
137 − |> update_in([:inflight], &Map.delete(&1, ref))}
162 + |> update_in([:inflight], &Map.delete(&1, ref))
163 + |> fill_slots()}
138 164
139 165 :error ->
140 166 {:noreply, state}
@@ -154,7 +180,7 @@ defmodule RoomSanctum.Worker.Vision do
154 180 "Vision #{state.id}: #{inspect(key)} overran #{@query_timeout_ms}ms; keeping its last answer"
155 181 )
156 182
157 − {:noreply, update_in(state, [:inflight], &Map.delete(&1, ref))}
183 + {:noreply, state |> update_in([:inflight], &Map.delete(&1, ref)) |> fill_slots()}
158 184
159 185 :error ->
160 186 {:noreply, state}
@@ -172,7 +198,7 @@ defmodule RoomSanctum.Worker.Vision do
172 198 )
173 199 end
174 200
175 − {:noreply, update_in(state, [:inflight], &Map.delete(&1, ref))}
201 + {:noreply, state |> update_in([:inflight], &Map.delete(&1, ref)) |> fill_slots()}
176 202
177 203 :error ->
178 204 {:noreply, state}
modified apps/room_sanctum/test/room_sanctum/vision_merge_test.exs
+76 −1
@@ -18,11 +18,13 @@ defmodule RoomSanctum.VisionMergeTest do
18 18
19 19 defp state(attrs \\ %{}) do
20 20 Map.merge(
21 − %{id: "9", vision: nil, vision_q: [], data: %{}, data_at: %{}, inflight: %{}},
21 + %{id: "9", vision: nil, vision_q: [], data: %{}, data_at: %{}, inflight: %{}, pending: []},
22 22 attrs
23 23 )
24 24 end
25 25
26 + defp query(id, type), do: %{id: id, source: %{id: id, type: type}, query: %{}}
27 +
26 28 defp outstanding(key) do
27 29 ref = make_ref()
28 30 timer = Process.send_after(self(), :never, 60_000)
@@ -117,6 +119,79 @@ defmodule RoomSanctum.VisionMergeTest do
117 119 end
118 120 end
119 121
122 + describe "how many go out at once" do
123 + # Asking every query together emptied the database pool: six visions of
124 + # seventeen queries each, every thirty seconds, against fifteen
125 + # connections. Nothing could get one inside the cap, so every query
126 + # overran and every panel fell back to its previous answer -- a board that
127 + # had stopped updating at all.
128 + test "only a couple leave at once; the rest queue" do
129 + queries = for n <- 1..8, do: query(n, :gtfs)
130 +
131 + {:noreply, s} = Vision.handle_cast(:query_workers, state(%{vision_q: queries}))
132 +
133 + assert map_size(s.inflight) == 2
134 + assert length(s.pending) == 6
135 +
136 + for %{task: task} <- Map.values(s.inflight), do: Task.shutdown(task, :brutal_kill)
137 + end
138 +
139 + test "an answer coming back lets the next one go" do
140 + queries = for n <- 1..8, do: query(n, :gtfs)
141 + {:noreply, s} = Vision.handle_cast(:query_workers, state(%{vision_q: queries}))
142 +
143 + {ref, entry} = s.inflight |> Enum.at(0)
144 + {:noreply, s} = Vision.handle_info({ref, [:done]}, s)
145 +
146 + # Still two out, one fewer waiting: the slot was refilled.
147 + assert map_size(s.inflight) == 2
148 + assert length(s.pending) == 5
149 + assert s.data[{entry.key |> elem(0), :gtfs}] == [:done]
150 +
151 + for %{task: task} <- Map.values(s.inflight), do: Task.shutdown(task, :brutal_kill)
152 + end
153 +
154 + test "a query that overran also frees its slot" do
155 + queries = for n <- 1..8, do: query(n, :gtfs)
156 + {:noreply, s} = Vision.handle_cast(:query_workers, state(%{vision_q: queries}))
157 +
158 + {ref, _entry} = s.inflight |> Enum.at(0)
159 + {:noreply, s} = Vision.handle_info({:query_overran, ref}, s)
160 +
161 + assert map_size(s.inflight) == 2
162 + assert length(s.pending) == 5
163 +
164 + for %{task: task} <- Map.values(s.inflight), do: Task.shutdown(task, :brutal_kill)
165 + end
166 +
167 + test "a query already out or queued is not asked twice" do
168 + queries = for n <- 1..8, do: query(n, :gtfs)
169 + s0 = state(%{vision_q: queries})
170 +
171 + {:noreply, s} = Vision.handle_cast(:query_workers, s0)
172 + # The next tick arrives while the first round is still draining.
173 + {:noreply, s} = Vision.handle_cast(:query_workers, %{s | vision_q: queries})
174 +
175 + assert map_size(s.inflight) == 2
176 + assert length(s.pending) == 6
177 +
178 + for %{task: task} <- Map.values(s.inflight), do: Task.shutdown(task, :brutal_kill)
179 + end
180 +
181 + test "a query added to the vision since the last round does get asked" do
182 + queries = for n <- 1..8, do: query(n, :gtfs)
183 + {:noreply, s} = Vision.handle_cast(:query_workers, state(%{vision_q: queries}))
184 +
185 + grown = queries ++ [query(99, :tidal)]
186 + {:noreply, s} = Vision.handle_cast(:query_workers, %{s | vision_q: grown})
187 +
188 + assert length(s.pending) == 7
189 + assert {99, :tidal} in Enum.map(s.pending, &{&1.id, &1.source.type})
190 +
191 + for %{task: task} <- Map.values(s.inflight), do: Task.shutdown(task, :brutal_kill)
192 + end
193 + end
194 +
120 195 describe "state given out" do
121 196 test "carries when each answer arrived" do
122 197 at = %{@key => ~U[2026-09-01 12:00:00Z]}

Parents: da45658