From 6476bc8045a0646405b6ce1e2d2ff03b3259ce7a Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Wed, 8 Apr 2026 15:52:42 -0500 Subject: [PATCH] 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. --- .../workers/contact_weather_enqueue_worker.ex | 14 ++++- .../workers/era5_fetch_worker.ex | 13 +++- lib/mix/tasks/era5_backfill.ex | 8 ++- .../workers/era5_fetch_worker_test.exs | 59 +++++++++++++++++++ 4 files changed, 87 insertions(+), 7 deletions(-) create mode 100644 test/microwaveprop/workers/era5_fetch_worker_test.exs diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index ad168155..ac6148b2 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/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 - insert_all_chunked(all_jobs) + if bulk_jobs != [] do + 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 diff --git a/lib/microwaveprop/workers/era5_fetch_worker.ex b/lib/microwaveprop/workers/era5_fetch_worker.ex index e1153db6..eb74ab47 100644 --- a/lib/microwaveprop/workers/era5_fetch_worker.ex +++ b/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 diff --git a/lib/mix/tasks/era5_backfill.ex b/lib/mix/tasks/era5_backfill.ex index 1151f5f7..3441064d 100644 --- a/lib/mix/tasks/era5_backfill.ex +++ b/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 - |> Enum.chunk_every(400) - |> Enum.each(&Oban.insert_all/1) + # Use Oban.insert/1 so the worker's unique constraint collapses + # duplicate (lat, lon, valid_time) jobs that may already be in flight + # 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 diff --git a/test/microwaveprop/workers/era5_fetch_worker_test.exs b/test/microwaveprop/workers/era5_fetch_worker_test.exs new file mode 100644 index 00000000..9b0c3987 --- /dev/null +++ b/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