Dedupe ERA5 fetch jobs by (lat, lon, valid_time)

The Era5FetchWorker now declares an Oban unique constraint on
{lat, lon, valid_time} for any job in available/scheduled/executing/
retryable, so two backfill runs targeting the same contact grid point
can no longer spawn parallel CDS requests for the same hour.

Because Oban OSS insert_all doesn't honor unique, ERA5 jobs are now
routed through Oban.insert/1 from ContactWeatherEnqueueWorker and the
era5_backfill mix task. Other worker types still use insert_all.

lib/microwaveprop/workers/contact_weather_enqueue_worker.ex

@@ -29,12 +29,16 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
     iemre_jobs = if :iemre in types, do: build_iemre_jobs([contact]), else: []
     era5_jobs = if :era5 in types, do: build_era5_jobs([contact]), else: []
 
-    all_jobs = weather_jobs ++ hrrr_jobs ++ terrain_jobs ++ iemre_jobs ++ era5_jobs
+    bulk_jobs = weather_jobs ++ hrrr_jobs ++ terrain_jobs ++ iemre_jobs
 
-    if all_jobs != [] do
+    if bulk_jobs != [] do
-      insert_all_chunked(all_jobs)
+      insert_all_chunked(bulk_jobs)
     end
 
+    # ERA5 jobs must go through Oban.insert/1 so the worker's unique
+    # constraint is honored; Oban OSS insert_all does not check uniqueness.
+    insert_unique(era5_jobs)
+
     ids = [contact.id]
 
     if contact.pos1 do
@@ -303,6 +307,10 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
     |> Enum.each(&Oban.insert_all/1)
   end
 
+  defp insert_unique(jobs) do
+    Enum.each(jobs, &Oban.insert/1)
+  end
+
   defp mark_status!(ids, field, [_ | _]), do: Radio.set_enrichment_status!(ids, field, :queued)
   defp mark_status!(ids, field, []), do: Radio.set_enrichment_status!(ids, field, :complete)
 end

lib/microwaveprop/workers/era5_fetch_worker.ex

@@ -2,8 +2,19 @@ defmodule Microwaveprop.Workers.Era5FetchWorker do
   @moduledoc """
   Fetches ERA5 reanalysis profiles for contacts where HRRR data is unavailable.
   Primarily for pre-2014 contacts that predate HRRR availability.
+
+  Jobs are unique on `{lat, lon, valid_time}` — if another fetch for the same
+  grid point and hour is already queued, scheduled, running, or retrying, a
+  duplicate insert is collapsed onto the existing job so we never hit the
+  Copernicus CDS API twice for the same data.
   """
-  use Oban.Worker, queue: :era5, max_attempts: 5
+  use Oban.Worker,
+    queue: :era5,
+    max_attempts: 5,
+    unique: [
+      period: :infinity,
+      states: [:available, :scheduled, :executing, :retryable]
+    ]
 
   alias Microwaveprop.Repo
   alias Microwaveprop.Weather.Era5Client

lib/mix/tasks/era5_backfill.ex

@@ -64,9 +64,11 @@ defmodule Mix.Tasks.Era5Backfill do
     if jobs == [] do
       Mix.shell().info("No ERA5 jobs needed (all data already exists)")
     else
-      jobs
+      # Use Oban.insert/1 so the worker's unique constraint collapses
-      |> Enum.chunk_every(400)
+      # duplicate (lat, lon, valid_time) jobs that may already be in flight
-      |> Enum.each(&Oban.insert_all/1)
+      # from a previous backfill run. Oban OSS insert_all does not check
+      # uniqueness.
+      Enum.each(jobs, &Oban.insert/1)
 
       Mix.shell().info("Enqueued #{length(jobs)} ERA5 fetch jobs")
     end

test/microwaveprop/workers/era5_fetch_worker_test.exs

@@ -0,0 +1,59 @@
+defmodule Microwaveprop.Workers.Era5FetchWorkerTest do
+  use Microwaveprop.DataCase
+
+  alias Microwaveprop.Workers.Era5FetchWorker
+
+  describe "unique constraint" do
+    test "deduplicates identical (lat, lon, valid_time) jobs across inserts" do
+      args = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, job1} = Oban.insert(Era5FetchWorker.new(args))
+        {:ok, job2} = Oban.insert(Era5FetchWorker.new(args))
+
+        # Second insert should hit the unique constraint and return the
+        # existing job (same id) instead of creating a new one.
+        assert job1.id == job2.id
+
+        count = Repo.aggregate(Oban.Job, :count, :id)
+        assert count == 1
+      end)
+    end
+
+    test "allows distinct hours on the same date" do
+      args_00 = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
+      args_06 = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T06:00:00Z"}
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, job1} = Oban.insert(Era5FetchWorker.new(args_00))
+        {:ok, job2} = Oban.insert(Era5FetchWorker.new(args_06))
+
+        assert job1.id != job2.id
+        assert Repo.aggregate(Oban.Job, :count, :id) == 2
+      end)
+    end
+
+    test "allows distinct grid points for the same valid_time" do
+      args_a = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
+      args_b = %{"lat" => 33.00, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, job1} = Oban.insert(Era5FetchWorker.new(args_a))
+        {:ok, job2} = Oban.insert(Era5FetchWorker.new(args_b))
+
+        assert job1.id != job2.id
+        assert Repo.aggregate(Oban.Job, :count, :id) == 2
+      end)
+    end
+
+    test "deduplicates across repeated Oban.insert calls" do
+      args = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        Enum.each(1..5, fn _ -> Oban.insert(Era5FetchWorker.new(args)) end)
+
+        assert Repo.aggregate(Oban.Job, :count, :id) == 1
+      end)
+    end
+  end
+end