From 7416583c277d222dace748083372419087a94129 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 18:22:15 -0500 Subject: [PATCH] feat(hrrr): route per-QSO enrichment through hrrr_fetch_tasks (Stream C Elixir) Phase 3 Stream C Elixir-side: HrrrFetchWorker is deleted; per-QSO HRRR enrichment now writes to the new hrrr_fetch_tasks table for the Rust hrrr-point-worker to drain. Table shape: - one row per valid_time (primary key) with a JSONB array of {lat, lon} points - UPSERT on conflict: array-union of points, status flips back to queued if previously done/failed so a backfill re-scan naturally refills the queue for Rust Elixir changes: - new migration 20260419231502_create_hrrr_fetch_tasks - new Microwaveprop.Weather.HrrrPointEnqueuer with enqueue/1 and enqueue_for_contacts/1. Pre-2014 contacts (NARR's territory) are skipped here so hrrr_status can pin them to :unavailable. - ContactWeatherEnqueueWorker: build_hrrr_jobs/1 removed; single- contact path and batch perform/1 both route through HrrrPointEnqueuer.enqueue_for_contacts/1. A placeholder jobs-list is kept just to feed mark_hrrr_status!. - contact_live/show.ex retry button enqueues via the same path. - :hrrr queue removed from dev/config/runtime.exs - HrrrFetchWorker module + test deleted BackfillEnqueueWorker scans continue to flow through ContactWeatherEnqueueWorker.enqueue_for_contact (unchanged), so the 30-min reconcile refills hrrr_fetch_tasks automatically. 4 new tests cover the routing, pre-2014 skip, UPSERT-union, and status-reset-on-reschedule behaviour. Rust-side hrrr-point-worker binary + k8s deployment land in the next commits. --- config/config.exs | 1 - config/dev.exs | 1 - config/runtime.exs | 3 +- .../weather/hrrr_point_enqueuer.ex | 150 +++++++++++++++++ .../workers/contact_weather_enqueue_worker.ex | 92 ++++------ .../workers/hrrr_fetch_worker.ex | 157 ------------------ .../live/contact_live/show.ex | 23 +-- ...20260419231502_create_hrrr_fetch_tasks.exs | 43 +++++ .../contact_weather_enqueue_worker_test.exs | 145 +++++----------- .../workers/hrrr_fetch_worker_test.exs | 152 ----------------- 10 files changed, 278 insertions(+), 489 deletions(-) create mode 100644 lib/microwaveprop/weather/hrrr_point_enqueuer.ex delete mode 100644 lib/microwaveprop/workers/hrrr_fetch_worker.ex create mode 100644 priv/repo/migrations/20260419231502_create_hrrr_fetch_tasks.exs delete mode 100644 test/microwaveprop/workers/hrrr_fetch_worker_test.exs diff --git a/config/config.exs b/config/config.exs index 97f6f21a..0e13c013 100644 --- a/config/config.exs +++ b/config/config.exs @@ -73,7 +73,6 @@ config :microwaveprop, Oban, weather: 20, enqueue: 1, gefs: 2, - hrrr: 5, terrain: 4, commercial: 2, iemre: 10, diff --git a/config/dev.exs b/config/dev.exs index 682a8355..74b0afa7 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -79,7 +79,6 @@ config :microwaveprop, Oban, weather: 20, enqueue: 1, gefs: 2, - hrrr: 5, terrain: 4, commercial: 2, iemre: 10, diff --git a/config/runtime.exs b/config/runtime.exs index 196dee74..b9c7988e 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -143,7 +143,8 @@ if config_env() == :prod do # forward progress. weather: 1, gefs: 1, - hrrr: 2, + # :hrrr queue removed — per-QSO HRRR fetches flow through + # hrrr_fetch_tasks and the Rust hrrr-point-worker (Phase 3 Stream C). terrain: 3, iemre: 3, # Historical backfill for pre-2014 contacts (pre-HRRR archive). diff --git a/lib/microwaveprop/weather/hrrr_point_enqueuer.ex b/lib/microwaveprop/weather/hrrr_point_enqueuer.ex new file mode 100644 index 00000000..b3362e07 --- /dev/null +++ b/lib/microwaveprop/weather/hrrr_point_enqueuer.ex @@ -0,0 +1,150 @@ +defmodule Microwaveprop.Weather.HrrrPointEnqueuer do + @moduledoc """ + Inserts rows into `hrrr_fetch_tasks` for the Rust hrrr-point-worker + to drain. + + Called from `Microwaveprop.Workers.ContactWeatherEnqueueWorker` in + place of the legacy `HrrrFetchWorker.new/1` fan-out. One row per + `valid_time` with a JSONB array of `{lat, lon}` points that share + that hour — additional points arriving from a later backfill tick + union into the existing row rather than creating a second one, so + the Rust worker fetches each GRIB2 once regardless of how many + contacts feed it. + + `BackfillEnqueueWorker` re-discovers contacts missing HRRR enrichment + every 30 minutes; each scan flows through `enqueue_for_contact` and + lands here, which keeps the backfill pipeline intact after the + cutover. + """ + + import Ecto.Query + + alias Microwaveprop.Repo + + require Logger + + @doc """ + Enqueue `valid_time -> [{lat, lon}, ...]` groups into + `hrrr_fetch_tasks`. On conflict the incoming points are array-unioned + into the existing row and the status is reset to `queued` so a + previously-done valid_time re-scheduled by backfill picks up the new + points without duplicating a fetch. + """ + @spec enqueue(%{DateTime.t() => [{float(), float()}]}) :: {:ok, non_neg_integer()} + def enqueue(groups) when is_map(groups) do + now = DateTime.truncate(DateTime.utc_now(), :microsecond) + + count = + Enum.reduce(groups, 0, fn {valid_time, points}, acc -> + valid_time = DateTime.truncate(valid_time, :second) + json_points = Enum.map(points, fn {lat, lon} -> %{"lat" => lat, "lon" => lon} end) + + # Union with the existing row's points (JSONB array) — use a + # subquery with jsonb_array_elements to dedupe. Postgres + # handles the set math so the Elixir side stays simple. + # + # Insert uses a wrapping schema query so Postgrex's jsonb + # extension encodes the list directly; raw Repo.query! with + # a binding would fall through to `text->jsonb` cast which + # then stores the JSON as a jsonb *string* and breaks the + # || union on round-trip. + id = Ecto.UUID.bingenerate() + + Repo.insert_all( + "hrrr_fetch_tasks", + [ + %{ + id: id, + valid_time: valid_time, + points: json_points, + status: "queued", + attempt: 0, + inserted_at: now, + updated_at: now + } + ], + on_conflict: + from(t in "hrrr_fetch_tasks", + update: [ + set: [ + points: + fragment( + "(SELECT COALESCE(jsonb_agg(DISTINCT p), '[]'::jsonb) FROM jsonb_array_elements(? || EXCLUDED.points) AS p)", + t.points + ), + status: + fragment( + "CASE WHEN ? IN ('done', 'failed') THEN 'queued' ELSE ? END", + t.status, + t.status + ), + attempt: + fragment( + "CASE WHEN ? IN ('done', 'failed') THEN 0 ELSE ? END", + t.status, + t.attempt + ), + updated_at: ^now + ] + ] + ), + conflict_target: [:valid_time] + ) + + acc + 1 + end) + + {:ok, count} + rescue + e -> + Logger.error("HrrrPointEnqueuer: enqueue failed: #{inspect(e)}") + {:ok, 0} + end + + @doc """ + Convenience: given a list of `Contact`s, extract the + `(valid_time, [lat, lon])` groups and enqueue in one call. Mirrors + the signature of the removed `ContactWeatherEnqueueWorker.build_hrrr_jobs/1` + so the caller swap is a one-liner. + """ + @spec enqueue_for_contacts([map()]) :: {:ok, non_neg_integer()} + def enqueue_for_contacts(contacts) do + alias Microwaveprop.Radio + alias Microwaveprop.Weather + alias Microwaveprop.Weather.HrrrClient + alias Microwaveprop.Weather.NarrClient + + groups = + contacts + |> Enum.flat_map(fn contact -> + cond do + is_nil(contact.pos1) -> + [] + + # HRRR archive starts mid-2014; NARR covers the pre-2014 + # window. Pre-coverage contacts are NARR's responsibility. + NarrClient.in_coverage?(contact.qso_timestamp) -> + [] + + true -> + rounded_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp) + + contact + |> Radio.contact_path_points() + |> Enum.flat_map(fn {lat, lon} -> + {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon) + + if Weather.has_hrrr_profile?(rlat, rlon, rounded_time) do + [] + else + [{rounded_time, {rlat, rlon}}] + end + end) + end + end) + |> Enum.group_by(fn {time, _pt} -> time end, fn {_time, pt} -> pt end) + |> Map.new(fn {time, pts} -> {time, Enum.uniq(pts)} end) + + enqueue(groups) + end +end diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 23c487ee..eef0ee23 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -4,10 +4,9 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do alias Microwaveprop.Radio alias Microwaveprop.Weather - alias Microwaveprop.Weather.HrrrClient + alias Microwaveprop.Weather.HrrrPointEnqueuer alias Microwaveprop.Weather.NarrClient alias Microwaveprop.Workers.CommonVolumeRadarWorker - alias Microwaveprop.Workers.HrrrFetchWorker alias Microwaveprop.Workers.IemreFetchWorker alias Microwaveprop.Workers.MechanismClassifyWorker alias Microwaveprop.Workers.NarrFetchWorker @@ -30,7 +29,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do bulk_jobs = jobs_by_type[:weather] ++ - jobs_by_type[:hrrr] ++ jobs_by_type[:terrain] ++ jobs_by_type[:iemre] ++ jobs_by_type[:radar] ++ @@ -40,6 +38,15 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do insert_all_chunked(bulk_jobs) end + # HRRR enrichment runs via the Rust hrrr-point-worker via the + # hrrr_fetch_tasks table (Phase 3 Stream C). Grouping by valid_time + # and UPSERT-union means a backfill tick that adds a new contact in + # an already-scheduled hour collapses into the existing row instead + # of creating a duplicate fetch. + if :hrrr in types do + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact]) + end + # NARR 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(jobs_by_type[:narr]) @@ -52,7 +59,10 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do defp build_jobs_by_type(contact, types) do %{ weather: if(:weather in types, do: build_weather_jobs([contact]), else: []), - hrrr: if(:hrrr in types, do: build_hrrr_jobs([contact]), else: []), + # hrrr jobs are routed to hrrr_fetch_tasks (not Oban) but the + # status-marking path still wants to know whether any HRRR work + # was created for this contact, so we mirror the list-shape. + hrrr: if(:hrrr in types, do: hrrr_placeholder_jobs([contact]), else: []), terrain: if(:terrain in types, do: build_terrain_jobs([contact]), else: []), iemre: if(:iemre in types, do: build_iemre_jobs([contact]), else: []), narr: if(:narr in types, do: build_narr_jobs([contact]), else: []), @@ -61,6 +71,22 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do } end + # Placeholder for the HRRR job-list signal. Returns a truthy singleton + # if the contact has at least one HRRR-eligible path point so + # mark_hrrr_status!/3 flips hrrr_queued on the contact row. Actual + # scheduling happens via HrrrPointEnqueuer.enqueue_for_contacts/1 + # above. Returns [] for pre-coverage contacts (NARR's territory) so + # the status stays :unavailable. + defp hrrr_placeholder_jobs(contacts) do + Enum.flat_map(contacts, fn contact -> + cond do + is_nil(contact.pos1) -> [] + NarrClient.in_coverage?(contact.qso_timestamp) -> [] + true -> [contact.id] + end + end) + end + defp mark_enrichment_statuses(contact, types, jobs_by_type) do ids = [contact.id] @@ -152,11 +178,11 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do :ok contacts -> - jobs = build_hrrr_jobs(contacts) - - if jobs != [] do - insert_all_chunked(jobs) - end + # Rust Stream C path: route into hrrr_fetch_tasks instead of + # Oban. The enqueuer groups by valid_time and UPSERTs so a + # backfill cron sweep that rediscovers the same contacts + # simply refreshes the existing hourly row. + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts(contacts) ids = Enum.map(contacts, & &1.id) Radio.mark_hrrr_queued!(ids) @@ -219,23 +245,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do |> Enum.uniq_by(fn changeset -> changeset.changes.args end) end - def build_hrrr_jobs(contacts) do - contacts - |> Enum.flat_map(&hrrr_points_for_contact/1) - |> Enum.group_by(fn {_point, hour} -> hour end, fn {point, _hour} -> point end) - |> Enum.map(fn {hour, points} -> - unique_points = - points - |> Enum.uniq() - |> Enum.map(fn {lat, lon} -> %{"lat" => lat, "lon" => lon} end) - - HrrrFetchWorker.new(%{ - "points" => unique_points, - "valid_time" => DateTime.to_iso8601(hour) - }) - end) - end - def build_iemre_jobs(contacts) do contacts |> Enum.flat_map(&iemre_job_for_contact/1) @@ -314,37 +323,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do end end - defp hrrr_points_for_contact(%{pos1: nil}), do: [] - - defp hrrr_points_for_contact(contact) do - # HRRR archive starts mid-2014; NARR covers the 1979 → 2014-10-02 window. - # Short-circuit here so build_hrrr_jobs returns nothing for pre-coverage - # contacts and mark_hrrr_status! can pin them to :unavailable. - if NarrClient.in_coverage?(contact.qso_timestamp) do - [] - else - hrrr_points_for_path(contact) - end - end - - defp hrrr_points_for_path(contact) do - rounded_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp) - - contact - |> Radio.contact_path_points() - |> Enum.flat_map(&hrrr_point_or_skip(&1, rounded_time)) - end - - defp hrrr_point_or_skip({lat, lon}, rounded_time) do - {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon) - - if Weather.has_hrrr_profile?(rlat, rlon, rounded_time) do - [] - else - [{{rlat, rlon}, rounded_time}] - end - end - defp build_asos_jobs(lat, lon, timestamp) do start_dt = DateTime.add(timestamp, -2 * 3600, :second) end_dt = DateTime.add(timestamp, 2 * 3600, :second) diff --git a/lib/microwaveprop/workers/hrrr_fetch_worker.ex b/lib/microwaveprop/workers/hrrr_fetch_worker.ex deleted file mode 100644 index 83de03d7..00000000 --- a/lib/microwaveprop/workers/hrrr_fetch_worker.ex +++ /dev/null @@ -1,157 +0,0 @@ -defmodule Microwaveprop.Workers.HrrrFetchWorker do - @moduledoc false - use Oban.Worker, - queue: :hrrr, - max_attempts: 20, - unique: [period: :infinity, states: [:available, :scheduled, :executing, :retryable]] - - alias Microwaveprop.Weather - alias Microwaveprop.Weather.HrrrClient - alias Microwaveprop.Weather.SoundingParams - - require Logger - - @impl Oban.Worker - def backoff(%Oban.Job{attempt: attempt}) do - # Exponential backoff: 2m, 4m, 8m, 16m, 32m, 1h, 2h, 4h, 6h, 6h, ... - min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600) - end - - @impl Oban.Worker - def perform(%Oban.Job{args: %{"points" => points, "valid_time" => valid_time_str}}) do - # Batch mode: download GRIB2 once, extract multiple points - {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str) - - point_tuples = - points - |> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end) - |> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end) - - result = - if point_tuples == [] do - Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}") - :ok - else - Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}") - fetch_and_store_batch(point_tuples, valid_time, valid_time_str) - end - - # Reclaim the refc binary heap held from GRIB2 ranges before the - # Oban producer hands this process its next job. Without this each - # worker steady-states at 100+ MiB of retained HRRR payload. - :erlang.garbage_collect(self()) - result - end - - def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do - # Legacy single-point mode - {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str) - {lat, lon} = Weather.round_to_hrrr_grid(raw_lat, raw_lon) - - if Weather.has_hrrr_profile?(lat, lon, valid_time) do - Logger.info("HRRR profile already exists for #{lat},#{lon} @ #{valid_time_str}") - :ok - else - Logger.info("Fetching HRRR profile for #{lat},#{lon} @ #{valid_time_str}") - - case HrrrClient.fetch_profile(lat, lon, valid_time) do - {:ok, data} -> - store_profile(lat, lon, valid_time, data) - broadcast_enrichment(lat, lon, valid_time) - :ok - - {:error, reason} -> - handle_error(reason, "#{lat},#{lon} @ #{valid_time_str}") - end - end - end - - defp fetch_and_store_batch(point_tuples, valid_time, valid_time_str) do - case HrrrClient.fetch_grid(point_tuples, valid_time) do - {:ok, grid_data} -> - profiles = - Enum.map(grid_data, fn {{lat, lon}, data} -> - build_profile_attrs(lat, lon, valid_time, data) - end) - - Weather.upsert_hrrr_profiles_batch(profiles, skip_existing: true) - - Enum.each(grid_data, fn {{lat, lon}, _data} -> - broadcast_enrichment(lat, lon, valid_time) - end) - - Logger.info("HRRR batch: saved #{map_size(grid_data)} profiles for #{valid_time_str}") - :ok - - {:error, reason} -> - handle_error(reason, "batch @ #{valid_time_str}") - end - end - - defp store_profile(lat, lon, valid_time, data) do - attrs = build_profile_attrs(lat, lon, valid_time, data) - Weather.upsert_hrrr_profile(attrs) - end - - defp build_profile_attrs(lat, lon, valid_time, data) do - params = SoundingParams.derive(data.profile) - - maybe_add_derived_params( - %{ - valid_time: valid_time, - lat: lat, - lon: lon, - run_time: data.run_time, - profile: data.profile, - hpbl_m: data.hpbl_m, - pwat_mm: data.pwat_mm, - surface_temp_c: data.surface_temp_c, - surface_dewpoint_c: data.surface_dewpoint_c, - surface_pressure_mb: data.surface_pressure_mb - }, - params - ) - end - - defp handle_error(reason, label) do - if transient_failure?(reason) do - Logger.error("HRRR transient error for #{label}: #{inspect(reason)}") - {:error, reason} - else - Logger.warning("HRRR permanent failure for #{label}: #{inspect(reason)}") - {:cancel, reason} - end - end - - # Only retry on transient network/server errors — everything else is permanent - defp transient_failure?(%{__exception__: true}), do: true - defp transient_failure?("HRRR idx HTTP " <> status), do: server_error?(status) - defp transient_failure?("HRRR grib HTTP " <> status), do: server_error?(status) - defp transient_failure?(_), do: false - - defp server_error?(status) do - case Integer.parse(status) do - {code, _} when code in [429, 500, 502, 503, 504] -> true - _ -> false - end - end - - defp maybe_add_derived_params(attrs, nil), do: attrs - - defp maybe_add_derived_params(attrs, params) do - Map.merge(attrs, %{ - surface_refractivity: params.surface_refractivity, - min_refractivity_gradient: params.min_refractivity_gradient, - ducting_detected: params.ducting_detected, - duct_characteristics: params.duct_characteristics - }) - end - - defp broadcast_enrichment(lat, lon, valid_time) do - Phoenix.PubSub.broadcast( - Microwaveprop.PubSub, - "contact_enrichment:hrrr", - {:hrrr_ready, %{lat: lat, lon: lon, valid_time: valid_time}} - ) - end -end diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index 8bd08404..3361b004 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -13,9 +13,9 @@ defmodule MicrowavepropWeb.ContactLive.Show do alias Microwaveprop.Weather alias Microwaveprop.Weather.HrrrClient alias Microwaveprop.Weather.HrrrNativeProfile + alias Microwaveprop.Weather.HrrrPointEnqueuer alias Microwaveprop.Weather.NarrClient alias Microwaveprop.Workers.ContactWeatherEnqueueWorker - alias Microwaveprop.Workers.HrrrFetchWorker alias Microwaveprop.Workers.NarrFetchWorker alias Microwaveprop.Workers.SolarIndexWorker alias Microwaveprop.Workers.TerrainProfileWorker @@ -752,22 +752,11 @@ defmodule MicrowavepropWeb.ContactLive.Show do valid_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp) {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon) - case Oban.insert( - HrrrFetchWorker.new(%{ - "lat" => rlat, - "lon" => rlon, - "valid_time" => DateTime.to_iso8601(valid_time) - }) - ) do - {:ok, %{conflict?: false}} -> - Radio.set_enrichment_status!([contact.id], :hrrr_status, :queued) - {nil, %{contact | hrrr_status: :queued}} - - _ -> - # Unique conflict — job already ran recently, mark unavailable - Radio.set_enrichment_status!([contact.id], :hrrr_status, :unavailable) - {nil, %{contact | hrrr_status: :unavailable}} - end + # Phase 3 Stream C: route retries through hrrr_fetch_tasks so the + # Rust hrrr-point-worker picks them up alongside backfill work. + {:ok, _} = HrrrPointEnqueuer.enqueue(%{valid_time => [{rlat, rlon}]}) + Radio.set_enrichment_status!([contact.id], :hrrr_status, :queued) + {nil, %{contact | hrrr_status: :queued}} else {nil, contact} end diff --git a/priv/repo/migrations/20260419231502_create_hrrr_fetch_tasks.exs b/priv/repo/migrations/20260419231502_create_hrrr_fetch_tasks.exs new file mode 100644 index 00000000..19863e33 --- /dev/null +++ b/priv/repo/migrations/20260419231502_create_hrrr_fetch_tasks.exs @@ -0,0 +1,43 @@ +defmodule Microwaveprop.Repo.Migrations.CreateHrrrFetchTasks do + use Ecto.Migration + + # Hand-off queue between the Elixir enrichment enqueuer and the Rust + # hrrr-point-worker (Phase 3 Stream C). One row per valid_time with a + # JSONB array of {lat, lon} points the workers needs to extract from + # that hour's HRRR cycle. Rust claims by SKIP LOCKED, fetches the + # GRIB2 once, extracts every point in the batch, bulk-upserts to + # hrrr_profiles, and marks the row done. + def change do + create table(:hrrr_fetch_tasks, primary_key: false) do + add :id, :binary_id, primary_key: true + add :valid_time, :utc_datetime, null: false + + # [{"lat": f, "lon": f}, …] — additional points for the same + # valid_time get merged via jsonb array-union UPSERT so a + # backfill tick doesn't schedule a new row per contact. + add :points, :map, null: false, default: fragment("'[]'::jsonb") + + add :status, :string, null: false, default: "queued" + add :attempt, :integer, null: false, default: 0 + add :claimed_at, :utc_datetime_usec + add :completed_at, :utc_datetime_usec + add :error, :text + + timestamps(type: :utc_datetime) + end + + # One task per valid_time. Merges all points for the hour into a + # single fetch so a full backfill scan costs one GRIB2 download + # per hour, not one per contact. + create unique_index(:hrrr_fetch_tasks, [:valid_time]) + + # Hot-path claim query scans the queued subset. + create index(:hrrr_fetch_tasks, [:status], + where: "status = 'queued'", + name: :hrrr_fetch_tasks_queued_idx + ) + + # Prune index for retention. + create index(:hrrr_fetch_tasks, [:completed_at]) + end +end diff --git a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs index a3887627..8503d332 100644 --- a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs @@ -255,122 +255,61 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do end end - describe "build_hrrr_jobs/1" do - test "batches all path points into one job per HRRR hour" do + describe "HRRR enrichment routing (Phase 3 Stream C → hrrr_fetch_tasks)" do + alias Microwaveprop.Weather.HrrrPointEnqueuer + + test "enqueue_for_contacts writes one hrrr_fetch_tasks row per valid_time" do + import Ecto.Query + contact = create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}}) + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact]) - jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) + rows = + Repo.all(from t in "hrrr_fetch_tasks", select: %{vt: t.valid_time, points: t.points, status: t.status}) - # All 3 path points share the same HRRR hour → 1 batch job - assert length(jobs) == 1 - job = hd(jobs) - points = job.changes.args["points"] - # pos1, midpoint, pos2 = 3 distinct grid points in the batch - assert length(points) == 3 - lats = Enum.map(points, & &1["lat"]) - assert 32.91 in lats + assert [row] = rows + # pos1 + midpoint + pos2 → 3 distinct HRRR grid cells in the batch. + assert length(row.points) == 3 + assert row.status == "queued" end - test "batch contains single point when pos2 is nil" do - contact = create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}, pos2: nil}) + test "enqueue_for_contacts skips pre-2014 contacts (NARR's territory)" do + import Ecto.Query - jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) - - assert length(jobs) == 1 - job = hd(jobs) - [point] = job.changes.args["points"] - assert point["lat"] == 32.91 - assert point["lon"] == -97.04 - end - - test "rounds valid_time to nearest hour" do - contact = create_contact(%{qso_timestamp: ~U[2026-03-28 18:45:00Z], pos2: nil}) - - jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) - - job = hd(jobs) - assert job.changes.args["valid_time"] == "2026-03-28T19:00:00Z" - end - - test "deduplicates jobs across QSOs with identical path points" do - q1 = - create_contact(%{ - station1: "A1", - qso_timestamp: ~U[2026-03-28 18:10:00Z], - pos1: %{"lat" => 32.90, "lon" => -97.04}, - pos2: nil - }) - - q2 = - create_contact(%{ - station1: "A2", - qso_timestamp: ~U[2026-03-28 18:20:00Z], - pos1: %{"lat" => 32.90, "lon" => -97.04}, - pos2: nil - }) - - jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([q1, q2]) - - # Both round to same hour and same location → 1 job - assert length(jobs) == 1 - end - - test "deduplicates path points that round to the same HRRR grid cell" do - # pos1 and pos2 only 0.001° apart — all 3 points round to same HRRR cell - contact = - create_contact(%{ - pos1: %{"lat" => 32.901, "lon" => -97.041}, - pos2: %{"lat" => 32.902, "lon" => -97.042} - }) - - jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) - - # All 3 path points round to {32.90, -97.04} → 1 job - assert length(jobs) == 1 - end - - test "skips QSOs without pos1" do - contact = create_contact(%{pos1: nil}) - - assert ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) == [] - end - - # HRRR archive starts mid-2014; pre-coverage fetches always 404. Skipping - # here stops the :queued ↔ :unavailable ping-pong in BackfillEnqueueWorker - # where the reconcile flips pre-2014 rows to :unavailable and the same - # run immediately flips them back to :queued from a build_hrrr_jobs call - # that never had a chance of succeeding. - test "skips pre-2014 contacts (outside HRRR coverage — NARR's job)" do contact = create_contact(%{qso_timestamp: ~U[2010-06-15 18:00:00Z]}) + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact]) - assert ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) == [] + assert [] = Repo.all(from t in "hrrr_fetch_tasks", select: t.id) end - test "excludes existing points from batch" do - contact = create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}}) + test "re-enqueuing the same valid_time UPSERT-unions the points" do + import Ecto.Query - # Insert a profile at the pos1 rounded grid point only - Weather.upsert_hrrr_profile(%{ - valid_time: ~U[2026-03-28 18:00:00Z], - lat: 32.91, - lon: -97.04, - run_time: ~U[2026-03-28 18:00:00Z], - profile: [%{"pres" => 1000.0, "tmpc" => 25.0}], - hpbl_m: 1500.0, - pwat_mm: 25.0, - surface_temp_c: 25.0, - surface_dewpoint_c: 18.0, - surface_pressure_mb: 1013.0 - }) + c1 = create_contact(%{station1: "A1", pos1: %{"lat" => 32.90, "lon" => -97.04}, pos2: nil}) + c2 = create_contact(%{station1: "A2", pos1: %{"lat" => 33.50, "lon" => -96.50}, pos2: nil}) - jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([c1]) + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([c2]) - # 1 batch job, but pos1 excluded from points list - assert length(jobs) == 1 - points = hd(jobs).changes.args["points"] - assert length(points) == 2 - lats = Enum.map(points, & &1["lat"]) - refute 32.91 in lats + rows = Repo.all(from t in "hrrr_fetch_tasks", select: %{vt: t.valid_time, points: t.points}) + assert [row] = rows + assert length(row.points) == 2 + end + + test "re-enqueuing a 'done' row flips status back to queued" do + import Ecto.Query + + contact = create_contact(%{pos1: %{"lat" => 32.90, "lon" => -97.04}, pos2: nil}) + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact]) + + Repo.update_all( + from(t in "hrrr_fetch_tasks"), + set: [status: "done", completed_at: DateTime.utc_now()] + ) + + {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact]) + assert [row] = Repo.all(from t in "hrrr_fetch_tasks", select: %{status: t.status}) + assert row.status == "queued" end end diff --git a/test/microwaveprop/workers/hrrr_fetch_worker_test.exs b/test/microwaveprop/workers/hrrr_fetch_worker_test.exs deleted file mode 100644 index ff633280..00000000 --- a/test/microwaveprop/workers/hrrr_fetch_worker_test.exs +++ /dev/null @@ -1,152 +0,0 @@ -defmodule Microwaveprop.Workers.HrrrFetchWorkerTest do - use Microwaveprop.DataCase, async: true - - alias Microwaveprop.Weather - alias Microwaveprop.Weather.HrrrProfile - alias Microwaveprop.Workers.HrrrFetchWorker - - @sample_profile [ - %{"pres" => 1000.0, "hght" => 110.0, "tmpc" => 25.0, "dwpc" => 18.0}, - %{"pres" => 975.0, "hght" => 350.0, "tmpc" => 23.0, "dwpc" => 16.0}, - %{"pres" => 950.0, "hght" => 590.0, "tmpc" => 21.0, "dwpc" => 14.0} - ] - - @sample_client_result %{ - surface_temp_c: 25.0, - surface_dewpoint_c: 18.0, - surface_pressure_mb: 1013.0, - hpbl_m: 1500.0, - pwat_mm: 25.0, - run_time: ~U[2026-03-28 18:00:00Z], - profile: @sample_profile - } - - describe "perform/1" do - test "skips if HRRR profile already exists" do - Weather.upsert_hrrr_profile(%{ - valid_time: ~U[2026-03-28 18:00:00Z], - lat: 32.90, - lon: -97.04, - profile: @sample_profile - }) - - job = - HrrrFetchWorker.new(%{ - "lat" => 32.90, - "lon" => -97.04, - "valid_time" => "2026-03-28T18:00:00Z" - }) - - assert :ok = HrrrFetchWorker.perform(%Oban.Job{args: job.changes.args}) - end - - test "stores profile when client returns success" do - # We test the perform logic by mocking at the module level - # The actual HTTP calls are tested in HrrrClientTest - # Here we verify the worker's coordination logic - - lat = 32.90 - lon = -97.04 - valid_time = ~U[2026-03-28 18:00:00Z] - - # Verify no profile exists initially - refute Weather.has_hrrr_profile?(lat, lon, valid_time) - - # Simulate what perform does after a successful fetch - attrs = build_profile_attrs(lat, lon, valid_time, @sample_client_result) - {:ok, profile} = Weather.upsert_hrrr_profile(attrs) - - assert profile.lat == lat - assert profile.lon == lon - assert profile.hpbl_m == 1500.0 - assert profile.surface_temp_c == 25.0 - assert Weather.has_hrrr_profile?(lat, lon, valid_time) - end - - test "rounds lat/lon before checking existence" do - # Profile at rounded coordinates - Weather.upsert_hrrr_profile(%{ - valid_time: ~U[2026-03-28 18:00:00Z], - lat: 32.90, - lon: -97.04, - profile: [] - }) - - # Worker called with slightly different coords that round to same - job = - HrrrFetchWorker.new(%{ - "lat" => 32.904, - "lon" => -97.039, - "valid_time" => "2026-03-28T18:00:00Z" - }) - - # Should skip because rounded coords match - assert :ok = HrrrFetchWorker.perform(%Oban.Job{args: job.changes.args}) - - # Only one profile should exist - count = Repo.aggregate(HrrrProfile, :count) - assert count == 1 - end - end - - describe "perform/1 batch mode" do - test "skips all points when profiles already exist" do - Weather.upsert_hrrr_profile(%{ - valid_time: ~U[2026-03-28 18:00:00Z], - lat: 32.90, - lon: -97.04, - profile: @sample_profile - }) - - job = %Oban.Job{ - args: %{ - "points" => [%{"lat" => 32.90, "lon" => -97.04}], - "valid_time" => "2026-03-28T18:00:00Z" - } - } - - assert :ok = HrrrFetchWorker.perform(job) - end - end - - describe "backoff/1" do - test "returns exponential backoff capped at 6 hours" do - assert HrrrFetchWorker.backoff(%Oban.Job{attempt: 1}) == 120 - assert HrrrFetchWorker.backoff(%Oban.Job{attempt: 2}) == 240 - assert HrrrFetchWorker.backoff(%Oban.Job{attempt: 10}) <= 21_600 - end - end - - describe "permanent_failure?/1" do - test "cancels on 404 idx" do - job = %Oban.Job{ - args: %{ - "lat" => 32.90, - "lon" => -97.04, - "valid_time" => "1995-06-15T12:00:00Z" - } - } - - Req.Test.stub(Microwaveprop.Weather.HrrrClient, fn conn -> - Plug.Conn.send_resp(conn, 404, "not found") - end) - - assert {:cancel, "HRRR idx HTTP 404"} = HrrrFetchWorker.perform(job) - end - end - - defp build_profile_attrs(lat, lon, valid_time, client_result) do - %{ - valid_time: valid_time, - lat: lat, - lon: lon, - run_time: client_result.run_time, - profile: client_result.profile, - hpbl_m: client_result.hpbl_m, - pwat_mm: client_result.pwat_mm, - surface_temp_c: client_result.surface_temp_c, - surface_dewpoint_c: client_result.surface_dewpoint_c, - surface_pressure_mb: client_result.surface_pressure_mb - } - end -end