feat(status): show Rust prop-grid-rs work queue alongside Oban panel
Mirrors the Oban Job Queue Status table with a new section sourced from the grid_tasks hand-off table. Columns: Running, Queued, Retrying, Done (1h). Active rows render an f## badge per in-flight forecast hour and a detail line below with run_time / fh / attempt so it's obvious which HRRR cycle the worker is on. Pipes through the existing 2s debounced refresh plus a new PubSub subscription to "propagation:updated" — PropagationNotifyListener fans out the Rust worker's NOTIFY propagation_ready, so the panel transitions running → done the moment Rust writes the score file.
This commit is contained in:
parent
52c062c90a
commit
234a27b9bc
2 changed files with 173 additions and 0 deletions
|
|
@ -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
|
|||
|
||||
<div class="divider" />
|
||||
|
||||
<h2 class="text-base font-semibold mb-2">Rust Grid Worker (prop-grid-rs)</h2>
|
||||
<div id="grid-tasks-panel" class="overflow-x-auto mb-4">
|
||||
<table class="table table-sm">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Worker</th>
|
||||
<th>Running</th>
|
||||
<th>Queued</th>
|
||||
<th>Retrying</th>
|
||||
<th>Done (1h)</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
<tr>
|
||||
<td class="font-semibold">prop-grid-rs</td>
|
||||
<td>
|
||||
<%= if @grid_tasks.running == [] do %>
|
||||
0
|
||||
<% else %>
|
||||
<span class="flex items-center gap-2">
|
||||
<span class="loading loading-spinner loading-xs"></span>
|
||||
<span class="font-mono text-xs">
|
||||
<%= for r <- @grid_tasks.running do %>
|
||||
<span class="badge badge-sm badge-info mr-1">
|
||||
f{String.pad_leading(Integer.to_string(r.forecast_hour), 2, "0")}
|
||||
</span>
|
||||
<% end %>
|
||||
</span>
|
||||
<span class="opacity-60 text-xs">running</span>
|
||||
</span>
|
||||
<% end %>
|
||||
</td>
|
||||
<td>{@grid_tasks.queued}</td>
|
||||
<td>
|
||||
<%= if @grid_tasks.failed > 0 do %>
|
||||
<span class="text-warning">{@grid_tasks.failed}</span>
|
||||
<% else %>
|
||||
0
|
||||
<% end %>
|
||||
</td>
|
||||
<td>{format_number(@grid_tasks.done_1h)}</td>
|
||||
</tr>
|
||||
</tbody>
|
||||
</table>
|
||||
<%= if @grid_tasks.running != [] do %>
|
||||
<div class="text-xs text-base-content/60 mt-1">
|
||||
<%= for r <- @grid_tasks.running do %>
|
||||
<span class="font-mono mr-3">
|
||||
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}
|
||||
</span>
|
||||
<% end %>
|
||||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
|
||||
<h2 class="text-base font-semibold mb-2">Enrichment Status by Type</h2>
|
||||
<div class="grid grid-cols-2 md:grid-cols-4 gap-4 mb-6">
|
||||
<%= for {type, label} <- [{"hrrr", "HRRR"}, {"weather", "Weather"}, {"terrain", "Terrain"}, {"iemre", "IEMRE"}, {"narr", "NARR"}] do %>
|
||||
|
|
|
|||
|
|
@ -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>.*?<td>2<\/td>/s
|
||||
assert panel =~ ~r/Done \(1h\)[^0-9]*<\/th>.*?<td>1<\/td>/s
|
||||
end
|
||||
end
|
||||
|
||||
defp create_narr_candidate do
|
||||
{:ok, contact} =
|
||||
%Contact{}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue