Fix ERA5 cap math + stuck threshold, split chip label onto two lines

Era5SubmitWorker was counting era5_cds_jobs rows 1:1 against the CDS
150-job ceiling, but each row holds TWO CDS job IDs (single-level +
pressure-level). Prod hit 74 rows = 148 in-flight jobs and got stuck
in a reject→resubmit spiral. The cap guard now multiplies row count
by 2 to match what CDS sees.

Era5PollWorker @stuck_after_seconds goes from 4h to 18h. CDS
single-level routinely sits at 'accepted' for 12+h under load while
pressure finishes in ~90 min; the 4h threshold was firing on normal
slow runs and feeding the same spiral.

PipelineStatus now returns `label` + `detail` separately so the map
chip can render "Updating propagation" on one line and the
sub-detail ("HRRR run" / "ASOS nudge" / "+3h" / "now") on a second
line below it. running_label/1 is renamed to running_detail/1 and
returns just the sub-part.

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
+  @spec running_detail(map() | nil) :: String.t() | nil
-  def running_label(%{forecast_hour: 0}), do: "Updating propagation · now"
+  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_label(%{forecast_hour: fh}) when is_integer(fh) and fh > 0, do: "Updating propagation · +#{fh}h"
+  def running_detail(_), do: nil
-
-  def running_label(_), 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_detail(@grid_worker), do: "HRRR run"
-  defp base_running_label(@asos_worker), do: "Updating propagation (ASOS nudge)"
+  defp base_running_detail(@asos_worker), do: "ASOS nudge"
-  defp base_running_label(_), do: "Updating propagation"
+  defp base_running_detail(_), do: nil
 
   defp format_age(0), do: "just now"
   defp format_age(1), do: "1m"

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
+  # 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
+  # prod: pressure-level completes in ~90 min, but single-level
-  # 'accepted' for 16+ hours.) Once a row ages past this threshold we
+  # routinely sits at 'accepted' for 12+ hours under CDS load. The
-  # abandon the in-flight jobs, delete them from CDS, and re-submit the
+  # previous 4h threshold was firing on normal slow runs, forcing a
-  # tile-month. 4 hours is ~8× the normal completion time — well above
+  # resubmit which then got rejected by the per-user cap — an infinite
-  # normal queue variance but tight enough that stuck jobs self-heal
+  # reject→resubmit spiral that burned CDS quota without producing any
-  # within a working session.
+  # decoded data. 18h is comfortably above the worst observed queue
-  @stuck_after_seconds 4 * 3600
+  # 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

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 +
+  # tile-month submit lives in ONE era5_cds_jobs row but carries TWO CDS
-  # pressure-level), so we stop submitting once the in-flight count is
+  # job IDs (single-level + pressure-level), so the real in-flight count
-  # within `@cap_headroom` of the ceiling. Snoozing (instead of failing
+  # CDS sees is 2 × row_count. The cap guard multiplies accordingly.
-  # or sleeping) releases the Oban slot immediately and lets the poll
+  # Snoozing (instead of failing or sleeping) releases the Oban slot
-  # workers drain in-flight jobs before we push more in.
+  # 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)
+  # Headroom is deliberately generous (30 job slots, effective ceiling
-  # because the cap check races between concurrent workers: two workers
+  # 120 jobs = 60 rows) because the cap check races between concurrent
-  # that both observe count=119 will both submit, landing us at 121 after
+  # workers: two workers that both observe row_count=59 can both submit,
-  # both finish. With 30 slots of headroom the cluster would have to race
+  # landing us at row_count=61 (122 jobs) after both finish. With 30
-  # 15 workers wide to clip CDS's hard 150 ceiling. Observed:
+  # slots of headroom the cluster would have to race 15 workers wide to
-  # single-worker submits at count=141 → CDS reject storms, so we err on
+  # clip CDS's hard 150 ceiling. Observed in prod: the old 1× math was
-  # the side of staying well clear.
+  # 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

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"""
     <div
       id={@id}
       data-pipeline-state={@status.state}
       class="flex items-start gap-1.5 text-xs px-1 py-1 leading-tight"
-      title={@display_label}
+      title={@display_title}
     >
       <%= 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"
           />
-          <span class="opacity-90 break-words min-w-0">{@display_label}</span>
+          <span class="opacity-90 break-words min-w-0">
+            {@display_main}
+            <%= if @display_detail do %>
+              <span class="block text-[11px] opacity-70">{@display_detail}</span>
+            <% end %>
+          </span>
         <% :idle -> %>
           <span class="size-2 rounded-full bg-success inline-block shrink-0 mt-[5px]"></span>
-          <span class="opacity-70 break-words min-w-0">{@display_label}</span>
+          <span class="opacity-70 break-words min-w-0">{@display_main}</span>
         <% :stale -> %>
           <span class="size-2 rounded-full bg-warning inline-block shrink-0 mt-[5px]"></span>
-          <span class="opacity-80 break-words min-w-0">{@display_label}</span>
+          <span class="opacity-80 break-words min-w-0">{@display_main}</span>
         <% :unknown -> %>
           <span class="size-2 rounded-full bg-base-content/30 inline-block shrink-0 mt-[5px]"></span>
-          <span class="opacity-60 break-words min-w-0">{@display_label}</span>
+          <span class="opacity-60 break-words min-w-0">{@display_main}</span>
       <% end %>
     </div>
     """
   end
 
-  defp chip_display_label(%{state: :running} = status, progress) do
+  defp chip_display_parts(%{state: :running} = status, progress) do
-    PipelineStatus.running_label(progress) || status.label
+    {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

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}) ==
+      assert PipelineStatus.running_detail(%{forecast_hour: 0}) == "now"
-               "Updating propagation · now"
     end
 
     test "positive forecast hours render as +Nh" do
-      assert PipelineStatus.running_label(%{forecast_hour: 1}) ==
+      assert PipelineStatus.running_detail(%{forecast_hour: 1}) == "+1h"
-               "Updating propagation · +1h"
+      assert PipelineStatus.running_detail(%{forecast_hour: 18}) == "+18h"
-
-      assert PipelineStatus.running_label(%{forecast_hour: 18}) ==
-               "Updating propagation · +18h"
     end
 
     test "returns nil when no forecast hour is provided" do
-      assert PipelineStatus.running_label(nil) == nil
+      assert PipelineStatus.running_detail(nil) == nil
-      assert PipelineStatus.running_label(%{}) == nil
+      assert PipelineStatus.running_detail(%{}) == nil
     end
   end
 end

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 =

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
+    # a user has too many pending jobs. Each tile-month lives in ONE
-    # era5_cds_jobs (single-level + pressure-level), so we snooze when
+    # era5_cds_jobs row but carries TWO CDS job IDs (single-level +
-    # the in-flight count is close to the ceiling. The exact threshold
+    # pressure-level), so the real in-flight count is 2 × row_count. The
-    # (headroom) is tuned in the worker; the test just seeds well above
+    # cap guard multiplies accordingly. The exact threshold (headroom) is
-    # the effective threshold (140 rows) so whatever the worker picks,
+    # tuned in the worker; the tests below exercise both the obvious
-    # it should trip the snooze.
+    # "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 =