feat(hrdps): grid_tasks source column, Canadian mask, HrdpsGridWorker

Stage 3 (Elixir foundation) of the HRDPS plan. Lands the scaffolding
that lets the Rust prop-grid-rs HRDPS branch be wired up cleanly when
it ships, without further DB or Elixir-side changes.

What ships:

- `grid_tasks.source` column (default 'hrrr', backfill is a no-op).
  Replaces the (run_time, forecast_hour, kind) unique index with a
  4-column key (..., source) so HRRR and HRDPS analysis/forecast rows
  for the same cycle coexist.
- `Grid.hrdps_only_points/0` — Canadian cells inside HRDPS bbox
  (49-60°N, -141 to -52°W) but outside HRRR's CONUS bbox. Disjoint
  from `conus_points/0` by construction so the two grids never
  double-write the same (lat, lon).
- `Grid.point_source/1` — `:hrrr | :hrdps | :neither` classification
  for a (lat, lon) pair. Routes per-QSO enrichment to the right NWP
  source.
- `GridTaskEnqueuer.seed_with_analysis/2` and `seed/2` accept a
  `:source` option; defaults to "hrrr" so existing callers are
  unchanged.
- `HrdpsGridWorker` cycle seeder mirroring PropagationGridWorker's
  shape but for HRDPS's 4×/day cadence. `:hrdps` queue added to base
  Oban config.

What's deferred:

- The Rust `prop-grid-rs` worker doesn't yet branch on
  `task.source`. Until that ships, `HrdpsGridWorker` stays out of
  runtime.exs cron — it would just produce rows that the Rust worker
  mishandles as HRRR. Module doc explicitly notes the dormant state.
- Map overlay extension and FreshnessMonitor HRDPS staleness deferred
  to follow-up sessions once real data flows.

Activation path documented in the updated plan file.

Coverage cap recorded as Stage 8 decision: 60°N for v1 (SRTM stops
there). Arctic CDEM deferred to a follow-up plan.

config/config.exs

@@ -85,7 +85,13 @@ config :microwaveprop, Oban,
     ionosphere: 1,
     space_weather: 1,
     mechanism: 4,
-    contact_import: 4
+    contact_import: 4,
+    # HRDPS Canadian propagation chain seeder. Single slot — the worker
+    # only inserts grid_tasks rows, so concurrent execution buys nothing
+    # and uniqueness is enforced by the (run_time, kind, source) index.
+    # Dormant until the Rust prop-grid-rs HRDPS branch ships and the
+    # cron entry in runtime.exs is uncommented.
+    hrdps: 1
   ],
   plugins: [
     {Oban.Plugins.Pruner, max_age: 3600 * 24},

lib/microwaveprop/propagation/grid.ex

@@ -1,5 +1,19 @@
 defmodule Microwaveprop.Propagation.Grid do
-  @moduledoc "CONUS grid definition for propagation scoring. 0.125 degree resolution (~14 km)."
+  @moduledoc """
+  Propagation scoring grid definitions at 0.125° (~14 km).
+
+  Two regions, disjoint by construction:
+
+  * `conus_points/0` — HRRR-sourced CONUS cells (lat 25-50, lon -125 to -66).
+  * `hrdps_only_points/0` — Canadian cells covered by HRDPS but **not** by
+    HRRR. Bounded north at 60°N because SRTM elevation data stops at 60°N
+    (Stage 8 of the HRDPS plan defers Arctic coverage). The mask is a
+    rectangular bbox minus the HRRR overlap; over-water cells (Hudson Bay,
+    Great Lakes spillover) get scored and ignored downstream.
+
+  `point_source/1` returns `:hrrr | :hrdps | :neither` so callers can
+  route fetch/score work to the correct NWP source.
+  """
 
   @lat_min 25.0
   @lat_max 50.0
@@ -7,6 +21,11 @@ defmodule Microwaveprop.Propagation.Grid do
   @lon_max -66.0
   @step 0.125
 
+  @hrdps_lat_min 49.0
+  @hrdps_lat_max 60.0
+  @hrdps_lon_min -141.0
+  @hrdps_lon_max -52.0
+
   @doc "Returns all grid points as `{lat, lon}` tuples covering CONUS at 0.125 degree spacing."
   @spec conus_points() :: [{float(), float()}]
   def conus_points do
@@ -66,6 +85,52 @@ defmodule Microwaveprop.Propagation.Grid do
     }
   end
 
