feat(hrdps): scope to current-hour-only at 0.5° step

The HRDPS pipeline as written was unusable in production: each chain
step took ~5 hours (not 30-90s as the dev-bench claim) because every
`wgrib2 -lon` batch redundantly JPEG2000-decodes ~150 records. With 57
batches × 18 forecast hours, a single cycle would take days to drain.

The /weather Canadian view only needs the current hour, so the cheapest
fix is to stop seeding the rest of the chain entirely:

* Rust grid: HRDPS_STEP=0.5 (was 0.125) drops point count 16× to ~3.5k.
  Coarser cells than HRRR, but /weather gets visible Canadian coverage
  inside one chain step (~20 min) instead of never.
* GridTaskEnqueuer.seed_current_hour/2 inserts exactly one forecast row
  whose valid_time is closest to `now`. fh is clamped to 1..18 so the
  Rust F00Reserved branch never fires.
* HrdpsGridWorker switches to seed_current_hour and drops the
  6-hour-cycle cron in favor of HH:35 hourly fires. Reseeds are
  idempotent on the 4-column unique key.

Also fixes a pre-existing credo "nested too deep" in
ScalarFile.read_point by extracting lookup_in_chunk + hit_or_false
helpers — happened to be on the read path that surfaces this work, so
folding it into the same commit.

config/runtime.exs

@@ -248,12 +248,13 @@ if config_env() == :prod do
        # idempotent INSERT batch.
        {"5,20,35,50 * * * *", Microwaveprop.Workers.PropagationGridWorker},
        # HRDPS Canadian propagation chain. ECCC publishes HRDPS f000
-       # ~3-4h after each 00/06/12/18Z cycle, so seeding at HH:35 of
-       # each cycle hour (5h after cycle, 30 min offset from HRRR's
-       # */15 schedule) gives a comfortable margin before the f000
-       # files land. The Rust prop-grid-rs HRDPS branch (run_chain_step_hrdps)
-       # drains the resulting source='hrdps' rows.
-       {"35 5,11,17,23 * * *", Microwaveprop.Workers.HrdpsGridWorker},
+       # ~3-4h after each 00/06/12/18Z cycle. Worker seeds exactly one
+       # forecast row per fire (current hour), so we run hourly to keep
+       # the /weather Canadian cell within ~1h of clock time. Reseeds
+       # are idempotent through the (run_time, forecast_hour, kind, source)
+       # unique index — duplicate fires for an already-seeded hour are
+       # no-ops at the DB level.
+       {"35 * * * *", Microwaveprop.Workers.HrdpsGridWorker},
        # GEFS runs publish ~3-4h after the 00/06/12/18Z cycle; seed the
        # Day 2-7 outlook 5h after each run to stay safely past NOMADS'
        # publication lag. Seeder enqueues f024-f168 at 6-hour cadence

lib/microwaveprop/propagation/grid_task_enqueuer.ex

@@ -164,6 +164,52 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
     %{requeued: requeued_n, failed: failed_n}
   end
 
+  @doc """
+  Seed exactly one kind='forecast' row whose valid_time is closest to
+  `now` for the given `run_time`. Used by `HrdpsGridWorker` because the
+  /weather map only consumes the current-hour HRDPS cell — the rest of
+  the f01..f48 chain isn't worth the wgrib2 cost (each HRDPS chain step
+  is ~20 min wall time at the production point density).
+
+  Forecast hour is clamped to `1..@max_forecast_hour`. When `now` lands
+  before `run_time + 1h`, we still seed `fh=1`; when it's past
+  `run_time + @max_forecast_hour`, we cap there. The Rust worker rejects
+  `forecast_hour=0` (`F00Reserved`) so we never seed the analysis hour
+  through this path.
+  """
+  @spec seed_current_hour(DateTime.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()}
+  def seed_current_hour(%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)
+    target_valid = now |> DateTime.truncate(:second) |> Map.put(:minute, 0) |> Map.put(:second, 0)
+
+    fh =
+      target_valid
+      |> DateTime.diff(run_time, :second)
+      |> div(3600)
+      |> max(1)
+      |> min(@max_forecast_hour)
+
+    row = %{
+      id: Ecto.UUID.bingenerate(),
+      run_time: run_time,
+      forecast_hour: fh,
+      valid_time: DateTime.add(run_time, fh * 3600, :second),
+      status: "queued",
+      attempt: 0,
+      kind: "forecast",
+      source: source,
+      claimed_at: nil,
+      completed_at: nil,
+      error: nil,
+      inserted_at: now,
+      updated_at: now
+    }
+
+    do_insert([row], run_time, source)
+  end
+
   defp do_insert(rows, run_time, source) do
     {count, _} =
       Repo.insert_all("grid_tasks", rows,

lib/microwaveprop/weather/scalar_file.ex

@@ -218,19 +218,21 @@ defmodule Microwaveprop.Weather.ScalarFile do
         Path.join(dir_for_hrdps(valid_time), chunk_name)
       ],
       :miss,
