feat(propagation): record per-forecast-hour chain step timings

Add a propagation_run_timings table so the wall-clock duration of each
(run_time, forecast_hour) step is queryable long after the run is over.
Keyed by (run_time, forecast_hour) with a status column that captures
whether the step succeeded or bailed out, and an error string on
failure.

PropagationGridWorker stamps every step (ok and failed) via
Propagation.record_run_timing/1. Timing inserts are wrapped in rescue +
changeset-error handling so the instrumentation can never brick the
chain.

lib/microwaveprop/propagation.ex

@@ -1,12 +1,16 @@
 defmodule Microwaveprop.Propagation do
   @moduledoc false
 
+  import Ecto.Query
+
   alias Microwaveprop.Propagation.BandConfig
   alias Microwaveprop.Propagation.Grid
   alias Microwaveprop.Propagation.ProfilesFile
+  alias Microwaveprop.Propagation.RunTiming
   alias Microwaveprop.Propagation.ScoreCache
   alias Microwaveprop.Propagation.Scorer
   alias Microwaveprop.Propagation.ScoresFile
+  alias Microwaveprop.Repo
   alias Microwaveprop.Weather.SoundingParams
 
   require Logger
@@ -447,6 +451,38 @@ defmodule Microwaveprop.Propagation do
     end
   end
 
+  ## Run timings
+
+  @doc """
+  Record wall-clock duration for a single forecast-hour chain step.
+
+  Called by `PropagationGridWorker` at the end of every step (success or
+  failure) so the timing history survives pod restarts and can be
+  inspected later to see which steps are slow or flaky.
+  """
+  @spec record_run_timing(map()) ::
+          {:ok, RunTiming.t()} | {:error, Ecto.Changeset.t()}
+  def record_run_timing(attrs) do
+    %RunTiming{}
+    |> RunTiming.changeset(attrs)
+    |> Repo.insert()
+  end
+
+  @doc """
+  List the most-recently-started run-timing rows, newest first.
+
+  Defaults to 100 rows; pass `:limit` to override.
+  """
+  @spec list_recent_run_timings(keyword()) :: [RunTiming.t()]
+  def list_recent_run_timings(opts \\ []) do
+    limit = Keyword.get(opts, :limit, 100)
+
+    RunTiming
+    |> order_by(desc: :started_at)
+    |> limit(^limit)
+    |> Repo.all()
+  end
+
   # Prefer the persisted scalar — `hrrr_profiles` already stored this at
   # ingestion time and AsosAdjustmentWorker loads 92k rows per tick without
   # the JSONB `profile` column to avoid a Jason.decode! storm on the DB pool.

lib/microwaveprop/propagation/run_timing.ex

@@ -0,0 +1,39 @@
+defmodule Microwaveprop.Propagation.RunTiming do
+  @moduledoc """
+  One row per forecast hour of a `PropagationGridWorker` chain, recording
+  how long that hour's fetch + score + persist cycle took.
+
+  Rows are keyed by `(run_time, forecast_hour)`. `run_time` is the HRRR
+  model cycle and `forecast_hour` ∈ 0..18.
+  """
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+  schema "propagation_run_timings" do
+    field :run_time, :utc_datetime
+    field :forecast_hour, :integer
+    field :valid_time, :utc_datetime
+    field :started_at, :utc_datetime_usec
+    field :finished_at, :utc_datetime_usec
+    field :duration_ms, :integer
+    field :status, Ecto.Enum, values: [:ok, :failed]
+    field :error, :string
+
+    timestamps(type: :utc_datetime, updated_at: false)
+  end
+
+  @required ~w(run_time forecast_hour valid_time started_at finished_at duration_ms status)a
+  @optional ~w(error)a
+
+  def changeset(row, attrs) do
+    row
+    |> cast(attrs, @required ++ @optional)
+    |> validate_required(@required)
+    |> validate_number(:forecast_hour, greater_than_or_equal_to: 0, less_than_or_equal_to: 18)
+    |> validate_number(:duration_ms, greater_than_or_equal_to: 0)
+    |> unique_constraint([:run_time, :forecast_hour])
+  end
+end

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -107,6 +107,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
 
   defp run_chain_step(run_time, fh) do
     t_start = System.monotonic_time(:millisecond)
+    started_at = DateTime.utc_now()
     points = Grid.conus_points()
     valid_time = DateTime.add(run_time, fh * 3600, :second)
 
@@ -115,6 +116,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
     result = process_forecast_hour(points, run_time, fh, valid_time)
     :erlang.garbage_collect()
 
+    total_ms = System.monotonic_time(:millisecond) - t_start
+    record_timing(run_time, fh, valid_time, started_at, total_ms, result)
+
     case result do
       :ok ->
         Phoenix.PubSub.broadcast(
@@ -123,7 +127,6 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
           {:propagation_updated, [valid_time]}
         )
 
-        total_ms = System.monotonic_time(:millisecond) - t_start
         Logger.info("PropagationGrid: fh=#{fh} step finished in #{format_duration(total_ms)}")
 
         run_time
@@ -137,6 +140,38 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
     end
   end
 
