diff --git a/lib/microwaveprop/propagation/pipeline_status.ex b/lib/microwaveprop/propagation/pipeline_status.ex index cffe5751..7fe18da5 100644 --- a/lib/microwaveprop/propagation/pipeline_status.ex +++ b/lib/microwaveprop/propagation/pipeline_status.ex @@ -33,22 +33,22 @@ defmodule Microwaveprop.Propagation.PipelineStatus do @type t :: %{ state: state(), label: String.t(), + detail: String.t() | nil, last_update_at: DateTime.t() | nil } @doc """ Format a forecast-progress payload (as broadcast by - `PropagationGridWorker`) into a short chip label for the map. + `PropagationGridWorker`) into a short detail line for the running + chip ("now", "+3h", …). Returns `nil` when the payload has no `forecast_hour` so callers can - fall back to the base running label from `current/0`. + fall back to the base detail (e.g. "HRRR run") from `current/0`. """ - @spec running_label(map() | nil) :: String.t() | nil - def running_label(%{forecast_hour: 0}), do: "Updating propagation · now" - - def running_label(%{forecast_hour: fh}) when is_integer(fh) and fh > 0, do: "Updating propagation · +#{fh}h" - - def running_label(_), do: nil + @spec running_detail(map() | nil) :: String.t() | nil + def running_detail(%{forecast_hour: 0}), do: "now" + def running_detail(%{forecast_hour: fh}) when is_integer(fh) and fh > 0, do: "+#{fh}h" + def running_detail(_), do: nil @spec current() :: t() def current do @@ -56,7 +56,8 @@ defmodule Microwaveprop.Propagation.PipelineStatus do worker when is_binary(worker) -> %{ state: :running, - label: base_running_label(worker), + label: "Updating propagation", + detail: base_running_detail(worker), last_update_at: latest_completed_at() } @@ -90,7 +91,7 @@ defmodule Microwaveprop.Propagation.PipelineStatus do end defp build_idle_or_stale(nil) do - %{state: :unknown, label: "Propagation status unknown", last_update_at: nil} + %{state: :unknown, label: "Propagation status unknown", detail: nil, last_update_at: nil} end defp build_idle_or_stale(%DateTime{} = last) do @@ -100,20 +101,22 @@ defmodule Microwaveprop.Propagation.PipelineStatus do %{ state: :stale, label: "Propagation data stale · last update #{format_age(age_minutes)} ago", + detail: nil, last_update_at: last } else %{ state: :idle, label: "Up to date · #{format_age(age_minutes)} ago", + detail: nil, last_update_at: last } end end - defp base_running_label(@grid_worker), do: "Updating propagation (HRRR run)" - defp base_running_label(@asos_worker), do: "Updating propagation (ASOS nudge)" - defp base_running_label(_), do: "Updating propagation" + 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(1), do: "1m" diff --git a/lib/microwaveprop/workers/era5_poll_worker.ex b/lib/microwaveprop/workers/era5_poll_worker.ex index 5edf6c17..fcca9246 100644 --- a/lib/microwaveprop/workers/era5_poll_worker.ex +++ b/lib/microwaveprop/workers/era5_poll_worker.ex @@ -33,14 +33,16 @@ defmodule Microwaveprop.Workers.Era5PollWorker do @snooze_seconds 5 * 60 # A CDS job that's been in 'accepted' state for more than this is stuck - # in the server's queue and almost certainly won't run. (Observed in - # prod: pressure-level completes in ~90 min while single-level sits at - # 'accepted' for 16+ hours.) Once a row ages past this threshold we - # abandon the in-flight jobs, delete them from CDS, and re-submit the - # tile-month. 4 hours is ~8× the normal completion time — well above - # normal queue variance but tight enough that stuck jobs self-heal - # within a working session. - @stuck_after_seconds 4 * 3600 + # in the server's queue and almost certainly won't run. Observed in + # prod: pressure-level completes in ~90 min, but single-level + # routinely sits at 'accepted' for 12+ hours under CDS load. The + # previous 4h threshold was firing on normal slow runs, forcing a + # resubmit which then got rejected by the per-user cap — an infinite + # reject→resubmit spiral that burned CDS quota without producing any + # decoded data. 18h is comfortably above the worst observed queue + # times while still self-healing within a day. + @stuck_after_seconds 18 * 3600 + @stuck_after_hours div(@stuck_after_seconds, 3600) @impl Oban.Worker def perform(%Oban.Job{args: %{"era5_cds_job_id" => id}}) do @@ -64,7 +66,7 @@ defmodule Microwaveprop.Workers.Era5PollWorker do _other -> if stuck?(row) do - resubmit_vanished(row, "both", "stuck in CDS queue > 4h") + resubmit_vanished(row, "both", "stuck in CDS queue > #{@stuck_after_hours}h") else handle_non_both_done(row, single, pressure) end diff --git a/lib/microwaveprop/workers/era5_submit_worker.ex b/lib/microwaveprop/workers/era5_submit_worker.ex index a79ac12d..0e75e720 100644 --- a/lib/microwaveprop/workers/era5_submit_worker.ex +++ b/lib/microwaveprop/workers/era5_submit_worker.ex @@ -39,19 +39,20 @@ defmodule Microwaveprop.Workers.Era5SubmitWorker do @poll_delay_seconds 5 * 60 # CDS enforces a server-side per-user cap of 150 in-flight jobs. Each - # successful submit adds 2 rows to era5_cds_jobs (single-level + - # pressure-level), so we stop submitting once the in-flight count is - # within `@cap_headroom` of the ceiling. Snoozing (instead of failing - # or sleeping) releases the Oban slot immediately and lets the poll - # workers drain in-flight jobs before we push more in. + # tile-month submit lives in ONE era5_cds_jobs row but carries TWO CDS + # job IDs (single-level + pressure-level), so the real in-flight count + # CDS sees is 2 × row_count. The cap guard multiplies accordingly. + # Snoozing (instead of failing or sleeping) releases the Oban slot + # immediately and lets the poll workers drain in-flight jobs before we + # push more in. # - # Headroom is deliberately generous (30 slots, ceiling of 120 effective) - # because the cap check races between concurrent workers: two workers - # that both observe count=119 will both submit, landing us at 121 after - # both finish. With 30 slots of headroom the cluster would have to race - # 15 workers wide to clip CDS's hard 150 ceiling. Observed: - # single-worker submits at count=141 → CDS reject storms, so we err on - # the side of staying well clear. + # Headroom is deliberately generous (30 job slots, effective ceiling + # 120 jobs = 60 rows) because the cap check races between concurrent + # workers: two workers that both observe row_count=59 can both submit, + # landing us at row_count=61 (122 jobs) after both finish. With 30 + # slots of headroom the cluster would have to race 15 workers wide to + # clip CDS's hard 150 ceiling. Observed in prod: the old 1× math was + # allowing 74 rows (148 jobs) which triggered CDS reject storms. @cds_ceiling 150 @cap_headroom 30 @snooze_seconds 5 * 60 @@ -84,10 +85,11 @@ defmodule Microwaveprop.Workers.Era5SubmitWorker do {:ok, :already_submitted} cds_queue_full?() -> - in_flight = Repo.aggregate(Era5CdsJob, :count, :id) + row_count = Repo.aggregate(Era5CdsJob, :count, :id) + in_flight_jobs = row_count * 2 Logger.info( - "Era5Submit: CDS in-flight cap reached (#{in_flight}/#{@cds_ceiling}) — snoozing #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}" + "Era5Submit: CDS in-flight cap reached (#{in_flight_jobs}/#{@cds_ceiling} jobs across #{row_count} rows) — snoozing #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}" ) {:snooze, @snooze_seconds} @@ -98,7 +100,8 @@ defmodule Microwaveprop.Workers.Era5SubmitWorker do end defp cds_queue_full? do - Repo.aggregate(Era5CdsJob, :count, :id) + 2 > @cds_ceiling - @cap_headroom + in_flight_jobs_after_submit = Repo.aggregate(Era5CdsJob, :count, :id) * 2 + 2 + in_flight_jobs_after_submit > @cds_ceiling - @cap_headroom end defp submit_and_persist(year, month, tile_lat, tile_lon) do diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 20b8ce62..7580633a 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -384,14 +384,20 @@ defmodule MicrowavepropWeb.MapLive do attr :progress, :any, default: nil defp pipeline_status_chip(assigns) do - assigns = assign(assigns, :display_label, chip_display_label(assigns.status, assigns.progress)) + {main, detail} = chip_display_parts(assigns.status, assigns.progress) + + assigns = + assigns + |> assign(:display_main, main) + |> assign(:display_detail, detail) + |> assign(:display_title, if(detail, do: "#{main} — #{detail}", else: main)) ~H"""
<%= case @status.state do %> <% :running -> %> @@ -399,26 +405,31 @@ defmodule MicrowavepropWeb.MapLive do name="hero-arrow-path" class="size-3.5 shrink-0 mt-0.5 text-info motion-safe:animate-spin" /> - {@display_label} + + {@display_main} + <%= if @display_detail do %> + {@display_detail} + <% end %> + <% :idle -> %> - {@display_label} + {@display_main} <% :stale -> %> - {@display_label} + {@display_main} <% :unknown -> %> - {@display_label} + {@display_main} <% end %>
""" end - defp chip_display_label(%{state: :running} = status, progress) do - PipelineStatus.running_label(progress) || status.label + defp chip_display_parts(%{state: :running} = status, progress) do + {status.label, PipelineStatus.running_detail(progress) || status.detail} end - defp chip_display_label(status, _progress), do: status.label + defp chip_display_parts(status, _progress), do: {status.label, nil} @impl true def render(assigns) do diff --git a/test/microwaveprop/propagation/pipeline_status_test.exs b/test/microwaveprop/propagation/pipeline_status_test.exs index 29b9979c..7a0ee22a 100644 --- a/test/microwaveprop/propagation/pipeline_status_test.exs +++ b/test/microwaveprop/propagation/pipeline_status_test.exs @@ -41,7 +41,7 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do assert is_binary(status.label) end - test "returns :running with grid worker label when PropagationGridWorker is executing" do + test "returns :running with grid worker detail when PropagationGridWorker is executing" do insert_oban_job(%{ state: "executing", worker: @grid_worker, @@ -51,10 +51,11 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do status = PipelineStatus.current() assert status.state == :running - assert status.label =~ "HRRR" + assert status.label == "Updating propagation" + assert status.detail =~ "HRRR" end - test "returns :running with ASOS label when AsosAdjustmentWorker is executing" do + test "returns :running with ASOS detail when AsosAdjustmentWorker is executing" do insert_oban_job(%{ state: "executing", worker: @asos_worker, @@ -64,7 +65,8 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do status = PipelineStatus.current() assert status.state == :running - assert status.label =~ "ASOS" + assert status.label == "Updating propagation" + assert status.detail =~ "ASOS" end test "returns :idle with Up to date label when last grid run completed recently" do @@ -141,23 +143,19 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do end end - describe "running_label/1" do + describe "running_detail/1" do test "forecast hour 0 renders as 'now'" do - assert PipelineStatus.running_label(%{forecast_hour: 0}) == - "Updating propagation · now" + assert PipelineStatus.running_detail(%{forecast_hour: 0}) == "now" end test "positive forecast hours render as +Nh" do - assert PipelineStatus.running_label(%{forecast_hour: 1}) == - "Updating propagation · +1h" - - assert PipelineStatus.running_label(%{forecast_hour: 18}) == - "Updating propagation · +18h" + assert PipelineStatus.running_detail(%{forecast_hour: 1}) == "+1h" + assert PipelineStatus.running_detail(%{forecast_hour: 18}) == "+18h" end test "returns nil when no forecast hour is provided" do - assert PipelineStatus.running_label(nil) == nil - assert PipelineStatus.running_label(%{}) == nil + assert PipelineStatus.running_detail(nil) == nil + assert PipelineStatus.running_detail(%{}) == nil end end end diff --git a/test/microwaveprop/workers/era5_poll_worker_test.exs b/test/microwaveprop/workers/era5_poll_worker_test.exs index e243859e..716211b3 100644 --- a/test/microwaveprop/workers/era5_poll_worker_test.exs +++ b/test/microwaveprop/workers/era5_poll_worker_test.exs @@ -14,7 +14,7 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do end defp insert_cds_job(overrides \\ %{}) do - # Default to "just now" so the stuck-after-4h guard doesn't misfire + # Default to "just now" so the stuck-after guard doesn't misfire # and re-submit in tests that want to exercise status-specific # branches. Tests that care about the stuck path pass a stale # submitted_at explicitly. @@ -166,10 +166,15 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do # on era5_cds_jobs, so once the row ages past @stuck_after, poll # treats it as terminal and re-submits — same path as :not_found / # :rejected. - test "re-submits when submitted_at > 4h ago and at least one leg is still running" do + # + # Threshold is 18h (see @stuck_after_seconds in Era5PollWorker): CDS + # single-level routinely takes 12+ hours under load while pressure + # finishes in ~90 minutes, and the 4h threshold was firing on normal + # slow runs, causing re-submit storms that hit the per-user cap. + test "re-submits when submitted_at is older than 18h and at least one leg is still running" do stale_submit = DateTime.utc_now() - |> DateTime.add(-5 * 3600, :second) + |> DateTime.add(-19 * 3600, :second) |> DateTime.truncate(:second) row = insert_cds_job(%{submitted_at: stale_submit}) @@ -201,6 +206,27 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do ) end + test "does NOT re-submit a 10h-old row that's still running (under the 18h threshold)" do + # Observed in prod: CDS single-level can sit at 'accepted' for 12+ + # hours while pressure completes in ~90 min. Anything under 18h + # should keep snoozing, not trigger a re-submit storm. + slowish_submit = + DateTime.utc_now() + |> DateTime.add(-10 * 3600, :second) + |> DateTime.truncate(:second) + + row = insert_cds_job(%{submitted_at: slowish_submit}) + + Req.Test.stub(Era5Client, fn conn -> + Req.Test.json(conn, %{"status" => "running"}) + end) + + assert {:snooze, _} = + Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}}) + + assert Repo.get(Era5CdsJob, row.id) + end + test "does NOT re-submit when submitted_at is recent even if running" do # A 30-minute-old row with both legs running is normal — snooze. fresh_submit = diff --git a/test/microwaveprop/workers/era5_submit_worker_test.exs b/test/microwaveprop/workers/era5_submit_worker_test.exs index ec997e61..d3cdc214 100644 --- a/test/microwaveprop/workers/era5_submit_worker_test.exs +++ b/test/microwaveprop/workers/era5_submit_worker_test.exs @@ -126,12 +126,13 @@ defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do describe "CDS in-flight cap" do # CDS rejects with "Number of queued requests is limited to 150" when - # a user has too many pending jobs. Each submit adds 2 rows to - # era5_cds_jobs (single-level + pressure-level), so we snooze when - # the in-flight count is close to the ceiling. The exact threshold - # (headroom) is tuned in the worker; the test just seeds well above - # the effective threshold (140 rows) so whatever the worker picks, - # it should trip the snooze. + # a user has too many pending jobs. Each tile-month lives in ONE + # era5_cds_jobs row but carries TWO CDS job IDs (single-level + + # pressure-level), so the real in-flight count is 2 × row_count. The + # cap guard multiplies accordingly. The exact threshold (headroom) is + # tuned in the worker; the tests below exercise both the obvious + # "way above the cap" path and a row count that should only trip the + # snooze if the 2× math is in place. @cds_ceiling 150 test "snoozes when era5_cds_jobs count is at or above the snooze threshold" do @@ -171,6 +172,43 @@ defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do ) end + test "snoozes when row count × 2 crosses the cap (each row = two CDS jobs)" do + # 65 rows = 130 in-flight CDS jobs — above the effective snooze + # threshold even though row count alone is nowhere near 150. This + # is the scenario that was escaping the old guard and driving the + # per-user-cap reject storms in prod. + rows = + for i <- 1..65 do + %{ + id: Ecto.UUID.generate(), + year: 2014, + month: rem(i - 1, 12) + 1, + tile_lat: 32, + tile_lon: -98 - i, + single_job_id: "x2-single-#{i}", + pressure_job_id: "x2-pressure-#{i}", + submitted_at: DateTime.truncate(DateTime.utc_now(), :second), + inserted_at: DateTime.truncate(DateTime.utc_now(), :second), + updated_at: DateTime.truncate(DateTime.utc_now(), :second), + poll_count: 0 + } + end + + Repo.insert_all(Era5CdsJob, rows) + + Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted when capped" end) + + Oban.Testing.with_testing_mode(:manual, fn -> + assert {:snooze, seconds} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args}) + assert is_integer(seconds) and seconds > 0 + end) + + refute Repo.one( + from j in Era5CdsJob, + where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98 + ) + end + test "submits normally when in-flight count is well below the ceiling" do # 10 in-flight — comfortably under the cap. rows =