From ddedd241de0ee0d148d5db302c63f2c2a337fd03 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Mon, 20 Apr 2026 14:21:50 -0500
Subject: [PATCH] fix(radar): dedupe RadarFrameWorker jobs by frame_ts
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 .../workers/radar_frame_worker.ex             | 19 +++++++--
 .../workers/radar_frame_worker_test.exs       | 42 +++++++++++++++++++
 2 files changed, 58 insertions(+), 3 deletions(-)

diff --git a/lib/microwaveprop/workers/radar_frame_worker.ex b/lib/microwaveprop/workers/radar_frame_worker.ex
index def16190..c20d430c 100644
--- a/lib/microwaveprop/workers/radar_frame_worker.ex
+++ b/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()
 
diff --git a/test/microwaveprop/workers/radar_frame_worker_test.exs b/test/microwaveprop/workers/radar_frame_worker_test.exs
index e46e4888..313bf3ad 100644
--- a/test/microwaveprop/workers/radar_frame_worker_test.exs
+++ b/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