fix(radar): dedupe RadarFrameWorker jobs by frame_ts

BackfillEnqueueWorker re-enqueued every :queued contact each cycle, and
RadarFrameWorker had no unique clause, so the same 5-min NEXRAD frame
got scheduled repeatedly with overlapping contact lists. Prod had 9,134
duplicate available jobs across 1,061 distinct frames — 89% waste.

Two coupled changes so contacts aren't stranded by the dedup:

  1. Worker now queries eligible contacts by the frame's 5-min bucket
     instead of the args' contact_ids list. A contact that flips to
     :queued between the enqueue and the job running gets picked up.

  2. unique: [frame_ts] on states [available, scheduled, retryable].
     :executing intentionally excluded — a contact that flips to :queued
     mid-execution can still enqueue a follow-up instead of getting
     stuck behind a running job that missed it.

lib/microwaveprop/workers/radar_frame_worker.ex

@@ -15,9 +15,15 @@ defmodule Microwaveprop.Workers.RadarFrameWorker do
   marked `:complete` by one is a no-op for the other.
   """
 
+  # unique on frame_ts: backfill sweeps that re-select the same still-:queued
+  # contacts each cycle would otherwise enqueue a fresh 1-job-per-contact
+  # frame fetch. 9k+ wasteful duplicates observed in prod before this guard.
+  # :executing deliberately excluded so a contact that flips to :queued mid-
+  # execution can enqueue a follow-up job rather than getting stuck.
   use Oban.Worker,
     queue: :radar,
-    max_attempts: 3
+    max_attempts: 3,
+    unique: [fields: [:args], keys: [:frame_ts], states: [:available, :scheduled, :retryable], period: :infinity]
 
   import Ecto.Query
 
@@ -30,15 +36,22 @@ defmodule Microwaveprop.Workers.RadarFrameWorker do
   require Logger
 
   @terminal_statuses [:complete, :unavailable]
+  @frame_window_seconds 5 * 60
 
   @impl Oban.Worker
-  def perform(%Oban.Job{args: %{"frame_ts" => frame_iso, "contact_ids" => contact_ids}}) do
+  def perform(%Oban.Job{args: %{"frame_ts" => frame_iso}}) do
     {:ok, frame_ts, _} = DateTime.from_iso8601(frame_iso)
     rounded = NexradClient.round_to_5min(frame_ts)
+    bucket_end = DateTime.add(rounded, @frame_window_seconds, :second)
 
+    # Query by the frame's 5-min bucket rather than a passed contact_ids
+    # list. The unique-on-frame_ts guard prevents new sweeps from enqueueing
+    # duplicate jobs, which means a contact that flips to :queued after the
+    # job is enqueued won't land in a separate job's contact_ids — it has to
+    # be picked up by the already-queued frame job instead.
     contacts =
       Contact
-      |> where([c], c.id in ^contact_ids)
+      |> where([c], c.qso_timestamp >= ^rounded and c.qso_timestamp < ^bucket_end)
       |> where([c], c.radar_status not in ^@terminal_statuses)
       |> Repo.all()
 

test/microwaveprop/workers/radar_frame_worker_test.exs

@@ -103,5 +103,47 @@ defmodule Microwaveprop.Workers.RadarFrameWorkerTest do
       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