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 a query once for every vision that wants it

fe3bfd7 · Gabriel Morell · 2026-09-02 17:37

12 files +335 -8
Message 
{commit_body(@commit)}

Files changed

modified apps/room_sanctum/lib/room_sanctum/application.ex
+2 −0
@@ -19,6 +19,8 @@ defmodule RoomSanctum.Application do
19 19 # Start a worker by calling: RoomSanctum.Worker.start_link(arg)
20 20 # {RoomSanctum.Worker, arg}
21 21 {Oban, Application.fetch_env!(:room_sanctum, Oban)},
22 + # One answer per query, shared by every vision that wants it.
23 + RoomSanctum.QueryCache,
22 24 ]
23 25
24 26 # See https://hexdocs.pm/elixir/Supervisor.html
added apps/room_sanctum/lib/room_sanctum/query_cache.ex
+155 −0
@@ -0,0 +1,155 @@
1 +defmodule RoomSanctum.QueryCache do
2 + @moduledoc """
3 + One answer per query, shared by everything that wants it.
4 +
5 + A query belongs to as many visions as care about it, and every vision asks
6 + for it separately on its own thirty second clock. In prod that is eighteen
7 + distinct queries being asked across six visions -- the same GTFS lookup, the
8 + one that takes ten seconds, run several times over for answers that would
9 + have been identical.
10 +
11 + Two things stop that here, and the second matters more than the first:
12 +
13 + * An answer is reused for `ttl` rather than fetched again.
14 +
15 + * While one caller is fetching, the others do not queue up behind it and do
16 + not start their own. They are told so, and go away. That is what stops a
17 + round of visions all missing the cache at the same instant and doing the
18 + work six times anyway, which a plain time-to-live cache would have
19 + allowed -- they tick together, so they miss together.
20 +
21 + Being told "someone else is on it" is a useful answer rather than a failure:
22 + the vision worker keeps the answer it already had, which is exactly what it
23 + does for a query that was slow or broken. A caller never waits.
24 +
25 + Kept in ETS rather than a GenServer for the same reason the realtime feeds
26 + are: this is read far more often than it is written, and a process holding it
27 + would serialise every reader behind whichever fetch is in progress -- which
28 + is the problem, not the solution.
29 + """
30 +
31 + require Logger
32 +
33 + @table :query_cache
34 +
35 + # A vision ticks every thirty seconds. Holding an answer for one tick means a
36 + # query runs at most once a round however many visions want it, and the board
37 + # is at most a tick behind where it would otherwise have been -- against
38 + # lookups that take ten seconds to produce, which is the trade being made.
39 + @default_ttl_ms 30_000
40 +
41 + # A claim that outlives this was abandoned by a caller that died mid-fetch.
42 + # Without an expiry the query would never be fetched again.
43 + @claim_ttl_ms 60_000
44 +
45 + def table, do: @table
46 +
47 + @doc false
48 + def child_spec(_opts) do
49 + %{id: __MODULE__, start: {__MODULE__, :start_link, []}}
50 + end
51 +
52 + @doc false
53 + def start_link do
54 + Task.start_link(fn ->
55 + :ets.new(@table, [:named_table, :public, :set, read_concurrency: true])
56 + Process.hibernate(Process, :sleep, [:infinity])
57 + end)
58 + end
59 +
60 + @doc """
61 + A fresh answer for `key`, computing it only if nobody else is.
62 +
63 + Returns `{:ok, value}` with a cached or freshly computed answer, or `:busy`
64 + when another caller is already fetching this one -- in which case the caller
65 + should keep whatever it had rather than wait.
66 + """
67 + def fetch(key, fun, ttl_ms \\ @default_ttl_ms) do
68 + case lookup(key, ttl_ms) do
69 + {:ok, value} ->
70 + {:ok, value}
71 +
72 + :miss ->
73 + if claim(key) do
74 + try do
75 + value = fun.()
76 + put(key, value)
77 + {:ok, value}
78 + after
79 + release(key)
80 + end
81 + else
82 + :busy
83 + end
84 + end
85 + end
86 +
87 + @doc """
88 + The answer held for `key`, if it is younger than `ttl_ms`.
89 + """
90 + def lookup(key, ttl_ms \\ @default_ttl_ms) do
91 + with true <- exists?(),
92 + [{_k, value, at}] <- :ets.lookup(@table, {:answer, key}),
93 + true <- fresh?(at, ttl_ms) do
94 + {:ok, value}
95 + else
96 + _otherwise -> :miss
97 + end
98 + end
99 +
100 + @doc false
101 + def put(key, value) do
102 + if exists?(), do: :ets.insert(@table, {{:answer, key}, value, now()})
103 + :ok
104 + end
105 +
106 + @doc false
107 + def forget(key) do
108 + if exists?() do
109 + :ets.delete(@table, {:answer, key})
110 + :ets.delete(@table, {:claim, key})
111 + end
112 +
113 + :ok
114 + end
115 +
116 + # insert_new is the whole of the mutual exclusion: exactly one caller gets
117 + # true for a given key, and the rest are told to go away.
118 + defp claim(key) do
119 + cond do
120 + not exists?() ->
121 + # No table means no sharing, but the work still has to happen.
122 + true
123 +
124 + :ets.insert_new(@table, {{:claim, key}, now()}) ->
125 + true
126 +
127 + stale_claim?(key) ->
128 + # Whoever held this died without releasing it.
129 + Logger.warning("QueryCache: taking over an abandoned claim on #{inspect(key)}")
130 + :ets.insert(@table, {{:claim, key}, now()})
131 + true
132 +
133 + true ->
134 + false
135 + end
136 + end
137 +
138 + defp release(key) do
139 + if exists?(), do: :ets.delete(@table, {:claim, key})
140 + :ok
141 + end
142 +
143 + defp stale_claim?(key) do
144 + case :ets.lookup(@table, {:claim, key}) do
145 + [{_k, at}] -> not fresh?(at, @claim_ttl_ms)
146 + [] -> true
147 + end
148 + end
149 +
150 + defp fresh?(at, ttl_ms), do: now() - at < ttl_ms
151 + defp now, do: System.monotonic_time(:millisecond)
152 +
153 + @doc false
154 + def exists?, do: :ets.whereis(@table) != :undefined
155 +end
modified apps/room_sanctum/lib/room_sanctum/workers/vision_worker.ex
+36 −4
@@ -147,7 +147,7 @@ defmodule RoomSanctum.Worker.Vision do
147 147 end
148 148
149 149 defp take_next(%{pending: [q | rest]} = state) do
150 − task = Task.async(fn -> query_safely(q) end)
150 + task = Task.async(fn -> query_cached(q) end)
151 151 timer = Process.send_after(self(), {:query_overran, task.ref}, @query_timeout_ms)
152 152
153 153 %{state | pending: rest}
@@ -161,7 +161,7 @@ defmodule RoomSanctum.Worker.Vision do
161 161 handle_cast(:refresh_db_cfg, state)
162 162 end
163 163
164 − def handle_info({ref, result}, state) when is_reference(ref) do
164 + def handle_info({ref, outcome}, state) when is_reference(ref) do
165 165 case Map.fetch(state.inflight, ref) do
166 166 {:ok, %{key: key, timer: timer}} ->
167 167 Process.demonitor(ref, [:flush])
@@ -169,8 +169,7 @@ defmodule RoomSanctum.Worker.Vision do
169 169
170 170 {:noreply,
171 171 state
172 − |> put_in([:data, key], result)
173 − |> put_in([:data_at, key], DateTime.utc_now())
172 + |> record(key, outcome)
174 173 |> update_in([:inflight], &Map.delete(&1, ref))
175 174 |> fill_slots()}
176 175
@@ -179,6 +178,7 @@ defmodule RoomSanctum.Worker.Vision do
179 178 end
180 179 end
181 180
181 +
182 182 # A query that overran. Shut it down, and leave what it answered last time
183 183 # where it is -- a board carrying times from two minutes ago is worth more
184 184 # than one carrying none, and blanking a panel because a source hiccuped is
@@ -228,6 +228,25 @@ defmodule RoomSanctum.Worker.Vision do
228 228 {:noreply, state}
229 229 end
230 230
231 + # Another vision was already fetching this one. Nothing arrived, so nothing
232 + # is written -- the panel keeps what it had, and picks up the shared answer
233 + # on the next round.
234 + defp record(state, _key, :skip), do: state
235 +
236 + defp record(state, key, {:ok, result}) do
237 + state
238 + |> put_in([:data, key], result)
239 + |> put_in([:data_at, key], DateTime.utc_now())
240 + end
241 +
242 + # Anything else is a task that did not return what this expects. Keeping the
243 + # previous answer is the same choice made everywhere else here, and is better
244 + # than a match error taking the worker down over one odd reply.
245 + defp record(state, key, other) do
246 + Logger.warning("Vision #{state.id}: #{inspect(key)} answered #{inspect(other)}; ignoring")
247 + state
248 + end
249 +
231 250 def handle_cast(_msg, state) do
232 251 {:noreply, state}
233 252 end
@@ -249,6 +268,19 @@ defmodule RoomSanctum.Worker.Vision do
249 268 end
250 269
251 270 # Private helper to query all workers
271 + # Through the cache, so a query wanted by several visions is run once.
272 + #
273 + # `:busy` means another vision is fetching this very query right now. There
274 + # is nothing useful to do with that except carry on with the answer we
275 + # already have -- which is the same thing this worker does for a query that
276 + # was slow or broken, and is why waiting would be the wrong move.
277 + defp query_cached(q) do
278 + case RoomSanctum.QueryCache.fetch({q.id, q.source.type}, fn -> query_safely(q) end) do
279 + {:ok, result} -> {:ok, result}
280 + :busy -> :skip
281 + end
282 + end
283 +
252 284 defp query_safely(q) do
253 285 query_one(q)
254 286 rescue
added apps/room_sanctum/test/room_sanctum/query_cache_test.exs
+138 −0
@@ -0,0 +1,138 @@
1 +defmodule RoomSanctum.QueryCacheTest do
2 + @moduledoc """
3 + One answer per query, shared by every vision that wants it.
4 +
5 + In prod, eighteen distinct queries were being asked across six visions on
6 + their own thirty second clocks -- the same ten second GTFS lookup run several
7 + times over for answers that would have been identical.
8 +
9 + A plain time-to-live cache would not have fixed that on its own: the visions
10 + tick together, so they miss together, and all six would still have done the
11 + work. Refusing the duplicate fetches is the part that matters.
12 + """
13 + use ExUnit.Case, async: false
14 +
15 + alias RoomSanctum.QueryCache
16 +
17 + setup do
18 + if :ets.whereis(QueryCache.table()) == :undefined do
19 + {:ok, _} = QueryCache.start_link()
20 + Process.sleep(20)
21 + end
22 +
23 + key = {System.unique_integer([:positive]), :gtfs}
24 + on_exit(fn -> QueryCache.forget(key) end)
25 + %{key: key}
26 + end
27 +
28 + describe "reuse" do
29 + test "the second caller gets the first one's answer without running", %{key: key} do
30 + me = self()
31 +
32 + ran = fn tag, value -> fn -> send(me, tag); value end end
33 +
34 + assert {:ok, [:answer]} = QueryCache.fetch(key, ran.(:ran, [:answer]))
35 + assert_received :ran
36 +
37 + assert {:ok, [:answer]} = QueryCache.fetch(key, ran.(:ran_again, [:nope]))
38 + refute_received :ran_again
39 + end
40 +
41 + test "an answer older than its ttl is fetched again", %{key: key} do
42 + {:ok, [:old]} = QueryCache.fetch(key, fn -> [:old] end, 20)
43 + Process.sleep(40)
44 +
45 + assert {:ok, [:new]} = QueryCache.fetch(key, fn -> [:new] end, 20)
46 + end
47 +
48 + test "one query's answer is not another's", %{key: key} do
49 + other = {System.unique_integer([:positive]), :gtfs}
50 + on_exit(fn -> QueryCache.forget(other) end)
51 +
52 + {:ok, [:a]} = QueryCache.fetch(key, fn -> [:a] end)
53 +
54 + assert {:ok, [:b]} = QueryCache.fetch(other, fn -> [:b] end)
55 + end
56 +
57 + test "lookup on its own does not compute anything", %{key: key} do
58 + assert QueryCache.lookup(key) == :miss
59 +
60 + QueryCache.put(key, [:stored])
61 +
62 + assert {:ok, [:stored]} = QueryCache.lookup(key)
63 + end
64 + end
65 +
66 + describe "while one caller is fetching" do
67 + test "the others are told, rather than queuing or duplicating", %{key: key} do
68 + me = self()
69 +
70 + slow =
71 + Task.async(fn ->
72 + QueryCache.fetch(key, fn ->
73 + send(me, :started)
74 + Process.sleep(300)
75 + [:answer]
76 + end)
77 + end)
78 +
79 + assert_receive :started, 500
80 +
81 + # Six visions ticking together must not become six fetches.
82 + assert QueryCache.fetch(key, fn -> flunk("should not have run") end) == :busy
83 + assert QueryCache.fetch(key, fn -> flunk("should not have run") end) == :busy
84 +
85 + assert {:ok, [:answer]} = Task.await(slow)
86 + end
87 +
88 + test "the claim is released once the answer lands", %{key: key} do
89 + {:ok, _} = QueryCache.fetch(key, fn -> [:answer] end)
90 + QueryCache.put(key, [:answer])
91 +
92 + # Not busy any more: expiring the answer lets the next caller fetch.
93 + assert {:ok, [:answer]} = QueryCache.fetch(key, fn -> [:other] end)
94 + end
95 +
96 + test "a caller that dies mid-fetch does not lock the query out for ever", %{key: key} do
97 + me = self()
98 +
99 + dead =
100 + spawn(fn ->
101 + QueryCache.fetch(key, fn ->
102 + send(me, :claimed)
103 + Process.sleep(:infinity)
104 + end)
105 + end)
106 +
107 + assert_receive :claimed, 500
108 + Process.exit(dead, :kill)
109 +
110 + # The claim outlives the process, which is why it carries an expiry.
111 + assert QueryCache.fetch(key, fn -> [:x] end) == :busy
112 + assert :ets.insert(QueryCache.table(), {{:claim, key}, System.monotonic_time(:millisecond) - 120_000})
113 + assert {:ok, [:recovered]} = QueryCache.fetch(key, fn -> [:recovered] end)
114 + end
115 +
116 + test "a raising fetch releases its claim" do
117 + key = {System.unique_integer([:positive]), :gtfs}
118 + on_exit(fn -> QueryCache.forget(key) end)
119 +
120 + assert_raise RuntimeError, fn -> QueryCache.fetch(key, fn -> raise "boom" end) end
121 +
122 + # The `after` is what makes the next caller able to try at all.
123 + assert {:ok, [:second]} = QueryCache.fetch(key, fn -> [:second] end)
124 + end
125 + end
126 +
127 + describe "without a table" do
128 + setup do
129 + if :ets.whereis(QueryCache.table()) != :undefined, do: :ets.delete(QueryCache.table())
130 + :ok
131 + end
132 +
133 + test "the work still happens, it is simply not shared", %{key: key} do
134 + assert {:ok, [:answer]} = QueryCache.fetch(key, fn -> [:answer] end)
135 + assert QueryCache.lookup(key) == :miss
136 + end
137 + end
138 +end
modified apps/room_sanctum/test/room_sanctum/vision_merge_test.exs
+4 −4
@@ -38,7 +38,7 @@ defmodule RoomSanctum.VisionMergeTest do
38 38 {ref, entry} = outstanding(@key)
39 39 before = DateTime.utc_now()
40 40
41 − {:noreply, s} = Vision.handle_info({ref, [:arrival]}, state(%{inflight: %{ref => entry}}))
41 + {:noreply, s} = Vision.handle_info({ref, {:ok, [:arrival]}}, state(%{inflight: %{ref => entry}}))
42 42
43 43 assert s.data[@key] == [:arrival]
44 44 assert DateTime.compare(s.data_at[@key], before) != :lt
@@ -53,7 +53,7 @@ defmodule RoomSanctum.VisionMergeTest do
53 53
54 54 {:noreply, s} =
55 55 Vision.handle_info(
56 − {ref, [:arrival]},
56 + {ref, {:ok, [:arrival]}},
57 57 state(%{data: existing, data_at: at, inflight: %{ref => entry}})
58 58 )
59 59
@@ -64,7 +64,7 @@ defmodule RoomSanctum.VisionMergeTest do
64 64 end
65 65
66 66 test "an answer nobody is waiting for is ignored" do
67 − {:noreply, s} = Vision.handle_info({make_ref(), [:stray]}, state())
67 + {:noreply, s} = Vision.handle_info({make_ref(), {:ok, [:stray]}}, state())
68 68
69 69 assert s.data == %{}
70 70 end
@@ -141,7 +141,7 @@ defmodule RoomSanctum.VisionMergeTest do
141 141 {:noreply, s} = Vision.handle_cast(:query_workers, state(%{vision_q: queries}))
142 142
143 143 {ref, entry} = s.inflight |> Enum.at(0)
144 − {:noreply, s} = Vision.handle_info({ref, [:done]}, s)
144 + {:noreply, s} = Vision.handle_info({ref, {:ok, [:done]}}, s)
145 145
146 146 # Still two out, one fewer waiting: the slot was refilled.
147 147 assert map_size(s.inflight) == 2

Parents: 51f8710