Process one forecast hour per PropagationGridWorker perform

A full f00-f18 sweep takes ~170 min of wall time, longer than the
~90 min gap between deploys on this cluster. Over the last 7 days
every run was killed mid-sweep and zero runs completed. The
propagation_scores table only ever held the scraps from partial
runs that landed before the pod died, which is why the map looked
like it was showing "the current hour" (or nothing).

The worker now processes exactly one forecast hour per perform/1
(~8-10 min) and enqueues the next hour as a fresh Oban job. A cron
fire with empty args seeds the chain at f00; subsequent runs carry
run_time + forecast_hour args. At f18 the chain stops and the
pruner runs. Each step broadcasts propagation:updated immediately
so new hours appear on the map as they land.

Wall time per perform drops from ~170 min to ~10 min, so Lifeline's
2h rescue window is no longer a factor and a pod restart loses at
most one forecast hour instead of the whole sweep. Retries and
max_attempts now describe a single fh, not the whole chain.

Also: bump the :propagation queue from 1 to 2 slots so
PropagationPruneWorker can run alongside the chain job, drop the
worker timeout/1 from 90 min to 20 min to match single-fh runs,
and pull Lifeline rescue_after from 120 min to 45 min to keep the
safety net above the step timeout.

Adds a "Data from HH:MM UTC · Nh ago/now/+Nh" indicator above the
pipeline chip in both the mobile and desktop sidebars so the user
can tell which valid_time the map is showing when the bottom
timeline isn't visible.

config/config.exs

@@ -63,14 +63,16 @@ config :microwaveprop, Oban,
     terrain: 4,
     commercial: 2,
     iemre: 10,