-      fn path ->
-        if File.exists?(path) do
-          case find_point_in_chunk(path, snapped_lat, snapped_lon) do
-            {:ok, row} -> {:ok, row}
-            :miss -> false
-          end
-        else
-          false
-        end
-      end
+      &lookup_in_chunk(&1, snapped_lat, snapped_lon)
     )
   end
 
+  defp lookup_in_chunk(path, snapped_lat, snapped_lon) do
+    if File.exists?(path) do
+      hit_or_false(find_point_in_chunk(path, snapped_lat, snapped_lon))
+    else
+      false
+    end
+  end
+
+  defp hit_or_false({:ok, _} = hit), do: hit
+  defp hit_or_false(:miss), do: false
+
   defp find_point_in_chunk(path, snapped_lat, snapped_lon) do
     case Enum.find(decode_chunk(path), fn r ->
            r.lat == snapped_lat and r.lon == snapped_lon

lib/microwaveprop/workers/hrdps_grid_worker.ex

@@ -17,6 +17,15 @@ defmodule Microwaveprop.Workers.HrdpsGridWorker do
   `pipeline::run_chain_step_hrdps`, fetching from MSC Datamart, writing
   scores to `<base>/<band>/<iso>.hrdps.prop` (sibling of the HRRR
   `.prop` file for the same cycle). Cron entry lives in `runtime.exs`.
+
+  ## Scope
+
+  Seeds exactly one forecast row per fire — the one whose valid_time
+  matches the current hour. The Canadian propagation/weather UI only
+  surfaces the present moment, so producing 18 forecast hours per cycle
+  was wasted wgrib2 work (~20 min/chain step at production point
+  density). If forecast hours come back into scope later, swap the
+  enqueuer call back to `GridTaskEnqueuer.seed_with_analysis/2`.
   """
 
   use Oban.Worker,
@@ -46,9 +55,9 @@ defmodule Microwaveprop.Workers.HrdpsGridWorker do
 
     run_time = pick_run_time(DateTime.utc_now())
 
-    Logger.info("HrdpsGrid: seeding chain run_time=#{run_time} source=hrdps")
+    Logger.info("HrdpsGrid: seeding current-hour task run_time=#{run_time} source=hrdps")
 
-    case GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") do
+    case GridTaskEnqueuer.seed_current_hour(run_time, source: "hrdps") do
       {:ok, count} ->
         :telemetry.execute(
           [:microwaveprop, :hrdps, :grid_worker, :seeded],

rust/prop_grid_rs/src/grid.rs

@@ -20,6 +20,16 @@ pub const HRDPS_LAT_MAX: f64 = 60.0;
 pub const HRDPS_LON_MIN: f64 = -141.0;
 pub const HRDPS_LON_MAX: f64 = -52.0;
 
+// HRDPS uses a coarser grid than HRRR. Reason: each `wgrib2 -lon` point
+// extraction redundantly JPEG2000-decodes every matched record, and the
+// HRDPS rotated lat/lon source amplifies the per-record cost. At 0.125°
+// (matching HRRR) production observed ~5 min/batch × 57 batches = ~5 h
+// per chain step, far slower than the dev-bench claim of 30-90 s. Until
+// the decoder is rewritten to decode-once + lookup-many, 0.5° (~55 km
+// cells) keeps Canadian coverage visible on /weather without choking the
+// pipeline. ~3.5 k cells fit in 4 batches → ~20 min/chain step.
+pub const HRDPS_STEP: f64 = 0.5;
+
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct GridSpec {
     pub lon_start: f64,
@@ -61,15 +71,15 @@ pub fn conus_points() -> Vec<(f64, f64)> {
 /// extraction needs the full bbox so the score-file's lat_start/lon_start
 /// align with the cells the worker actually wrote.
 pub fn hrdps_grid_spec() -> GridSpec {
-    let lon_count = ((HRDPS_LON_MAX - HRDPS_LON_MIN) / STEP).round() as usize + 1;
-    let lat_count = ((HRDPS_LAT_MAX - HRDPS_LAT_MIN) / STEP).round() as usize + 1;
+    let lon_count = ((HRDPS_LON_MAX - HRDPS_LON_MIN) / HRDPS_STEP).round() as usize + 1;
+    let lat_count = ((HRDPS_LAT_MAX - HRDPS_LAT_MIN) / HRDPS_STEP).round() as usize + 1;
     GridSpec {
         lon_start: HRDPS_LON_MIN,
         lon_count,
-        lon_step: STEP,
+        lon_step: HRDPS_STEP,
         lat_start: HRDPS_LAT_MIN,
         lat_count,
-        lat_step: STEP,
+        lat_step: HRDPS_STEP,
     }
 }
 
@@ -84,6 +94,12 @@ pub fn hrdps_only_points() -> Vec<(f64, f64)> {
         let lat = round3(spec.lat_start + j as f64 * spec.lat_step);
         for i in 0..spec.lon_count {
             let lon = round3(spec.lon_start + i as f64 * spec.lon_step);
+            // HRDPS now walks at 0.5° while HRRR walks at 0.125°, so the
+            // coarse cells are no longer subset-aligned with HRRR's CONUS
+            // grid. The disjointness rule is "if a coarse cell *center*
+            // falls inside HRRR's lat/lon range, drop it" — HRRR will
+            // cover that area at finer resolution anyway and the
+            // /weather merge prefers HRRR rows on collision.
             if !in_conus_bbox(lat, lon) {
                 out.push((lat, lon));
             }
@@ -135,15 +151,18 @@ mod tests {
     }
 
     #[test]
-    fn hrdps_grid_spec_matches_elixir_bbox() {
+    fn hrdps_grid_spec_uses_coarse_step() {
         let spec = hrdps_grid_spec();
-        // Elixir Grid: hrdps_lat_min=49, hrdps_lat_max=60, hrdps_lon_min=-141, hrdps_lon_max=-52
-        // lon_count = (-52 - -141) / 0.125 + 1 = 713
-        // lat_count = (60 - 49) / 0.125 + 1 = 89
-        assert_eq!(spec.lon_count, 713);
-        assert_eq!(spec.lat_count, 89);
+        // HRDPS_STEP = 0.5 (coarser than HRRR's 0.125) — see HRDPS_STEP
+        // doc for the wgrib2 perf rationale.
+        // lon_count = (-52 - -141) / 0.5 + 1 = 179
+        // lat_count = (60 - 49) / 0.5 + 1 = 23
+        assert_eq!(spec.lon_count, 179);
+        assert_eq!(spec.lat_count, 23);
         assert_eq!(spec.lon_start, -141.0);
         assert_eq!(spec.lat_start, 49.0);
+        assert_eq!(spec.lon_step, 0.5);
+        assert_eq!(spec.lat_step, 0.5);
     }
 
     #[test]

test/microwaveprop/propagation/grid_task_enqueuer_test.exs

@@ -191,6 +191,68 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuerTest do
     end
   end
 
+  describe "seed_current_hour/2" do
+    test "seeds exactly one forecast row at the hour offset closest to now" do
+      # run_time 5h before "now" → fh should be 5.
+      now = DateTime.utc_now() |> DateTime.truncate(:second) |> Map.put(:minute, 0) |> Map.put(:second, 0)
+      run_time = DateTime.add(now, -5 * 3600, :second)
+
+      assert {:ok, 1} = GridTaskEnqueuer.seed_current_hour(run_time, source: "hrdps")
+
+      rows =
+        Repo.all(
+          from(g in "grid_tasks",
+            where: g.run_time == ^run_time and g.source == "hrdps",
+            select: %{kind: g.kind, forecast_hour: g.forecast_hour, valid_time: g.valid_time}
+          )
+        )
+
+      assert length(rows) == 1
+      [%{kind: kind, forecast_hour: fh, valid_time: vt}] = rows
+      assert kind == "forecast"
+      assert fh == 5
+      assert NaiveDateTime.compare(vt, DateTime.to_naive(now)) == :eq
+    end
+
+    test "is idempotent on the (run_time, fh, kind, source) key" do
+      now = DateTime.utc_now() |> DateTime.truncate(:second) |> Map.put(:minute, 0) |> Map.put(:second, 0)
+      run_time = DateTime.add(now, -3 * 3600, :second)
+
+      assert {:ok, 1} = GridTaskEnqueuer.seed_current_hour(run_time, source: "hrdps")
+      assert {:ok, 0} = GridTaskEnqueuer.seed_current_hour(run_time, source: "hrdps")
+
+      count =
+        Repo.one(
+          from(g in "grid_tasks",
+            where: g.run_time == ^run_time and g.source == "hrdps",
+            select: count()
+          )
+        )
+
+      assert count == 1
+    end
+
+    test "clamps fh to 1 when the cycle is in the future" do
+      # run_time 1h *after* now → diff is -3600 → would naturally be 0 or
+      # negative. Clamp pulls it to 1 so the F00Reserved branch in Rust
+      # is never hit through this path.
+      now = DateTime.utc_now() |> DateTime.truncate(:second) |> Map.put(:minute, 0) |> Map.put(:second, 0)
+      run_time = DateTime.add(now, 3600, :second)
+
+      assert {:ok, 1} = GridTaskEnqueuer.seed_current_hour(run_time, source: "hrdps")
+
+      [%{forecast_hour: fh}] =
+        Repo.all(
+          from(g in "grid_tasks",
+            where: g.run_time == ^run_time and g.source == "hrdps",
+            select: %{forecast_hour: g.forecast_hour}
+          )
+        )
+
+      assert fh == 1
+    end
+  end
+
   defp insert_task(attrs) do
     now = DateTime.truncate(DateTime.utc_now(), :microsecond)
 

test/microwaveprop/workers/hrdps_grid_worker_test.exs

@@ -40,15 +40,25 @@ defmodule Microwaveprop.Workers.HrdpsGridWorkerTest do
   end
 
   describe "perform/1" do
-    test "seeds 19 grid_tasks rows tagged with source='hrdps'" do
+    test "seeds exactly one current-hour grid_task row 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()))
+      rows =
+        Repo.all(
+          from(g in "grid_tasks",
+            where: g.source == "hrdps",
+            select: %{kind: g.kind, forecast_hour: g.forecast_hour}
+          )
+        )
 
-      assert sources_count == 19
+      # One forecast row, no analysis. /weather only consumes current
+      # hour so we don't pay the wgrib2 cost for the rest of the chain.
+      assert length(rows) == 1
+      [%{kind: kind, forecast_hour: fh}] = rows
+      assert kind == "forecast"
+      assert fh >= 1 and fh <= 18
     end
 
     test "is idempotent — re-running the worker for the same cycle inserts nothing extra" do
@@ -60,7 +70,7 @@ defmodule Microwaveprop.Workers.HrdpsGridWorkerTest do
       total =
         Repo.one(from(g in "grid_tasks", where: g.source == "hrdps", select: count()))
 
-      assert total == 19
+      assert total == 1
     end
   end
 end