+  @doc """
+  Canadian-only grid points. Cells inside the HRDPS bbox (49-60°N,
+  -141 to -52°W) but outside HRRR's CONUS bbox. Disjoint from
+  `conus_points/0` by construction so the two grids never double-write
+  the same `(lat, lon)`.
+  """
+  @spec hrdps_only_points() :: [{float(), float()}]
+  def hrdps_only_points do
+    for lat <- float_range(@hrdps_lat_min, @hrdps_lat_max, @step),
+        lon <- float_range(@hrdps_lon_min, @hrdps_lon_max, @step),
+        not in_conus_bbox?(lat, lon) do
+      {Float.round(lat, 3), Float.round(lon, 3)}
+    end
+  end
+
+  @doc """
+  Returns the source NWP model for a `{lat, lon}` point:
+
+    * `:hrrr` — inside the CONUS bbox; route to HrrrClient.
+    * `:hrdps` — inside the Canadian extent (excluding HRRR overlap);
+      route to HrdpsClient.
+    * `:neither` — outside both. Callers should fall back to the
+      IEMRE/RAOB enrichment paths.
+  """
+  @spec point_source({float(), float()} | %{lat: number(), lon: number()}) ::
+          :hrrr | :hrdps | :neither
+  def point_source({lat, lon}), do: classify(lat, lon)
+  def point_source(%{lat: lat, lon: lon}), do: classify(lat, lon)
+
+  defp classify(lat, lon) do
+    cond do
+      in_conus_bbox?(lat, lon) -> :hrrr
+      in_hrdps_bbox?(lat, lon) -> :hrdps
+      true -> :neither
+    end
+  end
+
+  defp in_conus_bbox?(lat, lon) do
+    lat >= @lat_min and lat <= @lat_max and lon >= @lon_min and lon <= @lon_max
+  end
+
+  defp in_hrdps_bbox?(lat, lon) do
+    lat >= @hrdps_lat_min and lat <= @hrdps_lat_max and
+      lon >= @hrdps_lon_min and lon <= @hrdps_lon_max
+  end
+
   defp float_range(start, stop, step) do
     count = round((stop - start) / step) + 1
     Enum.map(0..(count - 1), fn i -> start + i * step end)

lib/microwaveprop/propagation/grid_task_enqueuer.ex

@@ -36,9 +36,17 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
   Seed one kind='analysis' row (f00) plus 18 kind='forecast' rows
   (f01..f18) for `run_time`. This is the Phase 3 Stream A cutover
   shape: Rust owns the entire chain end-to-end.
+
+  ## Options
+
+    * `:source` — `"hrrr"` (default) or `"hrdps"`. The Rust worker branches
+      on this column to pick the right URL pattern + decoder. HRDPS rows
+      are queued only once the Rust HRDPS branch ships (deferred plan
+      stage 4); until then no caller passes `:source`.
   """
-  @spec seed_with_analysis(DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
+  @spec seed_with_analysis(DateTime.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()}
-  def seed_with_analysis(%DateTime{} = run_time) do
+  def seed_with_analysis(%DateTime{} = run_time, opts \\ []) do
+    source = Keyword.get(opts, :source, "hrrr")
     run_time = DateTime.truncate(run_time, :second)
     now = DateTime.truncate(DateTime.utc_now(), :microsecond)
 
@@ -50,6 +58,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
       status: "queued",
       attempt: 0,
       kind: "analysis",
+      source: source,
       claimed_at: nil,
       completed_at: nil,
       error: nil,
@@ -67,6 +76,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
           status: "queued",
           attempt: 0,
           kind: "forecast",
+          source: source,
           claimed_at: nil,
           completed_at: nil,
           error: nil,
@@ -75,16 +85,17 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
         }
       end
 
-    do_insert([analysis_row | forecast_rows], run_time)
+    do_insert([analysis_row | forecast_rows], run_time, source)
   end
 
   @doc """
   Legacy: seed only kind='forecast' rows (f01..f18). Retained for
   tooling that needs to re-seed the forecast lane without touching