-    propagation: 1,
+    # 2 slots so PropagationPruneWorker can run alongside the
+    # forecast-hour chain job.
+    propagation: 2,
     admin: 1,
     nexrad: 2
   ],
   plugins: [
     {Oban.Plugins.Pruner, max_age: 3600 * 24},
-    # See runtime.exs — must exceed PropagationGridWorker's 90-min timeout.
-    {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 120)},
+    # See runtime.exs — must exceed the worker's timeout/1 callback.
+    {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 45)},
     {Oban.Plugins.Cron,
      crontab: [
        {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker},

config/dev.exs

@@ -72,7 +72,9 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
 
 config :microwaveprop, Oban,
   queues: [
-    propagation: 1,
+    # 2 slots so PropagationPruneWorker can run alongside the
+    # forecast-hour chain job.
+    propagation: 2,
     solar: 1,
     weather: 20,
     enqueue: 1,
@@ -84,8 +86,8 @@ config :microwaveprop, Oban,
   ],
   plugins: [
     {Oban.Plugins.Pruner, max_age: 3600 * 24},
-    # See runtime.exs — must exceed PropagationGridWorker's 90-min timeout.
-    {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 120)},
+    # See runtime.exs — must exceed the worker's timeout/1 callback.
+    {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 45)},
     {Oban.Plugins.Cron,
      crontab: [
        {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker},

config/runtime.exs

@@ -159,7 +159,9 @@ if config_env() == :prod do
     engine: Oban.Pro.Engines.Smart,
     # Per-pod concurrency (×3 pods = effective cluster total)
     queues: [
-      propagation: 1,
+      # 2 slots so PropagationPruneWorker can run alongside the
+      # forecast-hour chain job without blocking.
+      propagation: 2,
       commercial: 1,
       solar: 1,
       weather: 3,
@@ -199,14 +201,12 @@ if config_env() == :prod do
       {Oban.Plugins.Pruner, max_age: 3600 * 24},
       # Lifeline rescues jobs stuck in `executing` after this window.
       # MUST be larger than the longest `timeout/1` callback on any worker
-      # or Lifeline races the job's own deadline: PropagationGridWorker
-      # caps itself at 90 min and a real run is ~95 min of wall time, so
-      # 120 min gives enough headroom for the hourly sweep to finish
-      # before the safety net trips. Setting it shorter caused the grid
-      # job to be rescued mid-run and retried from f00, which discarded
-      # with empty errors after 3 rescues and left only 4-5 valid_times
-      # in propagation_scores.
-      {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 120)},
+      # or Lifeline races the job's own deadline. The chain-style
+      # PropagationGridWorker caps a single forecast-hour step at 20
+      # min and real steps run ~8-10 min, so 45 min gives comfortable
+      # headroom for a slow step without letting a truly stuck job
+      # linger for hours before the safety net trips.
+      {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 45)},
       {Oban.Plugins.Cron,
        crontab: [
          {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker},

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -1,16 +1,26 @@
 defmodule Microwaveprop.Workers.PropagationGridWorker do
   @moduledoc """
-  Hourly Oban worker that downloads HRRR data (analysis + forecast hours)
-  and computes propagation scores across the CONUS grid for all bands.
-  Fetches f00-f18 to provide 18-hour forecast timeline.
+  Oban worker that downloads HRRR data and computes propagation scores
+  across the CONUS grid for all bands, one forecast hour at a time.
+
+  Each `perform/1` processes a single forecast hour (~8–10 min of wall
+  time) and enqueues the next hour in the chain. A cron fire with
+  empty args seeds the chain at f00; subsequent runs carry
+  `forecast_hour` + `run_time` args. At f18 the chain stops and the
+  pruner cleans up old scores.
+
+  Splitting by forecast hour is a resilience play: a full sweep takes
+  ~3 hours of wall time, longer than the typical pod-restart interval
+  on this deployment. Under the old "one big perform" design, any
+  deploy mid-sweep killed the whole run, and max_attempts would
+  exhaust without recording an error. Per-hour jobs survive deploys
+  because Lifeline only needs to rescue a single 10-minute step, and
+  retries re-fetch just that forecast hour.
   """
 
-  # `unique.period` matches the 3-hour cron so a retrying job can't be
-  # duplicated by the next cron firing.
   use Oban.Worker,
     queue: :propagation,
-    max_attempts: 3,
-    unique: [period: 10_800, states: [:available, :scheduled, :executing, :retryable]]
+    max_attempts: 3
 
   alias Microwaveprop.Commercial
   alias Microwaveprop.Propagation
@@ -26,12 +36,11 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
 
   require Logger
 
-  # Hard ceiling per run. A healthy run is ~60 min (19 forecast hours × ~3 min
-  # each). 90 min gives 1.5× headroom and is under the 3-hour cron cadence, so
-  # a stuck wgrib2/HTTP call can't block the queue indefinitely. Oban kills
-  # the executing process on timeout, which closes linked ports and cascades
-  # SIGKILL to any child wgrib2 subprocess.
-  @run_timeout_ms 90 * 60 * 1000
+  # Hard ceiling for one forecast hour. A healthy step is ~8-10 min.
+  # 20 min gives 2× headroom for a slow HRRR fetch or scoring batch.
+  # Oban kills the executing process on timeout, which closes linked
+  # ports and cascades SIGKILL to any child wgrib2 subprocess.
+  @run_timeout_ms 20 * 60 * 1000
 
   @impl Oban.Worker
   def timeout(_job), do: @run_timeout_ms
@@ -39,48 +48,87 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   @max_forecast_hour 18
 
   @impl Oban.Worker
-  def perform(%Oban.Job{}) do
-    t_start = System.monotonic_time(:millisecond)
+  def perform(%Oban.Job{args: %{"forecast_hour" => fh, "run_time" => run_time_iso}}) do
+    {:ok, run_time, _} = DateTime.from_iso8601(run_time_iso)
+    run_chain_step(run_time, fh)
+  end
 
-    # HRRR takes ~45min to publish after the hour. Use 2 hours ago to ensure availability.
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    seed_chain()
+  end
+
+  @doc """
+  Enqueue the next chain step after a successful forecast-hour run.
+
+  Returns `{:ok, job}` when a new step is enqueued, or `:final` when
+  `fh` is already `@max_forecast_hour` so the chain has no more work.
+  Public so the chain entry point and tests can both exercise the
+  same enqueue path.
+  """
+  @spec enqueue_next_step(DateTime.t(), non_neg_integer()) :: {:ok, Oban.Job.t()} | :final
+  def enqueue_next_step(_run_time, fh) when fh >= @max_forecast_hour, do: :final
+
+  def enqueue_next_step(%DateTime{} = run_time, fh) when fh >= 0 do
+    {:ok, _job} =
+      %{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh + 1}
+      |> new()
+      |> Oban.insert()
+  end
+
+  defp seed_chain do
+    # HRRR takes ~45min to publish after the hour. Use 2 hours ago to
+    # ensure availability.
     two_hours_ago = DateTime.add(DateTime.utc_now(), -2, :hour)
     run_time = two_hours_ago |> HrrrClient.nearest_hrrr_hour() |> DateTime.truncate(:second)
-    points = Grid.conus_points()
 
-    Logger.info("PropagationGrid: run_time=#{run_time}, #{length(points)} points, f00-f#{@max_forecast_hour}")
+    Logger.info("PropagationGrid: seeding chain run_time=#{run_time}, f00-f#{@max_forecast_hour}")
 
-    for_result =
-      for fh <- 0..@max_forecast_hour do
-        valid_time = DateTime.add(run_time, fh * 3600, :second)
-        result = process_forecast_hour(points, run_time, fh, valid_time)
-        # Reclaim grid_data/profiles from this forecast hour before starting the next
-        :erlang.garbage_collect()
-
-        case result do
-          :ok -> valid_time
-          _ -> nil
-        end
-      end
-
-    valid_times = Enum.reject(for_result, &is_nil/1)
-
-    if valid_times != [] do
-      Phoenix.PubSub.broadcast(
-        Microwaveprop.PubSub,
-        "propagation:updated",
-        {:propagation_updated, valid_times}
-      )
-    end
-
-    total_ms = System.monotonic_time(:millisecond) - t_start
-    Logger.info("PropagationGrid: total time #{format_duration(total_ms)} (#{length(valid_times)} forecast hours)")
-
-    Weather.prune_old_grid_profiles()
-    Propagation.prune_old_scores()
+    {:ok, _job} =
+      %{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => 0}
+      |> new()
+      |> Oban.insert()
 
     :ok
   end
 
+  defp run_chain_step(run_time, fh) do
+    t_start = System.monotonic_time(:millisecond)
+    points = Grid.conus_points()
+    valid_time = DateTime.add(run_time, fh * 3600, :second)
+
+    Logger.info("PropagationGrid: chain step run_time=#{run_time} fh=#{fh} (#{length(points)} points)")
+
+    result = process_forecast_hour(points, run_time, fh, valid_time)
+    :erlang.garbage_collect()
+
+    case result do
+      :ok ->
+        Phoenix.PubSub.broadcast(
+          Microwaveprop.PubSub,
+          "propagation:updated",
+          {:propagation_updated, [valid_time]}
+        )
+
+        total_ms = System.monotonic_time(:millisecond) - t_start
+        Logger.info("PropagationGrid: fh=#{fh} step finished in #{format_duration(total_ms)}")
+
+        case enqueue_next_step(run_time, fh) do
+          :final ->
+            Weather.prune_old_grid_profiles()
+            Propagation.prune_old_scores()
+            Logger.info("PropagationGrid: chain complete for run_time=#{run_time}")
+
+          {:ok, _job} ->
+            :ok
+        end
+
+        :ok
+
+      other ->
+        other
+    end
+  end
+
   defp process_forecast_hour(points, run_time, forecast_hour, valid_time) do
     label = "f#{String.pad_leading(Integer.to_string(forecast_hour), 2, "0")}"
 

lib/microwaveprop_web/live/map_live.ex

@@ -342,6 +342,34 @@ defmodule MicrowavepropWeb.MapLive do
     Calendar.strftime(dt, "%H:%M")
   end
 
+  @doc false
+  # Render the small "Data from HH:MM UTC · Nh ago" indicator that sits
+  # above the pipeline status chip. Tells the user at a glance which
+  # valid_time the map is currently showing when the bottom timeline
+  # isn't visible.
+  @spec format_data_timestamp(DateTime.t() | nil) :: String.t()
+  def format_data_timestamp(nil), do: "No data"
+
+  def format_data_timestamp(%DateTime{} = dt) do
+    diff_minutes = DateTime.diff(dt, DateTime.utc_now(), :minute)
+    relative = format_relative_minutes(diff_minutes)
+    "Data from #{Calendar.strftime(dt, "%H:%M")} UTC · #{relative}"
+  end
+
+  defp format_relative_minutes(diff) when diff >= -30 and diff <= 30, do: "now"
+
+  defp format_relative_minutes(diff) when diff > 30 do
+    # Future
+    hours = div(diff + 30, 60)
+    "+#{hours}h"
+  end
+
+  defp format_relative_minutes(diff) do
+    # Past
+    hours = div(-diff + 30, 60)
+    "#{hours}h ago"
+  end
+
   defp closest_to_now([]), do: nil
 
   defp closest_to_now(times) do
@@ -492,6 +520,13 @@ defmodule MicrowavepropWeb.MapLive do
               </div>
             </div>
 
+            <div
+              id="data-timestamp-mobile"
+              class="text-[11px] opacity-70 px-1 leading-tight"
+            >
+              {format_data_timestamp(@selected_time)}
+            </div>
+
             <.pipeline_status_chip
               id="pipeline-status-mobile"
               status={@pipeline_status}
@@ -766,6 +801,13 @@ defmodule MicrowavepropWeb.MapLive do
           </ul>
 
           <div class="border-t border-base-300 pt-2">
+            <div
+              id="data-timestamp-desktop"
+              class="text-[11px] opacity-70 px-1 pb-1 leading-tight"
+            >
+              {format_data_timestamp(@selected_time)}
+            </div>
+
             <.pipeline_status_chip
               id="pipeline-status-desktop"
               status={@pipeline_status}

test/microwaveprop/workers/propagation_grid_worker_test.exs

@@ -0,0 +1,56 @@
+defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
+  @moduledoc """
+  Tests the chain-orchestration behavior of PropagationGridWorker.
+
+  The worker processes forecast hours f00–f18 across the CONUS grid,
+  but a single full sweep takes ~2 hours of wall time — longer than a
+  typical pod restart window. To survive deploys, the worker processes
+  ONE forecast hour per `perform/1` call and enqueues the next hour as
+  a fresh Oban job. The tests here cover the dispatch + chain logic
+  without mocking the full HRRR / scoring stack.
+  """
+  use Microwaveprop.DataCase, async: false
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  alias Microwaveprop.Workers.PropagationGridWorker
+
+  describe "perform/1 — chain seeding (empty args)" do
+    test "enqueues a single fh=0 chain step with a normalized run_time" do
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}})
+
+        [child] = all_enqueued(worker: PropagationGridWorker)
+        assert child.args["forecast_hour"] == 0
+        assert is_binary(child.args["run_time"])
+
+        {:ok, run_time, _} = DateTime.from_iso8601(child.args["run_time"])
+        assert run_time.minute == 0
+        assert run_time.second == 0
+        assert DateTime.before?(run_time, DateTime.utc_now())
+      end)
+    end
+  end
+
+  describe "enqueue_next_step/2" do
+    test "enqueues fh+1 when fh < max" do
+      run_time = ~U[2026-04-14 16:00:00Z]
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert {:ok, _} = PropagationGridWorker.enqueue_next_step(run_time, 0)
+
+        [child] = all_enqueued(worker: PropagationGridWorker)
+        assert child.args["forecast_hour"] == 1
+        assert child.args["run_time"] == DateTime.to_iso8601(run_time)
+      end)
+    end
+
+    test "does not enqueue anything after the final forecast hour" do
+      run_time = ~U[2026-04-14 16:00:00Z]
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert :final = PropagationGridWorker.enqueue_next_step(run_time, 18)
+        assert [] = all_enqueued(worker: PropagationGridWorker)
+      end)
+    end
+  end
+end

