From b6b5945002e42c2780975455370ef8c3bf07df38 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 20 Apr 2026 12:12:46 -0500 Subject: [PATCH] feat(radar): batch per NEXRAD frame instead of per contact MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CommonVolumeRadarWorker was enqueuing one job per contact — every job fetched + decoded the ~5 MB n0q PNG for its own 5-min frame. At current backlog depth that's 17,588 contacts across just 1,747 distinct 5-min frames, i.e. ~10 contacts per frame paying for the same PNG decode over and over. New RadarFrameWorker takes a batch of contact_ids sharing one frame, fetches + decodes the frame ONCE, then walks the in-memory pixel buffer per contact. build_radar_jobs/1 in ContactWeatherEnqueueWorker now groups the input contacts by their rounded 5-min timestamp and emits one RadarFrameWorker job per frame instead of one CommonVolumeRadarWorker per contact. process_frame/4 is public so the same code path is used both by perform/1 (production fetch) and tests (pre-decoded pixel buffer); keeps the happy-path test hermetic without hand-crafting a PNG. CommonVolumeRadarWorker stays in the tree — still used from the single-contact submit path where batching is pointless and the aggregate_stats/5 pure helper is a dependency of the new batched worker. Expected impact on the backfill: ~10x fewer fetch + decode cycles, so the 17k queued-contact backlog drains in minutes instead of ~50. --- .../workers/contact_weather_enqueue_worker.ex | 21 +++- .../workers/radar_frame_worker.ex | 119 ++++++++++++++++++ .../workers/radar_frame_worker_test.exs | 107 ++++++++++++++++ 3 files changed, 243 insertions(+), 4 deletions(-) create mode 100644 lib/microwaveprop/workers/radar_frame_worker.ex create mode 100644 test/microwaveprop/workers/radar_frame_worker_test.exs diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index eef0ee23..914229e8 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -6,10 +6,11 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do alias Microwaveprop.Weather alias Microwaveprop.Weather.HrrrPointEnqueuer alias Microwaveprop.Weather.NarrClient - alias Microwaveprop.Workers.CommonVolumeRadarWorker + alias Microwaveprop.Weather.NexradClient alias Microwaveprop.Workers.IemreFetchWorker alias Microwaveprop.Workers.MechanismClassifyWorker alias Microwaveprop.Workers.NarrFetchWorker + alias Microwaveprop.Workers.RadarFrameWorker alias Microwaveprop.Workers.TerrainProfileWorker alias Microwaveprop.Workers.WeatherFetchWorker @@ -220,14 +221,26 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do end def build_radar_jobs(contacts) do + # Group contacts by their 5-min NEXRAD frame so the worker can + # fetch+decode the ~5 MB PNG once per frame and iterate every + # contact in that frame against the in-memory pixel buffer. + # Historical backlog averages ~10 contacts per distinct frame, + # collapsing the per-job cost by roughly that factor. contacts |> Enum.filter(fn contact -> contact.pos1 && contact.pos2 && not NarrClient.in_coverage?(contact.qso_timestamp) end) - |> Enum.map(fn contact -> - CommonVolumeRadarWorker.new(%{"contact_id" => contact.id}) + |> Enum.group_by(fn contact -> + contact.qso_timestamp + |> DateTime.from_naive!("Etc/UTC") + |> NexradClient.round_to_5min() + end) + |> Enum.map(fn {frame_ts, batch} -> + RadarFrameWorker.new(%{ + "frame_ts" => DateTime.to_iso8601(frame_ts), + "contact_ids" => Enum.map(batch, & &1.id) + }) end) - |> Enum.uniq_by(fn changeset -> changeset.changes.args end) end def build_mechanism_jobs(contacts) do diff --git a/lib/microwaveprop/workers/radar_frame_worker.ex b/lib/microwaveprop/workers/radar_frame_worker.ex new file mode 100644 index 00000000..def16190 --- /dev/null +++ b/lib/microwaveprop/workers/radar_frame_worker.ex @@ -0,0 +1,119 @@ +defmodule Microwaveprop.Workers.RadarFrameWorker do + @moduledoc """ + Batched NEXRAD common-volume radar enrichment. + + Each job processes every contact whose QSO timestamp rounds to the + same 5-minute NEXRAD frame — fetch + decode happens *once* per job + and the decoded pixel buffer is walked per contact. This collapses + the 1-job-per-contact backfill pattern (`CommonVolumeRadarWorker`) + down by roughly 10x at current backlog depth (17k contacts / 1.7k + distinct frames). + + `CommonVolumeRadarWorker` stays around for the submit-time single- + contact path where batching is pointless. Both workers end in the + same state transitions so they can coexist safely — a contact + marked `:complete` by one is a no-op for the other. + """ + + use Oban.Worker, + queue: :radar, + max_attempts: 3 + + import Ecto.Query + + alias Microwaveprop.Radio.Contact + alias Microwaveprop.Radio.ContactCommonVolumeRadar + alias Microwaveprop.Repo + alias Microwaveprop.Weather.NexradClient + alias Microwaveprop.Workers.CommonVolumeRadarWorker + + require Logger + + @terminal_statuses [:complete, :unavailable] + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"frame_ts" => frame_iso, "contact_ids" => contact_ids}}) do + {:ok, frame_ts, _} = DateTime.from_iso8601(frame_iso) + rounded = NexradClient.round_to_5min(frame_ts) + + contacts = + Contact + |> where([c], c.id in ^contact_ids) + |> where([c], c.radar_status not in ^@terminal_statuses) + |> Repo.all() + + case contacts do + [] -> + :ok + + contacts -> + run_batch(contacts, rounded) + end + end + + defp run_batch(contacts, rounded) do + case NexradClient.fetch_decoded_frame(rounded) do + {:ok, pixels, width} -> + process_frame(contacts, rounded, pixels, width) + + {:error, reason} -> + Enum.each(contacts, &mark_status(&1, :unavailable)) + + Logger.info( + "RadarFrameWorker: no frame for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}" + ) + end + + :ok + end + + @doc """ + Process a batch of contacts against an already-decoded frame. Exposed + for tests and for any caller that has the pixel buffer in hand. + """ + @spec process_frame([Contact.t()], DateTime.t(), binary(), pos_integer()) :: :ok + def process_frame(contacts, rounded, pixels, width) do + Enum.each(contacts, fn contact -> + process_contact(contact, pixels, width, rounded) + end) + + Logger.info("RadarFrameWorker: #{length(contacts)} contacts ingested for frame #{DateTime.to_iso8601(rounded)}") + + :ok + end + + defp process_contact(%Contact{pos1: p1, pos2: p2} = contact, pixels, width, rounded) when is_map(p1) and is_map(p2) do + pos1 = to_latlon(p1) + pos2 = to_latlon(p2) + + stats = CommonVolumeRadarWorker.aggregate_stats(pixels, width, pos1, pos2) + upsert_row!(contact, rounded, stats) + mark_status(contact, :complete) + end + + defp process_contact(contact, _pixels, _width, _rounded) do + # Positions missing — can't compute a common volume, but we got + # this far because the enqueue filter thought it was enrichable. + # Mark unavailable so the backfill cron stops re-queueing it. + mark_status(contact, :unavailable) + end + + defp upsert_row!(%Contact{id: contact_id}, observed_at, stats) do + %ContactCommonVolumeRadar{} + |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at})) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :contact_id, :inserted_at]}, + conflict_target: :contact_id + ) + end + + defp mark_status(%Contact{} = contact, status) do + contact + |> Ecto.Changeset.change(%{radar_status: status}) + |> Repo.update!() + end + + defp to_latlon(%{"lat" => lat, "lon" => lon}) when is_number(lat) and is_number(lon) do + {lat / 1.0, lon / 1.0} + end +end diff --git a/test/microwaveprop/workers/radar_frame_worker_test.exs b/test/microwaveprop/workers/radar_frame_worker_test.exs new file mode 100644 index 00000000..e46e4888 --- /dev/null +++ b/test/microwaveprop/workers/radar_frame_worker_test.exs @@ -0,0 +1,107 @@ +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 "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 + end +end