Graham McIntire
aab3c3736c
HRRR publishes f21-f48 at 3-hourly intervals in addition to the hourly f01-f18 we already fetch. This extends the grid_tasks seed from 19 rows (1 analysis + 18 forecast) to 29 rows (1 analysis + 18 hourly + 10 3-hourly), and bumps the UI forecast window constant from 18h to 48h so the map timeline and path calculator forecast chart automatically show the extended horizon. No Rust logic changes needed — the pipeline is data-driven and u8 forecast_hour already handles f48.
278 lines
9 KiB
Elixir
278 lines
9 KiB
Elixir
defmodule Microwaveprop.Propagation.GridTaskEnqueuerTest do
|
|
use Microwaveprop.DataCase, async: true
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Propagation.GridTaskEnqueuer
|
|
alias Microwaveprop.Repo
|
|
|
|
test "seeds one row per forecast hour" do
|
|
run_time = ~U[2026-04-19 15:00:00Z]
|
|
assert {:ok, 28} = GridTaskEnqueuer.seed(run_time)
|
|
|
|
rows =
|
|
Repo.all(
|
|
from(g in "grid_tasks",
|
|
where: g.run_time == ^run_time,
|
|
select: %{forecast_hour: g.forecast_hour, status: g.status, valid_time: g.valid_time}
|
|
)
|
|
)
|
|
|
|
expected_fhs = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 21, 24, 27, 30, 33, 36, 39, 42, 45, 48]
|
|
assert length(rows) == 28
|
|
assert rows |> Enum.map(& &1.forecast_hour) |> Enum.sort() == expected_fhs
|
|
assert Enum.all?(rows, &(&1.status == "queued"))
|
|
|
|
# Raw schemaless selects return NaiveDateTime; compare against the
|
|
# equivalent naive form.
|
|
expected_naive = DateTime.to_naive(run_time)
|
|
|
|
Enum.each(rows, fn r ->
|
|
expected = NaiveDateTime.add(expected_naive, r.forecast_hour * 3600, :second)
|
|
assert NaiveDateTime.compare(r.valid_time, expected) == :eq
|
|
end)
|
|
end
|
|
|
|
test "is idempotent: reseeding the same run_time inserts nothing" do
|
|
run_time = ~U[2026-04-19 16:00:00Z]
|
|
assert {:ok, 28} = GridTaskEnqueuer.seed(run_time)
|
|
assert {:ok, 0} = GridTaskEnqueuer.seed(run_time)
|
|
|
|
count = Repo.one(from(g in "grid_tasks", where: g.run_time == ^run_time, select: count()))
|
|
|
|
assert count == 28
|
|
end
|
|
|
|
test "never emits a row for fh=0 — Elixir keeps that chain step" do
|
|
run_time = ~U[2026-04-19 17:00:00Z]
|
|
assert {:ok, 28} = GridTaskEnqueuer.seed(run_time)
|
|
|
|
f0_count = Repo.one(from(g in "grid_tasks", where: g.run_time == ^run_time and g.forecast_hour == 0, select: count()))
|
|
|
|
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, 28} = 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, 29} = 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) == 29
|
|
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, 29} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrrr")
|
|
assert {:ok, 29} = 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 == 58
|
|
end
|
|
|
|
test "reseeding the same run_time + source is idempotent" do
|
|
run_time = ~U[2026-04-29 06:00:00Z]
|
|
assert {:ok, 29} = 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)
|
|
stale = DateTime.add(now, -30 * 60, :second)
|
|
|
|
insert_task(%{
|
|
run_time: ~U[2026-04-19 20:00:00Z],
|
|
forecast_hour: 5,
|
|
valid_time: ~U[2026-04-20 01:00:00Z],
|
|
status: "running",
|
|
attempt: 1,
|
|
claimed_at: stale
|
|
})
|
|
|
|
assert %{requeued: 1, failed: 0} = GridTaskEnqueuer.reclaim_stale_running()
|
|
|
|
row =
|
|
Repo.one!(
|
|
from(t in "grid_tasks",
|
|
where: t.run_time == ^~U[2026-04-19 20:00:00Z] and t.forecast_hour == 5,
|
|
select: %{status: t.status, claimed_at: t.claimed_at, attempt: t.attempt}
|
|
)
|
|
)
|
|
|
|
assert row.status == "queued"
|
|
assert is_nil(row.claimed_at)
|
|
# attempt is NOT incremented — claim_next does that on re-claim.
|
|
assert row.attempt == 1
|
|
end
|
|
|
|
test "leaves recently-claimed running rows untouched" do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
fresh = DateTime.add(now, -60, :second)
|
|
|
|
insert_task(%{
|
|
run_time: ~U[2026-04-19 21:00:00Z],
|
|
forecast_hour: 6,
|
|
valid_time: ~U[2026-04-20 03:00:00Z],
|
|
status: "running",
|
|
attempt: 1,
|
|
claimed_at: fresh
|
|
})
|
|
|
|
assert %{requeued: 0, failed: 0} = GridTaskEnqueuer.reclaim_stale_running()
|
|
|
|
row =
|
|
Repo.one!(
|
|
from(t in "grid_tasks",
|
|
where: t.run_time == ^~U[2026-04-19 21:00:00Z] and t.forecast_hour == 6,
|
|
select: %{status: t.status}
|
|
)
|
|
)
|
|
|
|
assert row.status == "running"
|
|
end
|
|
|
|
test "flips stale rows past max attempts to failed" do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
stale = DateTime.add(now, -30 * 60, :second)
|
|
|
|
insert_task(%{
|
|
run_time: ~U[2026-04-19 22:00:00Z],
|
|
forecast_hour: 7,
|
|
valid_time: ~U[2026-04-20 05:00:00Z],
|
|
status: "running",
|
|
attempt: 5,
|
|
claimed_at: stale
|
|
})
|
|
|
|
assert %{requeued: 0, failed: 1} = GridTaskEnqueuer.reclaim_stale_running()
|
|
|
|
row =
|
|
Repo.one!(
|
|
from(t in "grid_tasks",
|
|
where: t.run_time == ^~U[2026-04-19 22:00:00Z] and t.forecast_hour == 7,
|
|
select: %{status: t.status, error: t.error, claimed_at: t.claimed_at}
|
|
)
|
|
)
|
|
|
|
assert row.status == "failed"
|
|
assert is_nil(row.claimed_at)
|
|
assert row.error =~ "reclaim"
|
|
end
|
|
|
|
test "does not touch queued or done rows even if claimed_at is ancient" do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
stale = DateTime.add(now, -30 * 60, :second)
|
|
|
|
insert_task(%{
|
|
run_time: ~U[2026-04-19 23:00:00Z],
|
|
forecast_hour: 8,
|
|
valid_time: ~U[2026-04-20 07:00:00Z],
|
|
status: "done",
|
|
attempt: 1,
|
|
claimed_at: stale,
|
|
completed_at: stale
|
|
})
|
|
|
|
assert %{requeued: 0, failed: 0} = GridTaskEnqueuer.reclaim_stale_running()
|
|
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)
|
|
|
|
row =
|
|
Map.merge(
|
|
%{
|
|
id: Ecto.UUID.bingenerate(),
|
|
status: "queued",
|
|
attempt: 0,
|
|
kind: "forecast",
|
|
claimed_at: nil,
|
|
completed_at: nil,
|
|
error: nil,
|
|
inserted_at: now,
|
|
updated_at: now
|
|
},
|
|
attrs
|
|
)
|
|
|
|
{1, _} = Repo.insert_all("grid_tasks", [row])
|
|
end
|
|
end
|