-  the analysis row. `seed_with_analysis/1` is the production path.
+  the analysis row. `seed_with_analysis/2` is the production path.
   """
-  @spec seed(DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
+  @spec seed(DateTime.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()}
-  def seed(%DateTime{} = run_time) do
+  def seed(%DateTime{} = run_time, opts \\ []) do
+    source = Keyword.get(opts, :source, "hrrr")
     run_time = DateTime.truncate(run_time, :second)
     now = DateTime.truncate(DateTime.utc_now(), :microsecond)
 
@@ -98,6 +109,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
           status: "queued",
           attempt: 0,
           kind: "forecast",
+          source: source,
           claimed_at: nil,
           completed_at: nil,
           error: nil,
@@ -106,7 +118,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
         }
       end
 
-    do_insert(rows, run_time)
+    do_insert(rows, run_time, source)
   end
 
   @doc """
@@ -152,17 +164,17 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
     %{requeued: requeued_n, failed: failed_n}
   end
 
-  defp do_insert(rows, run_time) do
+  defp do_insert(rows, run_time, source) do
     {count, _} =
       Repo.insert_all("grid_tasks", rows,
         on_conflict: :nothing,
-        conflict_target: [:run_time, :forecast_hour, :kind]
+        conflict_target: [:run_time, :forecast_hour, :kind, :source]
       )
 
     if count > 0 do
-      Logger.info("GridTaskEnqueuer: seeded #{count} grid_tasks for run_time=#{run_time}")
+      Logger.info("GridTaskEnqueuer: seeded #{count} grid_tasks (source=#{source}) for run_time=#{run_time}")
     else
-      Logger.info("GridTaskEnqueuer: run_time=#{run_time} already seeded")
+      Logger.info("GridTaskEnqueuer: run_time=#{run_time} source=#{source} already seeded")
     end
 
     {:ok, count}

lib/microwaveprop/workers/hrdps_grid_worker.ex

@@ -0,0 +1,90 @@
+defmodule Microwaveprop.Workers.HrdpsGridWorker do
+  @moduledoc """
+  Cycle seed worker for the HRDPS Canadian propagation chain. Mirrors
+  `Microwaveprop.Workers.PropagationGridWorker` but seeds `grid_tasks`
+  rows with `source: "hrdps"` so the Rust worker routes them through
+  the HRDPS fetch path.
+
+  ## Cadence
+
+  HRDPS runs 4×/day at 00/06/12/18Z and publishes f000 ~3-4h after each
+  cycle. This worker is intended to fire at HH+5 of each cycle hour
+  (05:35Z, 11:35Z, 17:35Z, 23:35Z) — see plan stage 6 for the cron entry.
+
+  ## Activation
+
+  **Not yet wired into runtime.exs cron.** The Rust prop-grid-rs worker
+  needs an HRDPS branch (plan stage 4) before it can drain
+  `source: "hrdps"` rows correctly. Until then this module is dormant
+  scaffolding — building blocks ready for the cron wire-up once the
+  Rust counterpart ships.
+  """
+
+  use Oban.Worker,
+    queue: :hrdps,
+    priority: 0,
+    max_attempts: 5,
+    # Deduplicate seed jobs in a 6-hour window matching the cycle cadence.
+    unique: [period: 6 * 3600, states: [:available, :scheduled, :executing, :retryable]]
+
+  alias Microwaveprop.Instrument
+  alias Microwaveprop.Propagation.GridTaskEnqueuer
+  alias Microwaveprop.Weather.HrdpsClient
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    Instrument.span([:hrdps, :grid_worker, :perform], %{}, fn ->
+      seed_chain()
+    end)
+  end
+
+  defp seed_chain do
+    # HRDPS shares grid_tasks with HRRR; the same reclaim sweep covers
+    # both because reclaim_stale_running/0 doesn't filter on source.
+    _ = GridTaskEnqueuer.reclaim_stale_running()
+
+    run_time = pick_run_time(DateTime.utc_now())
+
+    Logger.info("HrdpsGrid: seeding chain run_time=#{run_time} source=hrdps")
+
+    case GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") do
+      {:ok, count} ->
+        :telemetry.execute(
+          [:microwaveprop, :hrdps, :grid_worker, :seeded],
+          %{queued_tasks: count},
+          %{run_time: run_time}
+        )
+
+        :ok
+
+      {:error, reason} ->
+        :telemetry.execute(
+          [:microwaveprop, :hrdps, :grid_worker, :exception],
+          %{},
+          %{run_time: run_time, reason: inspect(reason)}
+        )
+
+        {:error, reason}
+    end
+  end
+
+  @doc """
+  Pick the freshest published HRDPS cycle. Probes `now-4h` (just past
+  typical publish latency) first; falls back to `now-10h` (one cycle
+  earlier) if the fresher cycle isn't yet on the datamart.
+
+  Public so tests can exercise selection logic without going through Oban.
+  """
+  @spec pick_run_time(DateTime.t()) :: DateTime.t()
+  def pick_run_time(%DateTime{} = now) do
+    fresh = now |> DateTime.add(-4, :hour) |> HrdpsClient.nearest_hrdps_cycle() |> DateTime.truncate(:second)
+
+    if HrdpsClient.cycle_available?(fresh) do
+      fresh
+    else
+      now |> DateTime.add(-10, :hour) |> HrdpsClient.nearest_hrdps_cycle() |> DateTime.truncate(:second)
+    end
+  end
+end

