Graham McIntire
7b78a2574c
1. GridCache: auto-release fill lock when the claimer process crashes.
claim_fill/1 + release_fill/1 go through the GenServer so the
server can Process.monitor the caller and clean up the ETS entry
on :DOWN. Clear/0 now resets both the data table and the lock
table. Fixes a latent bug where a crashed fill leaked the lock
indefinitely, preventing every subsequent /weather mount for that
valid_time from claiming and leaving cache cold.
2. RadarFrameWorker: distinguish permanent vs transient fetch errors.
404 from the IEM n0q archive is permanent (file will never exist)
and marks contacts :unavailable as before. Any other error shape
(5xx, timeout, transport failure) now returns {:error, reason}
so Oban retries — previously those also pinned contacts at
:unavailable after a transient outage.
3. AdminTaskWorker.native_derive: replace per-row Repo.update_all
(N round-trips + N fsyncs) with one UPDATE ... FROM unnest(...)
per 2000-row batch. For the 10k-profile budget this is one
network round trip per chunk instead of 10k, and one fsync per
chunk instead of 10k. Restructured the clause to separate
derivation (pure) from persistence (I/O).
All three changes are test-covered (grid_cache_test auto-release
test, radar_frame_worker_test 5xx + transport tests, existing
admin_task_worker_test native_derive coverage exercises the new
bulk path). Also drops the scorer_diff no-op test that was
verifying the clause removed in 61da51c.
187 lines
6.7 KiB
Elixir
187 lines
6.7 KiB
Elixir
defmodule Microwaveprop.Workers.RadarFrameWorkerTest do
|
|
use Microwaveprop.DataCase, async: false
|
|
use Oban.Testing, repo: Microwaveprop.Repo
|
|
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Radio.ContactCommonVolumeRadar
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.NexradClient
|
|
alias Microwaveprop.Workers.RadarFrameWorker
|
|
|
|
setup do
|
|
Req.Test.set_req_test_from_context(%{async: false})
|
|
:ok
|
|
end
|
|
|
|
defp insert_contact(attrs \\ %{}) do
|
|
default = %{
|
|
station1: "W5XD",
|
|
station2: "K5TR",
|
|
qso_timestamp: ~U[2024-09-15 18:30:00Z],
|
|
mode: "CW",
|
|
band: Decimal.new("10000"),
|
|
grid1: "EM12",
|
|
grid2: "EM00",
|
|
pos1: %{"lat" => 32.9, "lon" => -97.0},
|
|
pos2: %{"lat" => 33.1, "lon" => -96.8},
|
|
distance_km: Decimal.new("30")
|
|
}
|
|
|
|
{:ok, contact} =
|
|
%Contact{}
|
|
|> Contact.changeset(Map.merge(default, attrs))
|
|
|> Repo.insert()
|
|
|
|
contact
|
|
end
|
|
|
|
describe "process_frame/4" do
|
|
test "writes a ContactCommonVolumeRadar row and marks :complete for every contact in the batch" do
|
|
c1 = insert_contact(%{station1: "A1"})
|
|
c2 = insert_contact(%{station1: "A2"})
|
|
|
|
# In-memory pixel buffer: 10x10 empty frame. The aggregator will
|
|
# observe zero echoes but still emit a stats row per contact.
|
|
pixels = :binary.copy(<<0>>, 100)
|
|
width = 10
|
|
frame_ts = ~U[2024-09-15 18:30:00Z]
|
|
|
|
assert :ok =
|
|
RadarFrameWorker.process_frame(
|
|
[Repo.reload!(c1), Repo.reload!(c2)],
|
|
frame_ts,
|
|
pixels,
|
|
width
|
|
)
|
|
|
|
assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id)
|
|
assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id)
|
|
|
|
assert %Contact{radar_status: :complete} = Repo.get!(Contact, c1.id)
|
|
assert %Contact{radar_status: :complete} = Repo.get!(Contact, c2.id)
|
|
end
|
|
end
|
|
|
|
describe "perform/1" do
|
|
test "marks every contact in the batch :unavailable when the frame 404s" do
|
|
c1 = insert_contact(%{station1: "B1"})
|
|
c2 = insert_contact(%{station1: "B2"})
|
|
|
|
Req.Test.stub(NexradClient, fn conn ->
|
|
Plug.Conn.send_resp(conn, 404, "not found")
|
|
end)
|
|
|
|
assert :ok =
|
|
perform_job(RadarFrameWorker, %{
|
|
"frame_ts" => "2024-09-15T18:30:00Z",
|
|
"contact_ids" => [c1.id, c2.id]
|
|
})
|
|
|
|
refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id)
|
|
refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id)
|
|
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c1.id)
|
|
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c2.id)
|
|
end
|
|
|
|
test "returns {:error, _} and leaves contacts :queued on a 5xx server error (so Oban retries)" do
|
|
c1 = insert_contact(%{station1: "S1"})
|
|
c2 = insert_contact(%{station1: "S2"})
|
|
|
|
Req.Test.stub(NexradClient, fn conn ->
|
|
Plug.Conn.send_resp(conn, 503, "service unavailable")
|
|
end)
|
|
|
|
assert {:error, _reason} =
|
|
perform_job(RadarFrameWorker, %{
|
|
"frame_ts" => "2024-09-15T18:30:00Z",
|
|
"contact_ids" => [c1.id, c2.id]
|
|
})
|
|
|
|
# Neither contact should be pinned :unavailable — a retry might succeed.
|
|
assert %Contact{radar_status: status1} = Repo.get!(Contact, c1.id)
|
|
assert %Contact{radar_status: status2} = Repo.get!(Contact, c2.id)
|
|
refute status1 == :unavailable
|
|
refute status2 == :unavailable
|
|
end
|
|
|
|
test "returns {:error, _} on a transport error without pinning contacts :unavailable" do
|
|
c1 = insert_contact(%{station1: "T1"})
|
|
|
|
Req.Test.stub(NexradClient, fn conn ->
|
|
Req.Test.transport_error(conn, :timeout)
|
|
end)
|
|
|
|
assert {:error, _reason} =
|
|
perform_job(RadarFrameWorker, %{
|
|
"frame_ts" => "2024-09-15T18:30:00Z",
|
|
"contact_ids" => [c1.id]
|
|
})
|
|
|
|
assert %Contact{radar_status: status} = Repo.get!(Contact, c1.id)
|
|
refute status == :unavailable
|
|
end
|
|
|
|
test "skips contacts already at a terminal status (idempotent replay)" do
|
|
c_done = insert_contact(%{station1: "DONE"})
|
|
{:ok, _} = c_done |> Ecto.Changeset.change(%{radar_status: :complete}) |> Repo.update()
|
|
|
|
c_new = insert_contact(%{station1: "NEW"})
|
|
|
|
Req.Test.stub(NexradClient, fn conn ->
|
|
Plug.Conn.send_resp(conn, 404, "not found")
|
|
end)
|
|
|
|
assert :ok =
|
|
perform_job(RadarFrameWorker, %{
|
|
"frame_ts" => "2024-09-15T18:30:00Z",
|
|
"contact_ids" => [c_done.id, c_new.id]
|
|
})
|
|
|
|
# Already-complete contact stays complete; new one is marked unavailable.
|
|
assert %Contact{radar_status: :complete} = Repo.get!(Contact, c_done.id)
|
|
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_new.id)
|
|
end
|
|
|
|
test "processes every eligible contact in the frame's 5-min bucket, not just contact_ids" do
|
|
# Both contacts land in the 18:30 bucket (18:30 ≤ ts < 18:35).
|
|
c_passed = insert_contact(%{station1: "PASSED", qso_timestamp: ~U[2024-09-15 18:30:00Z]})
|
|
c_omitted = insert_contact(%{station1: "OMITTED", qso_timestamp: ~U[2024-09-15 18:34:00Z]})
|
|
# Outside the 18:30 bucket — must not be touched.
|
|
c_other_bucket = insert_contact(%{station1: "OTHER", qso_timestamp: ~U[2024-09-15 18:35:00Z]})
|
|
|
|
Req.Test.stub(NexradClient, fn conn ->
|
|
Plug.Conn.send_resp(conn, 404, "not found")
|
|
end)
|
|
|
|
# Only c_passed is explicitly listed, but c_omitted is in the same
|
|
# bucket and still :pending — the worker must pick it up too.
|
|
assert :ok =
|
|
perform_job(RadarFrameWorker, %{
|
|
"frame_ts" => "2024-09-15T18:30:00Z",
|
|
"contact_ids" => [c_passed.id]
|
|
})
|
|
|
|
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_passed.id)
|
|
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_omitted.id)
|
|
# Outside-bucket contact stays :pending.
|
|
assert %Contact{radar_status: :pending} = Repo.get!(Contact, c_other_bucket.id)
|
|
end
|
|
end
|
|
|
|
describe "unique constraint" do
|
|
test "rejects a second insert with the same frame_ts while one is pending" do
|
|
frame_ts = "2024-09-15T18:30:00Z"
|
|
|
|
{:ok, first} =
|
|
Oban.insert(RadarFrameWorker.new(%{"frame_ts" => frame_ts, "contact_ids" => ["a"]}))
|
|
|
|
# Second insert with a different contact_ids list but same frame_ts
|
|
# must collapse into the existing job — otherwise backfill sweeps
|
|
# duplicate 5 MB PNG fetches per frame.
|
|
{:ok, second} =
|
|
Oban.insert(RadarFrameWorker.new(%{"frame_ts" => frame_ts, "contact_ids" => ["b"]}))
|
|
|
|
assert first.id == second.id
|
|
end
|
|
end
|
|
end
|