Show all executing pipeline workers in the status chip
The :propagation Oban queue runs with concurrency 2, so the grid
worker chain step and the 10-minute ASOS nudge can (and do) execute
in parallel. PipelineStatus.current/0 used to return only the
most-recently-started worker, which made the chip label flip to
"ASOS nudge" at every :00 / :10 tick even though HRRR was still
chugging through forecast hours behind it.
Replace the single :detail field with a list of running_detail
maps, each tagged with :grid or :asos so the chip can:
- Render one sub-line per currently-executing worker
- Sort them deterministically (grid before asos) so the eye
doesn't see rows swap
- Still override the grid entry with the forecast-hour progress
label when a propagation_pipeline_progress broadcast arrives
This commit is contained in:
parent
456e38d750
commit
c7a1e352a7
3 changed files with 87 additions and 34 deletions
|
|
@ -30,10 +30,14 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
|
||||||
|
|
||||||
@type state :: :running | :idle | :stale | :unknown
|
@type state :: :running | :idle | :stale | :unknown
|
||||||
|
|
||||||
|
@type worker_tag :: :grid | :asos
|
||||||
|
|
||||||
|
@type running_detail :: %{worker: worker_tag(), label: String.t()}
|
||||||
|
|
||||||
@type t :: %{
|
@type t :: %{
|
||||||
state: state(),
|
state: state(),
|
||||||
label: String.t(),
|
label: String.t(),
|
||||||
detail: String.t() | nil,
|
details: [running_detail()],
|
||||||
last_update_at: DateTime.t() | nil
|
last_update_at: DateTime.t() | nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -52,30 +56,41 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
|
||||||
|
|
||||||
@spec current() :: t()
|
@spec current() :: t()
|
||||||
def current do
|
def current do
|
||||||
case running_worker() do
|
case running_workers() do
|
||||||
worker when is_binary(worker) ->
|
[] ->
|
||||||
|
build_idle_or_stale(latest_completed_at())
|
||||||
|
|
||||||
|
details ->
|
||||||
%{
|
%{
|
||||||
state: :running,
|
state: :running,
|
||||||
label: "Updating propagation",
|
label: "Updating propagation",
|
||||||
detail: base_running_detail(worker),
|
details: details,
|
||||||
last_update_at: latest_completed_at()
|
last_update_at: latest_completed_at()
|
||||||
}
|
}
|
||||||
|
|
||||||
nil ->
|
|
||||||
build_idle_or_stale(latest_completed_at())
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp running_worker do
|
# Returns every pipeline worker currently in executing state, in a
|
||||||
Repo.one(
|
# stable grid-before-asos order so the chip sub-lines don't flicker
|
||||||
from j in "oban_jobs",
|
# as jobs come and go.
|
||||||
where: j.state == "executing" and j.worker in ^@workers,
|
defp running_workers do
|
||||||
order_by: [desc: j.attempted_at],
|
from(j in "oban_jobs",
|
||||||
limit: 1,
|
where: j.state == "executing" and j.worker in ^@workers,
|
||||||
select: j.worker
|
distinct: j.worker,
|
||||||
|
select: j.worker
|
||||||
)
|
)
|
||||||
|
|> Repo.all()
|
||||||
|
|> Enum.sort_by(&worker_sort_key/1)
|
||||||
|
|> Enum.map(&to_running_detail/1)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defp to_running_detail(@grid_worker), do: %{worker: :grid, label: "HRRR run"}
|
||||||
|
defp to_running_detail(@asos_worker), do: %{worker: :asos, label: "ASOS nudge"}
|
||||||
|
|
||||||
|
defp worker_sort_key(@grid_worker), do: 0
|
||||||
|
defp worker_sort_key(@asos_worker), do: 1
|
||||||
|
defp worker_sort_key(_), do: 99
|
||||||
|
|
||||||
defp latest_completed_at do
|
defp latest_completed_at do
|
||||||
case Repo.one(
|
case Repo.one(
|
||||||
from j in "oban_jobs",
|
from j in "oban_jobs",
|
||||||
|
|
@ -91,7 +106,7 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
|
||||||
end
|
end
|
||||||
|
|
||||||
defp build_idle_or_stale(nil) do
|
defp build_idle_or_stale(nil) do
|
||||||
%{state: :unknown, label: "Propagation status unknown", detail: nil, last_update_at: nil}
|
%{state: :unknown, label: "Propagation status unknown", details: [], last_update_at: nil}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp build_idle_or_stale(%DateTime{} = last) do
|
defp build_idle_or_stale(%DateTime{} = last) do
|
||||||
|
|
@ -101,23 +116,19 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
|
||||||
%{
|
%{
|
||||||
state: :stale,
|
state: :stale,
|
||||||
label: "Propagation data stale · last update #{format_age(age_minutes)} ago",
|
label: "Propagation data stale · last update #{format_age(age_minutes)} ago",
|
||||||
detail: nil,
|
details: [],
|
||||||
last_update_at: last
|
last_update_at: last
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
%{
|
%{
|
||||||
state: :idle,
|
state: :idle,
|
||||||
label: "Up to date · #{format_age(age_minutes)} ago",
|
label: "Up to date · #{format_age(age_minutes)} ago",
|
||||||
detail: nil,
|
details: [],
|
||||||
last_update_at: last
|
last_update_at: last
|
||||||
}
|
}
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp base_running_detail(@grid_worker), do: "HRRR run"
|
|
||||||
defp base_running_detail(@asos_worker), do: "ASOS nudge"
|
|
||||||
defp base_running_detail(_), do: nil
|
|
||||||
|
|
||||||
defp format_age(0), do: "just now"
|
defp format_age(0), do: "just now"
|
||||||
defp format_age(1), do: "1m"
|
defp format_age(1), do: "1m"
|
||||||
defp format_age(n) when n < 60, do: "#{n}m"
|
defp format_age(n) when n < 60, do: "#{n}m"
|
||||||
|
|
|
||||||
|
|
@ -412,13 +412,13 @@ defmodule MicrowavepropWeb.MapLive do
|
||||||
attr :progress, :any, default: nil
|
attr :progress, :any, default: nil
|
||||||
|
|
||||||
defp pipeline_status_chip(assigns) do
|
defp pipeline_status_chip(assigns) do
|
||||||
{main, detail} = chip_display_parts(assigns.status, assigns.progress)
|
details = chip_detail_labels(assigns.status, assigns.progress)
|
||||||
|
|
||||||
assigns =
|
assigns =
|
||||||
assigns
|
assigns
|
||||||
|> assign(:display_main, main)
|
|> assign(:display_main, assigns.status.label)
|
||||||
|> assign(:display_detail, detail)
|
|> assign(:display_details, details)
|
||||||
|> assign(:display_title, if(detail, do: "#{main} — #{detail}", else: main))
|
|> assign(:display_title, chip_title(assigns.status.label, details))
|
||||||
|
|
||||||
~H"""
|
~H"""
|
||||||
<div
|
<div
|
||||||
|
|
@ -435,8 +435,8 @@ defmodule MicrowavepropWeb.MapLive do
|
||||||
/>
|
/>
|
||||||
<span class="opacity-90 break-words min-w-0">
|
<span class="opacity-90 break-words min-w-0">
|
||||||
{@display_main}
|
{@display_main}
|
||||||
<%= if @display_detail do %>
|
<%= for d <- @display_details do %>
|
||||||
<span class="block text-[11px] opacity-70">{@display_detail}</span>
|
<span class="block text-[11px] opacity-70">{d}</span>
|
||||||
<% end %>
|
<% end %>
|
||||||
</span>
|
</span>
|
||||||
<% :idle -> %>
|
<% :idle -> %>
|
||||||
|
|
@ -453,11 +453,23 @@ defmodule MicrowavepropWeb.MapLive do
|
||||||
"""
|
"""
|
||||||
end
|
end
|
||||||
|
|
||||||
defp chip_display_parts(%{state: :running} = status, progress) do
|
# Map the PipelineStatus details list to a plain list of sub-line
|
||||||
{status.label, PipelineStatus.running_detail(progress) || status.detail}
|
# strings. For the grid worker specifically, the forecast-hour
|
||||||
|
# progress payload overrides the base label so the chip can show
|
||||||
|
# "+3h" / "now" instead of the generic "HRRR run".
|
||||||
|
defp chip_detail_labels(%{state: :running, details: details}, progress) do
|
||||||
|
override = PipelineStatus.running_detail(progress)
|
||||||
|
|
||||||
|
Enum.map(details, fn
|
||||||
|
%{worker: :grid} -> override || "HRRR run"
|
||||||
|
%{label: label} -> label
|
||||||
|
end)
|
||||||
end
|
end
|
||||||
|
|
||||||
defp chip_display_parts(status, _progress), do: {status.label, nil}
|
defp chip_detail_labels(_status, _progress), do: []
|
||||||
|
|
||||||
|
defp chip_title(main, []), do: main
|
||||||
|
defp chip_title(main, details), do: "#{main} — #{Enum.join(details, " · ")}"
|
||||||
|
|
||||||
@impl true
|
@impl true
|
||||||
def render(assigns) do
|
def render(assigns) do
|
||||||
|
|
|
||||||
|
|
@ -41,7 +41,7 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
|
||||||
assert is_binary(status.label)
|
assert is_binary(status.label)
|
||||||
end
|
end
|
||||||
|
|
||||||
test "returns :running with grid worker detail when PropagationGridWorker is executing" do
|
test "returns :running with a single grid-worker detail when only PropagationGridWorker is executing" do
|
||||||
insert_oban_job(%{
|
insert_oban_job(%{
|
||||||
state: "executing",
|
state: "executing",
|
||||||
worker: @grid_worker,
|
worker: @grid_worker,
|
||||||
|
|
@ -52,10 +52,11 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
|
||||||
|
|
||||||
assert status.state == :running
|
assert status.state == :running
|
||||||
assert status.label == "Updating propagation"
|
assert status.label == "Updating propagation"
|
||||||
assert status.detail =~ "HRRR"
|
assert [%{worker: :grid, label: grid_label}] = status.details
|
||||||
|
assert grid_label =~ "HRRR"
|
||||||
end
|
end
|
||||||
|
|
||||||
test "returns :running with ASOS detail when AsosAdjustmentWorker is executing" do
|
test "returns :running with a single ASOS detail when only AsosAdjustmentWorker is executing" do
|
||||||
insert_oban_job(%{
|
insert_oban_job(%{
|
||||||
state: "executing",
|
state: "executing",
|
||||||
worker: @asos_worker,
|
worker: @asos_worker,
|
||||||
|
|
@ -66,7 +67,36 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
|
||||||
|
|
||||||
assert status.state == :running
|
assert status.state == :running
|
||||||
assert status.label == "Updating propagation"
|
assert status.label == "Updating propagation"
|
||||||
assert status.detail =~ "ASOS"
|
assert [%{worker: :asos, label: asos_label}] = status.details
|
||||||
|
assert asos_label =~ "ASOS"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns :running with BOTH workers when grid + ASOS are executing at the same time" do
|
||||||
|
# Queue concurrency was bumped to 2 so the chain-step grid worker
|
||||||
|
# and the 10-minute ASOS nudge can run in parallel. The chip
|
||||||
|
# should show one sub-line per running worker, in a deterministic
|
||||||
|
# grid-before-asos order so the eye doesn't see them flicker.
|
||||||
|
insert_oban_job(%{
|
||||||
|
state: "executing",
|
||||||
|
worker: @grid_worker,
|
||||||
|
attempted_at: minutes_ago(5)
|
||||||
|
})
|
||||||
|
|
||||||
|
insert_oban_job(%{
|
||||||
|
state: "executing",
|
||||||
|
worker: @asos_worker,
|
||||||
|
attempted_at: minutes_ago(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
status = PipelineStatus.current()
|
||||||
|
|
||||||
|
assert status.state == :running
|
||||||
|
assert status.label == "Updating propagation"
|
||||||
|
assert [grid, asos] = status.details
|
||||||
|
assert grid.worker == :grid
|
||||||
|
assert grid.label =~ "HRRR"
|
||||||
|
assert asos.worker == :asos
|
||||||
|
assert asos.label =~ "ASOS"
|
||||||
end
|
end
|
||||||
|
|
||||||
test "returns :idle with Up to date label when last grid run completed recently" do
|
test "returns :idle with Up to date label when last grid run completed recently" do
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue