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