priv/repo/migrations/20260429220736_add_source_to_grid_tasks.exs

@@ -0,0 +1,22 @@
+defmodule Microwaveprop.Repo.Migrations.AddSourceToGridTasks do
+  use Ecto.Migration
+
+  # `source` splits grid_tasks between HRRR-derived (CONUS) and HRDPS-
+  # derived (Canada) work. Rust workers branch on this column to pick the
+  # right URL pattern + decoder. Default 'hrrr' so existing rows keep
+  # their meaning; HRDPS rows arrive only once HrdpsGridWorker is added
+  # to the cron (deferred until the Rust HRDPS branch ships).
+  def change do
+    alter table(:grid_tasks) do
+      add :source, :string, null: false, default: "hrrr"
+    end
+
+    # Replace (run_time, forecast_hour, kind) uniqueness with a 4-column
+    # key so HRRR and HRDPS analysis/forecast rows for the same cycle can
+    # coexist. The index also speeds up `claim_next` filters.
+    drop_if_exists unique_index(:grid_tasks, [:run_time, :forecast_hour, :kind])
+    create unique_index(:grid_tasks, [:run_time, :forecast_hour, :kind, :source])
+
+    create index(:grid_tasks, [:source])
+  end
+end

test/microwaveprop/propagation/grid_task_enqueuer_test.exs

@@ -51,6 +51,44 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuerTest do
     assert f0_count == 0
   end
 
+  describe "source-aware seeding" do
+    test "seed/2 defaults to source='hrrr'" do
+      run_time = ~U[2026-04-29 12:00:00Z]
+      assert {:ok, 18} = GridTaskEnqueuer.seed(run_time)
+
+      sources =
+        Repo.all(from(g in "grid_tasks", where: g.run_time == ^run_time, select: g.source))
+
+      assert Enum.uniq(sources) == ["hrrr"]
+    end
+
+    test "seed_with_analysis/2 with source='hrdps' tags every row" do
+      run_time = ~U[2026-04-29 12:00:00Z]
+      assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps")
+
+      sources =
+        Repo.all(from(g in "grid_tasks", where: g.run_time == ^run_time, select: g.source))
+
+      assert length(sources) == 19
+      assert Enum.uniq(sources) == ["hrdps"]
+    end
+
+    test "HRRR and HRDPS chains for the same run_time coexist (4-column unique key)" do
+      run_time = ~U[2026-04-29 18:00:00Z]
+      assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrrr")
+      assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps")
+
+      total = Repo.one(from(g in "grid_tasks", where: g.run_time == ^run_time, select: count()))
+      assert total == 38
+    end
+
+    test "reseeding the same run_time + source is idempotent" do
+      run_time = ~U[2026-04-29 06:00:00Z]
+      assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps")
+      assert {:ok, 0} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps")
+    end
+  end
+
   describe "reclaim_stale_running/1" do
     test "flips stale-running rows back to queued and clears claimed_at" do
       now = DateTime.truncate(DateTime.utc_now(), :second)

test/microwaveprop/propagation/grid_test.exs

@@ -114,4 +114,61 @@ defmodule Microwaveprop.Propagation.GridTest do
       assert length(points) == spec.lon_count * spec.lat_count
     end
   end
