diff --git a/lib/microwaveprop_web/live/status_live.ex b/lib/microwaveprop_web/live/status_live.ex
index 82eeaee6..d0d8e5b0 100644
--- a/lib/microwaveprop_web/live/status_live.ex
+++ b/lib/microwaveprop_web/live/status_live.ex
@@ -14,11 +14,17 @@ defmodule MicrowavepropWeb.StatusLive do
if connected?(socket) do
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:contact_status")
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:oban_jobs")
+ # Rust prop-grid-rs emits NOTIFY propagation_ready on completion —
+ # PropagationNotifyListener fans it out as `propagation:updated` so
+ # the grid_tasks panel transitions from "running" → "done" without
+ # a manual refresh.
+ Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated")
end
stats = fetch_stats()
unprocessed = count_unprocessed()
db_stats = fetch_db_stats()
+ grid_tasks = fetch_grid_tasks_stats()
{:ok,
assign(socket,
@@ -26,6 +32,7 @@ defmodule MicrowavepropWeb.StatusLive do
stats: stats,
unprocessed: unprocessed,
db_stats: db_stats,
+ grid_tasks: grid_tasks,
refresh_timer: nil
)}
end
@@ -39,12 +46,17 @@ defmodule MicrowavepropWeb.StatusLive do
{:noreply, schedule_refresh(socket)}
end
+ def handle_info({:propagation_updated, _valid_times}, socket) do
+ {:noreply, schedule_refresh(socket)}
+ end
+
def handle_info(:refresh_stats, socket) do
{:noreply,
assign(socket,
stats: fetch_stats(),
unprocessed: count_unprocessed(),
db_stats: fetch_db_stats(),
+ grid_tasks: fetch_grid_tasks_stats(),
refresh_timer: nil
)}
end
@@ -165,6 +177,53 @@ defmodule MicrowavepropWeb.StatusLive do
)
end
+ defp fetch_grid_tasks_stats do
+ Cache.fetch_or_store({__MODULE__, :grid_tasks_stats}, 2_000, fn -> fetch_grid_tasks_stats_raw() end)
+ end
+
+ # Snapshot of the Rust prop-grid-rs work queue. Mirrors the Oban panel —
+ # "what's running, what's pending, what's retrying" — but pulls from the
+ # grid_tasks hand-off table instead of oban_jobs.
+ defp fetch_grid_tasks_stats_raw do
+ running =
+ Repo.all(
+ from(t in "grid_tasks",
+ where: t.status == "running",
+ order_by: [asc: t.run_time, asc: t.forecast_hour],
+ select: %{
+ run_time: t.run_time,
+ forecast_hour: t.forecast_hour,
+ attempt: t.attempt,
+ claimed_at: t.claimed_at
+ }
+ )
+ )
+
+ counts_by_status =
+ from(t in "grid_tasks",
+ where: t.status in ["queued", "failed"],
+ group_by: t.status,
+ select: {t.status, count(t.id)}
+ )
+ |> Repo.all()
+ |> Map.new()
+
+ done_1h =
+ Repo.one(
+ from(t in "grid_tasks",
+ where: t.status == "done" and t.completed_at > ago(1, "hour"),
+ select: count(t.id)
+ )
+ )
+
+ %{
+ running: running,
+ queued: Map.get(counts_by_status, "queued", 0),
+ failed: Map.get(counts_by_status, "failed", 0),
+ done_1h: done_1h
+ }
+ end
+
defp fetch_stats do
Cache.fetch_or_store({__MODULE__, :stats}, 2_000, fn -> fetch_stats_raw() end)
end
@@ -456,6 +515,62 @@ defmodule MicrowavepropWeb.StatusLive do
+ Rust Grid Worker (prop-grid-rs)
+
+
+
+
+ | Worker |
+ Running |
+ Queued |
+ Retrying |
+ Done (1h) |
+
+
+
+
+ | prop-grid-rs |
+
+ <%= if @grid_tasks.running == [] do %>
+ 0
+ <% else %>
+
+
+
+ <%= for r <- @grid_tasks.running do %>
+
+ f{String.pad_leading(Integer.to_string(r.forecast_hour), 2, "0")}
+
+ <% end %>
+
+ running
+
+ <% end %>
+ |
+ {@grid_tasks.queued} |
+
+ <%= if @grid_tasks.failed > 0 do %>
+ {@grid_tasks.failed}
+ <% else %>
+ 0
+ <% end %>
+ |
+ {format_number(@grid_tasks.done_1h)} |
+
+
+
+ <%= if @grid_tasks.running != [] do %>
+
+ <%= for r <- @grid_tasks.running do %>
+
+ run_time={Calendar.strftime(r.run_time, "%Y-%m-%d %H:%M")}Z
+ fh=f{String.pad_leading(Integer.to_string(r.forecast_hour), 2, "0")} attempt={r.attempt}
+
+ <% end %>
+
+ <% end %>
+
Enrichment Status by Type
<%= for {type, label} <- [{"hrrr", "HRRR"}, {"weather", "Weather"}, {"terrain", "Terrain"}, {"iemre", "IEMRE"}, {"narr", "NARR"}] do %>
diff --git a/test/microwaveprop_web/live/status_live_test.exs b/test/microwaveprop_web/live/status_live_test.exs
index 70f568d9..e6e1d03d 100644
--- a/test/microwaveprop_web/live/status_live_test.exs
+++ b/test/microwaveprop_web/live/status_live_test.exs
@@ -13,6 +13,7 @@ defmodule MicrowavepropWeb.StatusLiveTest do
Cache.invalidate({MicrowavepropWeb.StatusLive, :count_unprocessed})
Cache.invalidate({MicrowavepropWeb.StatusLive, :stats})
Cache.invalidate({MicrowavepropWeb.StatusLive, :db_stats})
+ Cache.invalidate({MicrowavepropWeb.StatusLive, :grid_tasks_stats})
user = user_fixture()
{:ok, admin} = user |> Ecto.Changeset.change(%{is_admin: true}) |> Repo.update()
@@ -20,6 +21,63 @@ defmodule MicrowavepropWeb.StatusLiveTest do
{:ok, conn: conn}
end
+ describe "grid_tasks panel — Rust prop-grid-rs worker" do
+ defp insert_grid_task(attrs) do
+ id = Ecto.UUID.bingenerate()
+ now = DateTime.truncate(DateTime.utc_now(), :second)
+
+ row =
+ Map.merge(
+ %{
+ id: id,
+ run_time: ~N[2026-04-19 19:00:00],
+ forecast_hour: 5,
+ valid_time: ~N[2026-04-20 00:00:00],
+ status: "queued",
+ attempt: 0,
+ inserted_at: now,
+ updated_at: now
+ },
+ attrs
+ )
+
+ {1, nil} = Repo.insert_all("grid_tasks", [row])
+ id
+ end
+
+ test "shows an in-flight row with the forecast hour it's working on", %{conn: conn} do
+ insert_grid_task(%{
+ status: "running",
+ forecast_hour: 7,
+ attempt: 1,
+ claimed_at: DateTime.truncate(DateTime.utc_now(), :microsecond)
+ })
+
+ {:ok, lv, _html} = live(conn, ~p"/status")
+ panel = lv |> element("#grid-tasks-panel") |> render()
+ assert panel =~ "prop-grid-rs"
+ assert panel =~ "f07"
+ assert panel =~ "running"
+ end
+
+ test "shows queued and completed counts", %{conn: conn} do
+ insert_grid_task(%{forecast_hour: 1, status: "queued"})
+ insert_grid_task(%{forecast_hour: 2, status: "queued"})
+
+ insert_grid_task(%{
+ forecast_hour: 3,
+ status: "done",
+ completed_at: DateTime.truncate(DateTime.utc_now(), :microsecond)
+ })
+
+ {:ok, lv, _html} = live(conn, ~p"/status")
+ panel = lv |> element("#grid-tasks-panel") |> render()
+
+ assert panel =~ ~r/Queued[^0-9]*<\/th>.*?
2<\/td>/s
+ assert panel =~ ~r/Done \(1h\)[^0-9]*<\/th>.*? | 1<\/td>/s
+ end
+ end
+
defp create_narr_candidate do
{:ok, contact} =
%Contact{}
|