+  # Persist one row to propagation_run_timings. Swallow DB failures so
+  # the instrumentation can never brick the chain — if the table is
+  # missing or the connection is flaky, we log and carry on.
+  defp record_timing(run_time, fh, valid_time, started_at, duration_ms, result) do
+    {status, error} =
+      case result do
+        :ok -> {:ok, nil}
+        other -> {:failed, inspect(other)}
+      end
+
+    attrs = %{
+      run_time: DateTime.truncate(run_time, :second),
+      forecast_hour: fh,
+      valid_time: DateTime.truncate(valid_time, :second),
+      started_at: started_at,
+      finished_at: DateTime.add(started_at, duration_ms, :millisecond),
+      duration_ms: duration_ms,
+      status: status,
+      error: error
+    }
+
+    case Propagation.record_run_timing(attrs) do
+      {:ok, _} ->
+        :ok
+
+      {:error, changeset} ->
+        Logger.warning("PropagationGrid: timing insert failed: #{inspect(changeset.errors)}")
+    end
+  rescue
+    e -> Logger.warning("PropagationGrid: timing insert raised: #{inspect(e)}")
+  end
+
   # `enqueue_next_step/2` returns `:final` at fh=18 (the whole chain
   # is done) or `{:ok, job}` when the next hour has been scheduled.
   # Split into its own function to keep `run_chain_step/2` under

priv/repo/migrations/20260418143251_create_propagation_run_timings.exs

@@ -0,0 +1,32 @@
+defmodule Microwaveprop.Repo.Migrations.CreatePropagationRunTimings do
+  use Ecto.Migration
+
+  def change do
+    create table(:propagation_run_timings, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+
+      # `run_time` is the HRRR model cycle the chain is rolling through,
+      # `forecast_hour` ∈ 0..18 picks the step within that chain.
+      # (run_time, forecast_hour) is the natural key.
+      add :run_time, :utc_datetime, null: false
+      add :forecast_hour, :integer, null: false
+      add :valid_time, :utc_datetime, null: false
+
+      # Wall-clock window of this forecast hour's work, plus the total
+      # in ms so queries don't have to recompute it.
+      add :started_at, :utc_datetime_usec, null: false
+      add :finished_at, :utc_datetime_usec, null: false
+      add :duration_ms, :integer, null: false
+
+      # "ok" when replace_scores returned {:ok, _}, "failed" otherwise.
+      # `error` carries the inspected reason on failure, nil on success.
+      add :status, :string, null: false
+      add :error, :text
+
+      timestamps(type: :utc_datetime, updated_at: false)
+    end
+
+    create unique_index(:propagation_run_timings, [:run_time, :forecast_hour])
+    create index(:propagation_run_timings, [:started_at])
+  end
+end

test/microwaveprop/propagation_test.exs

@@ -579,4 +579,97 @@ defmodule Microwaveprop.PropagationTest do
       assert Propagation.latest_valid_time() == t2
     end
   end
+
+  describe "record_run_timing/1" do
+    test "persists a successful forecast-hour timing" do
+      run_time = ~U[2026-07-15 12:00:00Z]
+      started = ~U[2026-07-15 12:00:05Z]
+      finished = ~U[2026-07-15 12:09:20Z]
+
+      assert {:ok, row} =
+               Propagation.record_run_timing(%{
+                 run_time: run_time,
+                 forecast_hour: 3,
+                 valid_time: DateTime.add(run_time, 3, :hour),
+                 started_at: started,
+                 finished_at: finished,
+                 duration_ms: 555_000,
+                 status: :ok
+               })
+
+      assert row.run_time == run_time
+      assert row.forecast_hour == 3
+      assert row.valid_time == ~U[2026-07-15 15:00:00Z]
+      assert row.duration_ms == 555_000
+      assert row.status == :ok
+      assert is_nil(row.error)
+    end
+
+    test "persists a failure with an error message" do
+      run_time = ~U[2026-07-15 12:00:00Z]
+
+      assert {:ok, row} =
+               Propagation.record_run_timing(%{
+                 run_time: run_time,
+                 forecast_hour: 5,
+                 valid_time: DateTime.add(run_time, 5, :hour),
+                 started_at: ~U[2026-07-15 12:00:00Z],
+                 finished_at: ~U[2026-07-15 12:02:30Z],
+                 duration_ms: 150_000,
+                 status: :failed,
+                 error: "HRRR fetch timeout"
+               })
+
+      assert row.status == :failed
+      assert row.error == "HRRR fetch timeout"
+    end
+
+    test "rejects duplicate (run_time, forecast_hour) pairs" do
+      attrs = %{
+        run_time: ~U[2026-07-15 12:00:00Z],
+        forecast_hour: 0,
+        valid_time: ~U[2026-07-15 12:00:00Z],
+        started_at: ~U[2026-07-15 12:00:00Z],
+        finished_at: ~U[2026-07-15 12:05:00Z],
+        duration_ms: 300_000,
+        status: :ok
+      }
+
+      assert {:ok, _} = Propagation.record_run_timing(attrs)
+      assert {:error, changeset} = Propagation.record_run_timing(attrs)
+      assert "has already been taken" in errors_on(changeset).run_time
+    end
+
+    test "requires run_time, forecast_hour, duration_ms, status" do
+      assert {:error, changeset} = Propagation.record_run_timing(%{})
+      errors = errors_on(changeset)
+      assert errors.run_time
+      assert errors.forecast_hour
+      assert errors.duration_ms
+      assert errors.status
+    end
+  end
+
+  describe "list_recent_run_timings/1" do
+    test "returns rows ordered by started_at descending, most recent first" do
+      run_time = ~U[2026-07-15 12:00:00Z]
+
+      for fh <- 0..2 do
+        {:ok, _} =
+          Propagation.record_run_timing(%{
+            run_time: run_time,
+            forecast_hour: fh,
+            valid_time: DateTime.add(run_time, fh, :hour),
+            started_at: DateTime.add(run_time, fh * 600, :second),
+            finished_at: DateTime.add(run_time, fh * 600 + 500, :second),
+            duration_ms: 500_000,
+            status: :ok
+          })
+      end
+
+      rows = Propagation.list_recent_run_timings(limit: 10)
+      assert length(rows) == 3
+      assert Enum.map(rows, & &1.forecast_hour) == [2, 1, 0]
+    end
+  end
 end