+
+  describe "hrdps_only_points/0" do
+    test "covers cells north of HRRR's lat_max" do
+      points = Grid.hrdps_only_points()
+      # At least one cell strictly north of HRRR's lat_max=50
+      assert Enum.any?(points, fn {lat, _lon} -> lat > 50.0 end)
+    end
+
+    test "is disjoint from conus_points/0" do
+      conus = MapSet.new(Grid.conus_points())
+      hrdps = MapSet.new(Grid.hrdps_only_points())
+      assert MapSet.disjoint?(conus, hrdps), "hrdps_only_points must not overlap conus_points"
+    end
+
+    test "all points within HRDPS bbox" do
+      Enum.each(Grid.hrdps_only_points(), fn {lat, lon} ->
+        assert lat >= 49.0 and lat <= 60.0
+        assert lon >= -141.0 and lon <= -52.0
+      end)
+    end
+
+    test "points are on 0.125 degree grid" do
+      Enum.each(Grid.hrdps_only_points(), fn {lat, lon} ->
+        assert Float.round(lat * 8, 0) == lat * 8
+        assert Float.round(lon * 8, 0) == lon * 8
+      end)
+    end
+
+    test "no point is inside the CONUS bounding box" do
+      Enum.each(Grid.hrdps_only_points(), fn {lat, lon} ->
+        in_conus = lat >= 25.0 and lat <= 50.0 and lon >= -125.0 and lon <= -66.0
+        refute in_conus, "point #{lat},#{lon} is inside CONUS bbox"
+      end)
+    end
+  end
+
+  describe "point_source/1" do
+    test "classifies cells inside CONUS as :hrrr" do
+      assert Grid.point_source({32.9, -97.0}) == :hrrr
+      assert Grid.point_source({49.0, -98.0}) == :hrrr
+    end
+
+    test "classifies Canadian cells outside HRRR as :hrdps" do
+      assert Grid.point_source({53.32, -60.42}) == :hrdps
+      assert Grid.point_source({50.125, -97.0}) == :hrdps
+    end
+
+    test "classifies points outside both bboxes as :neither" do
+      assert Grid.point_source({-30.0, 150.0}) == :neither
+      assert Grid.point_source({65.0, -90.0}) == :neither
+    end
+
+    test "accepts %{lat: _, lon: _} maps as well as tuples" do
+      assert Grid.point_source(%{lat: 32.9, lon: -97.0}) == :hrrr
+      assert Grid.point_source(%{lat: 53.32, lon: -60.42}) == :hrdps
+    end
+  end
 end

test/microwaveprop/workers/hrdps_grid_worker_test.exs

@@ -0,0 +1,66 @@
+defmodule Microwaveprop.Workers.HrdpsGridWorkerTest do
+  use Microwaveprop.DataCase, async: false
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Workers.HrdpsGridWorker
+
+  setup do
+    original = Application.get_env(:microwaveprop, :hrdps_cycle_available_fn)
+
+    on_exit(fn ->
+      if original,
+        do: Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, original),
+        else: Application.delete_env(:microwaveprop, :hrdps_cycle_available_fn)
+    end)
+  end
+
+  describe "pick_run_time/1" do
+    test "returns the now-4h cycle when it's published" do
+      Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> true end)
+
+      now = ~U[2026-04-29 16:30:00Z]
+      run_time = HrdpsGridWorker.pick_run_time(now)
+
+      # now - 4h = 12:30Z; snap down to cycle = 12:00Z
+      assert run_time == ~U[2026-04-29 12:00:00Z]
+    end
+
+    test "falls back to the previous cycle when fresh isn't yet published" do
+      Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> false end)
+
+      now = ~U[2026-04-29 16:30:00Z]
+      run_time = HrdpsGridWorker.pick_run_time(now)
+
+      # now - 10h = 06:30Z; snap down to cycle = 06:00Z
+      assert run_time == ~U[2026-04-29 06:00:00Z]
+    end
+  end
+
+  describe "perform/1" do
+    test "seeds 19 grid_tasks rows tagged with source='hrdps'" do
+      Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> true end)
+
+      assert :ok = perform_job(HrdpsGridWorker, %{})
+
+      sources_count =
+        Repo.one(from(g in "grid_tasks", where: g.source == "hrdps", select: count()))
+
+      assert sources_count == 19
+    end
+
+    test "is idempotent — re-running the worker for the same cycle inserts nothing extra" do
+      Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> true end)
+
+      assert :ok = perform_job(HrdpsGridWorker, %{})
+      assert :ok = perform_job(HrdpsGridWorker, %{})
+
+      total =
+        Repo.one(from(g in "grid_tasks", where: g.source == "hrdps", select: count()))
+
+      assert total == 19
+    end
+  end
+end