test/microwaveprop_web/live/map_live_test.exs

@@ -98,6 +98,41 @@ defmodule MicrowavepropWeb.MapLiveTest do
     end
   end
 
+  describe "data timestamp indicator" do
+    test "renders a 'Data from …' line on both mobile and desktop sidebars", %{conn: conn} do
+      {:ok, _lv, html} = live(conn, ~p"/map")
+      assert html =~ ~s(id="data-timestamp-mobile")
+      assert html =~ ~s(id="data-timestamp-desktop")
+      # Clean test DB has no scores; the indicator falls back to "No data".
+      assert html =~ "No data"
+    end
+  end
+
+  describe "format_data_timestamp/1" do
+    alias MicrowavepropWeb.MapLive
+
+    test "returns 'No data' for nil" do
+      assert MapLive.format_data_timestamp(nil) == "No data"
+    end
+
+    test "renders the valid_time and a 'now' suffix when within ±30 minutes" do
+      now = DateTime.utc_now()
+      label = MapLive.format_data_timestamp(now)
+      assert label =~ "UTC"
+      assert label =~ "now"
+    end
+
+    test "renders a '+Nh' suffix when the valid_time is in the future" do
+      future = DateTime.add(DateTime.utc_now(), 3 * 3600, :second)
+      assert MapLive.format_data_timestamp(future) =~ "+3h"
+    end
+
+    test "renders a 'Nh ago' suffix when the valid_time is in the past" do
+      past = DateTime.add(DateTime.utc_now(), -4 * 3600, :second)
+      assert MapLive.format_data_timestamp(past) =~ "4h ago"
+    end
+  end
+
   describe "pipeline status chip" do
     test "renders the aggregated pipeline status chip on mount", %{conn: conn} do
       {:ok, _lv, html} = live(conn, ~p"/map")