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.

Ask a vision's queries together, so one slow source is not all of them

911c973 · Gabriel Morell · 2026-09-01 22:08

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

Files changed

modified apps/room_sanctum/lib/room_sanctum/workers/vision_worker.ex
+49 −21
@@ -128,33 +128,61 @@ defmodule RoomSanctum.Worker.Vision do
128 128 end
129 129
130 130 # 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 +
131 145 defp query_all_workers(vision_q) do
132 146 vision_q
133 − |> Enum.map(fn q ->
134 − r =
135 − try do
136 − query_one(q)
137 − rescue
138 − e ->
139 − Logger.warning(
140 − "Query failed for #{inspect(q.source.type)} (#{q.source.id}): #{Exception.message(e)}"
141 − )
142 −
143 − []
144 − catch
145 − kind, reason ->
146 − Logger.warning(
147 − "Query #{kind} for #{inspect(q.source.type)} (#{q.source.id}): #{inspect(reason)}"
148 − )
149 −
150 − []
151 − end
152 −
153 − {{q.id, q.source.type}, r}
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}, []}
154 164 end)
155 165 |> Enum.into(%{})
156 166 end
157 167
168 + defp query_safely(q) do
169 + query_one(q)
170 + rescue
171 + e ->
172 + Logger.warning(
173 + "Query failed for #{inspect(q.source.type)} (#{q.source.id}): #{Exception.message(e)}"
174 + )
175 +
176 + []
177 + catch
178 + kind, reason ->
179 + Logger.warning(
180 + "Query #{kind} for #{inspect(q.source.type)} (#{q.source.id}): #{inspect(reason)}"
181 + )
182 +
183 + []
184 + end
185 +
158 186 defp query_one(q) do
159 187 case q.source.type do
160 188 :gtfs ->
added apps/room_sanctum/test/room_sanctum/vision_query_batch_test.exs
+91 −0
@@ -0,0 +1,91 @@
1 +defmodule RoomSanctum.VisionQueryBatchTest do
2 + @moduledoc """
3 + A vision asks its queries together, and one of them overrunning is its own
4 + problem rather than everyone's.
5 +
6 + handle_cast(:query_workers) kills the previous task when the next tick
7 + arrives thirty seconds later. Asked one after another, a round costs the sum
8 + of every query's latency, so one slow source did not merely arrive late -- it
9 + took the whole round with it, including answers that had already come back.
10 + In prod that read as a board with nothing on it and three cancelled
11 + statements a minute.
12 + """
13 + use ExUnit.Case, async: true
14 +
15 + # The shape of the fix rather than the worker itself: query_all_workers is
16 + # private and reaches half a dozen apps' workers, none of which run here.
17 + # What is worth pinning is that the strategy has the properties claimed for
18 + # it, because the previous one did not.
19 + @concurrency 4
20 + @timeout_ms 200
21 +
22 + defp run(jobs) do
23 + jobs
24 + |> Task.async_stream(
25 + # Mirrors query_safely/1: each query is wrapped before it reaches the
26 + # stream, so a raising source is [] rather than an exit that would take
27 + # the stream with it.
28 + fn job ->
29 + try do
30 + case job do
31 + {_name, :slow} -> Process.sleep(@timeout_ms * 5)
32 + {_name, {:ok, v}} -> v
33 + {_name, :raise} -> raise "boom"
34 + end
35 + rescue
36 + _ -> []
37 + end
38 + end,
39 + max_concurrency: @concurrency,
40 + timeout: @timeout_ms,
41 + on_timeout: :kill_task
42 + )
43 + |> Enum.zip(jobs)
44 + |> Enum.map(fn
45 + {{:ok, v}, {name, _}} -> {name, v}
46 + {{:exit, _reason}, {name, _}} -> {name, []}
47 + end)
48 + |> Enum.into(%{})
49 + end
50 +
51 + test "a slow query costs only itself" do
52 + jobs = [{:a, {:ok, [1]}}, {:slowpoke, :slow}, {:c, {:ok, [3]}}]
53 +
54 + result = run(jobs)
55 +
56 + # The whole round used to be discarded. The two that answered now survive.
57 + assert result[:a] == [1]
58 + assert result[:c] == [3]
59 + assert result[:slowpoke] == []
60 + end
61 +
62 + test "the round costs the slowest query, not the sum of them" do
63 + jobs = for n <- 1..8, do: {n, :slow}
64 +
65 + {us, _} = :timer.tc(fn -> run(jobs) end)
66 + ms = us / 1000
67 +
68 + # Eight slow queries, four at a time: two rounds of the timeout, not eight
69 + # sequential waits.
70 + assert ms < @timeout_ms * 4,
71 + "expected the batch to be bounded by concurrency, took #{ms}ms"
72 + end
73 +
74 + test "results stay matched to the query that produced them" do
75 + # async_stream is ordered by default, which is what makes the zip sound.
76 + jobs = for n <- 1..12, do: {n, {:ok, [n]}}
77 +
78 + result = run(jobs)
79 +
80 + assert Enum.all?(1..12, fn n -> result[n] == [n] end)
81 + end
82 +
83 + test "a raising query is its own failure too" do
84 + jobs = [{:a, {:ok, [1]}}, {:bad, :raise}]
85 +
86 + result = run(jobs)
87 +
88 + assert result[:a] == [1]
89 + assert result[:bad] == []
90 + end
91 +end

Parents: 6029acf