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.

Land a vision's answers one at a time, and keep the last good one

da45658 · Gabriel Morell · 2026-09-02 14:55

9 files +238 -58
Message 
{commit_body(@commit)}

Files changed

modified apps/room_sanctum/lib/room_sanctum/workers/vision_worker.ex
+104 −58
@@ -8,6 +8,11 @@ defmodule RoomSanctum.Worker.Vision do
8 8
9 9 @registry :zeus
10 10
11 + # Each query is asked on its own, so this is the cap on one of them rather
12 + # than on a round of them.
13 + @query_timeout_ms 10_000
14 +
15 +
11 16 def start_link(opts) do
12 17 GenServer.start_link(__MODULE__, opts, name: via_tuple("vision" <> opts[:id]))
13 18 end
@@ -32,7 +37,19 @@ defmodule RoomSanctum.Worker.Vision do
32 37 initial_delay: 100
33 38 )
34 39
35 − {:ok, %{id: opts[:id], vision: nil, vision_q: [], data: %{}, query_task: nil}}
40 + {:ok,
41 + %{
42 + id: opts[:id],
43 + vision: nil,
44 + vision_q: [],
45 + # Answers, and when each one arrived. Held per query rather than as one
46 + # map replaced wholesale, so a source that is slow or briefly broken
47 + # costs only its own panel.
48 + data: %{},
49 + data_at: %{},
50 + # ref => %{key:, task:, timer:} for the queries currently out.
51 + inflight: %{}
52 + }}
36 53 end
37 54
38 55 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
@@ -64,20 +81,40 @@ defmodule RoomSanctum.Worker.Vision do
64 81 {:noreply, state |> Map.put(:vision, v) |> Map.put(:vision_q, queries)}
65 82 end
66 83
84 + # One task per query, each reporting on its own.
85 + #
86 + # This used to be a single task producing the whole map, which meant the
87 + # board did not move until the slowest source had answered, and the next
88 + # tick killed the round outright if it had not -- throwing away the answers
89 + # that had already come back. A source that is slow now costs its own panel
90 + # and nothing else.
91 + #
92 + # A query still out from the last round is not asked again. That is the
93 + # backpressure the brutal kill was standing in for, and unlike the kill it
94 + # does not destroy work in progress.
67 95 def handle_cast(:query_workers, state) do
68 − # todo filter out things deemed irrelevant by various timers
69 − # Cancel previous task if still running
70 − if state.query_task && Process.alive?(state.query_task.pid) do
71 − Task.shutdown(state.query_task, :brutal_kill)
72 − end
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}
73 101
74 − # Start async task to query workers
75 − task =
76 − Task.async(fn ->
77 − query_all_workers(state.vision_q)
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
78 108 end)
79 109
80 − {:noreply, state |> Map.put(:query_task, task)}
110 + {:noreply, state}
111 + end
112 +
113 + defp start_query(state, q, key) do
114 + task = Task.async(fn -> query_safely(q) end)
115 + timer = Process.send_after(self(), {:query_overran, task.ref}, @query_timeout_ms)
116 +
117 + put_in(state.inflight[task.ref], %{key: key, task: task, timer: timer})
81 118 end
82 119
83 120 # Handle async task completion
@@ -87,22 +124,65 @@ defmodule RoomSanctum.Worker.Vision do
87 124 handle_cast(:refresh_db_cfg, state)
88 125 end
89 126
90 − def handle_info({ref, queried_data}, state) do
91 − # Task completed successfully
92 − Process.demonitor(ref, [:flush])
93 − {:noreply, state |> Map.put(:data, queried_data)}
127 + def handle_info({ref, result}, state) when is_reference(ref) do
128 + case Map.fetch(state.inflight, ref) do
129 + {:ok, %{key: key, timer: timer}} ->
130 + Process.demonitor(ref, [:flush])
131 + Process.cancel_timer(timer)
132 +
133 + {:noreply,
134 + state
135 + |> put_in([:data, key], result)
136 + |> put_in([:data_at, key], DateTime.utc_now())
137 + |> update_in([:inflight], &Map.delete(&1, ref))}
138 +
139 + :error ->
140 + {:noreply, state}
141 + end
94 142 end
95 143
96 − def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
97 − # Task crashed or was killed, keep previous data
98 − {:noreply, state}
144 + # A query that overran. Shut it down, and leave what it answered last time
145 + # where it is -- a board carrying times from two minutes ago is worth more
146 + # than one carrying none, and blanking a panel because a source hiccuped is
147 + # what this worker used to do.
148 + def handle_info({:query_overran, ref}, state) do
149 + case Map.fetch(state.inflight, ref) do
150 + {:ok, %{key: key, task: task}} ->
151 + Task.shutdown(task, :brutal_kill)
152 +
153 + Logger.warning(
154 + "Vision #{state.id}: #{inspect(key)} overran #{@query_timeout_ms}ms; keeping its last answer"
155 + )
156 +
157 + {:noreply, update_in(state, [:inflight], &Map.delete(&1, ref))}
158 +
159 + :error ->
160 + {:noreply, state}
161 + end
99 162 end
100 163
101 − def handle_info({:EXIT, pid, reason}, state) do
102 − if state.query_task && state.query_task.pid == pid and reason != :normal do
103 − Logger.warning("Vision query task crashed: #{inspect(reason)}; keeping previous data")
164 + def handle_info({:DOWN, ref, :process, _pid, reason}, state) do
165 + case Map.fetch(state.inflight, ref) do
166 + {:ok, %{key: key, timer: timer}} ->
167 + Process.cancel_timer(timer)
168 +
169 + if reason != :normal do
170 + Logger.warning(
171 + "Vision #{state.id}: #{inspect(key)} died (#{inspect(reason)}); keeping its last answer"
172 + )
173 + end
174 +
175 + {:noreply, update_in(state, [:inflight], &Map.delete(&1, ref))}
176 +
177 + :error ->
178 + {:noreply, state}
104 179 end
180 + end
105 181
182 + # Trapped exits from the query tasks. The {:DOWN, ...} clause above is what
183 + # actually accounts for one ending; this only stops a task's exit from taking
184 + # the worker with it.
185 + def handle_info({:EXIT, _pid, _reason}, state) do
106 186 {:noreply, state}
107 187 end
108 188
@@ -117,6 +197,9 @@ defmodule RoomSanctum.Worker.Vision do
117 197 def handle_call(:return_state, _from, state) do
118 198 response = %{
119 199 data: state.data,
200 + # When each answer arrived, so a reader can say how old it is rather than
201 + # presenting a stale panel as though it were current.
202 + data_at: state.data_at,
120 203 queries: state.vision_q,
121 204 name: if(state.vision, do: state.vision.name, else: "Unknown")
122 205 }
@@ -128,43 +211,6 @@ defmodule RoomSanctum.Worker.Vision do
128 211 end
129 212
130 213 # Private helper to query all workers
131 − # A vision's queries reach different workers over different networks, and
132 − # asking them one after another spends the sum of their latencies. That
133 − # matters because handle_cast(:query_workers) kills the previous task when
134 − # the next tick arrives: one slow query does not merely arrive late, it
135 − # destroys the whole round, including the answers that had already come back.
136 − # In prod that showed up as a board with nothing on it and a steady three
137 − # cancelled statements a minute -- the GTFS query was simply the one holding
138 − # a database connection when the axe fell.
139 − #
140 − # Asked together, the round costs the slowest query rather than all of them,
141 − # and a query that overruns yields nothing for itself alone.
142 − @query_concurrency 4
143 − @query_timeout_ms 10_000
144 −
145 − defp query_all_workers(vision_q) do
146 − vision_q
147 − |> Task.async_stream(&query_safely/1,
148 − max_concurrency: @query_concurrency,
149 − timeout: @query_timeout_ms,
150 − on_timeout: :kill_task
151 − )
152 − |> Enum.zip(vision_q)
153 − |> Enum.map(fn
154 − {{:ok, result}, q} ->
155 − {{q.id, q.source.type}, result}
156 −
157 − {{:exit, reason}, q} ->
158 − # Its own failure, not everyone else's.
159 − Logger.warning(
160 − "Query gave up for #{inspect(q.source.type)} (#{q.source.id}): #{inspect(reason)}"
161 − )
162 −
163 − {{q.id, q.source.type}, []}
164 − end)
165 − |> Enum.into(%{})
166 − end
167 −
168 214 defp query_safely(q) do
169 215 query_one(q)
170 216 rescue
added apps/room_sanctum/test/room_sanctum/vision_merge_test.exs
+134 −0
@@ -0,0 +1,134 @@
1 +defmodule RoomSanctum.VisionMergeTest do
2 + @moduledoc """
3 + A vision's answers land one at a time, and a source that fails keeps the one
4 + it gave last.
5 +
6 + Both of those were wrong in the same place. The worker replaced its whole
7 + data map when a single task finished, so nothing on the board moved until the
8 + slowest source had answered -- and a query that failed contributed an empty
9 + list, which overwrote a perfectly good answer from thirty seconds earlier. A
10 + source that hiccupped once blanked its panel.
11 + """
12 + use ExUnit.Case, async: true
13 +
14 + alias RoomSanctum.Worker.Vision
15 +
16 + @key {1, :gtfs}
17 + @other {2, :tidal}
18 +
19 + defp state(attrs \\ %{}) do
20 + Map.merge(
21 + %{id: "9", vision: nil, vision_q: [], data: %{}, data_at: %{}, inflight: %{}},
22 + attrs
23 + )
24 + end
25 +
26 + defp outstanding(key) do
27 + ref = make_ref()
28 + timer = Process.send_after(self(), :never, 60_000)
29 + task = Task.async(fn -> Process.sleep(60_000) end)
30 +
31 + {ref, %{key: key, task: task, timer: timer}}
32 + end
33 +
34 + describe "an answer arriving" do
35 + test "merges into data and stamps when it arrived" do
36 + {ref, entry} = outstanding(@key)
37 + before = DateTime.utc_now()
38 +
39 + {:noreply, s} = Vision.handle_info({ref, [:arrival]}, state(%{inflight: %{ref => entry}}))
40 +
41 + assert s.data[@key] == [:arrival]
42 + assert DateTime.compare(s.data_at[@key], before) != :lt
43 + assert s.inflight == %{}
44 + end
45 +
46 + test "leaves the other panels alone" do
47 + {ref, entry} = outstanding(@key)
48 +
49 + existing = %{@other => [:tide]}
50 + at = %{@other => ~U[2026-09-01 00:00:00Z]}
51 +
52 + {:noreply, s} =
53 + Vision.handle_info(
54 + {ref, [:arrival]},
55 + state(%{data: existing, data_at: at, inflight: %{ref => entry}})
56 + )
57 +
58 + # The whole map used to be replaced, so a source still out lost its data.
59 + assert s.data[@other] == [:tide]
60 + assert s.data_at[@other] == ~U[2026-09-01 00:00:00Z]
61 + assert s.data[@key] == [:arrival]
62 + end
63 +
64 + test "an answer nobody is waiting for is ignored" do
65 + {:noreply, s} = Vision.handle_info({make_ref(), [:stray]}, state())
66 +
67 + assert s.data == %{}
68 + end
69 + end
70 +
71 + describe "a query that fails" do
72 + test "overrunning keeps the answer it gave last" do
73 + {ref, entry} = outstanding(@key)
74 +
75 + s0 =
76 + state(%{
77 + data: %{@key => [:yesterday]},
78 + data_at: %{@key => ~U[2026-09-01 00:00:00Z]},
79 + inflight: %{ref => entry}
80 + })
81 +
82 + {:noreply, s} = Vision.handle_info({:query_overran, ref}, s0)
83 +
84 + # A board carrying times from two minutes ago beats one carrying none.
85 + assert s.data[@key] == [:yesterday]
86 + assert s.data_at[@key] == ~U[2026-09-01 00:00:00Z]
87 + assert s.inflight == %{}
88 + end
89 +
90 + test "dying keeps it too" do
91 + {ref, entry} = outstanding(@key)
92 +
93 + s0 = state(%{data: %{@key => [:yesterday]}, inflight: %{ref => entry}})
94 +
95 + {:noreply, s} = Vision.handle_info({:DOWN, ref, :process, self(), :killed}, s0)
96 +
97 + assert s.data[@key] == [:yesterday]
98 + assert s.inflight == %{}
99 + end
100 +
101 + test "a normal exit is not reported as a failure" do
102 + {ref, entry} = outstanding(@key)
103 +
104 + {:noreply, s} =
105 + Vision.handle_info({:DOWN, ref, :process, self(), :normal}, state(%{inflight: %{ref => entry}}))
106 +
107 + assert s.inflight == %{}
108 + end
109 +
110 + test "a DOWN for something else does not disturb the round" do
111 + {ref, entry} = outstanding(@key)
112 + s0 = state(%{inflight: %{ref => entry}})
113 +
114 + {:noreply, s} = Vision.handle_info({:DOWN, make_ref(), :process, self(), :killed}, s0)
115 +
116 + assert Map.has_key?(s.inflight, ref)
117 + end
118 + end
119 +
120 + describe "state given out" do
121 + test "carries when each answer arrived" do
122 + at = %{@key => ~U[2026-09-01 12:00:00Z]}
123 +
124 + {:reply, reply, _s} =
125 + Vision.handle_call(:return_state, self(), state(%{data: %{@key => [:a]}, data_at: at}))
126 +
127 + assert reply.data_at[@key] == ~U[2026-09-01 12:00:00Z]
128 + # The existing shape is untouched; every reader matches on part of it.
129 + assert reply.data[@key] == [:a]
130 + assert Map.has_key?(reply, :queries)
131 + assert Map.has_key?(reply, :name)
132 + end
133 + end
134 +end

Parents: 911c973