diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 1adecfa0..69e41e01 100644 --- a/lib/microwaveprop/propagation.ex +++ b/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. diff --git a/lib/microwaveprop/propagation/run_timing.ex b/lib/microwaveprop/propagation/run_timing.ex new file mode 100644 index 00000000..782e2028 --- /dev/null +++ b/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 diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index e94db331..f1b9ff6d 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/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 diff --git a/priv/repo/migrations/20260418143251_create_propagation_run_timings.exs b/priv/repo/migrations/20260418143251_create_propagation_run_timings.exs new file mode 100644 index 00000000..f34c7a7b --- /dev/null +++ b/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 diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index b462961a..9b487958 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/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