From d5726a89d5ae6a30fb0c66e9217809a3aecd17bb Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 30 Apr 2026 10:36:00 -0500 Subject: [PATCH] =?UTF-8?q?feat(hrdps):=20scope=20to=20current-hour-only?= =?UTF-8?q?=20at=200.5=C2=B0=20step?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 13 ++-- .../propagation/grid_task_enqueuer.ex | 46 ++++++++++++++ lib/microwaveprop/weather/scalar_file.ex | 22 ++++--- .../workers/hrdps_grid_worker.ex | 13 +++- rust/prop_grid_rs/src/grid.rs | 39 +++++++++--- .../propagation/grid_task_enqueuer_test.exs | 62 +++++++++++++++++++ .../workers/hrdps_grid_worker_test.exs | 20 ++++-- 7 files changed, 182 insertions(+), 33 deletions(-) diff --git a/config/runtime.exs b/config/runtime.exs index da5d8e78..fc6ec5e7 100644 --- a/config/runtime.exs +++ b/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 diff --git a/lib/microwaveprop/propagation/grid_task_enqueuer.ex b/lib/microwaveprop/propagation/grid_task_enqueuer.ex index e0a8d234..7ebbc062 100644 --- a/lib/microwaveprop/propagation/grid_task_enqueuer.ex +++ b/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, diff --git a/lib/microwaveprop/weather/scalar_file.ex b/lib/microwaveprop/weather/scalar_file.ex index 7b7c3172..1c48a075 100644 --- a/lib/microwaveprop/weather/scalar_file.ex +++ b/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 diff --git a/lib/microwaveprop/workers/hrdps_grid_worker.ex b/lib/microwaveprop/workers/hrdps_grid_worker.ex index 65aeee02..9f39e640 100644 --- a/lib/microwaveprop/workers/hrdps_grid_worker.ex +++ b/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 `//.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], diff --git a/rust/prop_grid_rs/src/grid.rs b/rust/prop_grid_rs/src/grid.rs index 258cc4ba..eb002d1b 100644 --- a/rust/prop_grid_rs/src/grid.rs +++ b/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] diff --git a/test/microwaveprop/propagation/grid_task_enqueuer_test.exs b/test/microwaveprop/propagation/grid_task_enqueuer_test.exs index 05f01245..970ecb36 100644 --- a/test/microwaveprop/propagation/grid_task_enqueuer_test.exs +++ b/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) diff --git a/test/microwaveprop/workers/hrdps_grid_worker_test.exs b/test/microwaveprop/workers/hrdps_grid_worker_test.exs index 3890aa7c..3bf547e6 100644 --- a/test/microwaveprop/workers/hrdps_grid_worker_test.exs +++ b/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