From 2405c5f16928c6ec200d3d552e23785e8f525ddd Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 13 Apr 2026 08:20:28 -0500 Subject: [PATCH] Stop PropagationGridWorker crash-looping on forecast hours warm_grid_cache_and_broadcast was re-SELECTing 92k hrrr_profiles rows with JSONB profile/duct_characteristics columns on every forecast hour. Row decoding exceeded the 15s default DB connection timeout, killing the worker before f01-f18 could run. Oban retried from f00 and got stuck in a loop: for the last several hours, no forecast hours were being written and even f00 was stale. Build the weather cache rows directly from the in-memory grid_data the worker already has (Weather.build_grid_cache_rows/2) and GridCache.broadcast_put directly. No DB round trip, no JSONB decode. Also widen the cron from hourly to every 3 hours so a full f00-f18 sweep (~95 min) can actually complete before the next run starts, and drop the redundant prune_old_scores() at worker start (PropagationPruneWorker already runs every 15 min). Add a 60s query timeout to load_weather_grid_from_db as defense-in-depth for the cold-cache fill path that still uses it. --- config/config.exs | 2 +- config/dev.exs | 2 +- config/runtime.exs | 5 +- lib/microwaveprop/weather.ex | 69 +++++++- .../workers/propagation_grid_worker.ex | 11 +- test/microwaveprop/weather_grid_test.exs | 152 ++++++++++++++++++ 6 files changed, 228 insertions(+), 13 deletions(-) diff --git a/config/config.exs b/config/config.exs index 5dc4d56e..a623e4cf 100644 --- a/config/config.exs +++ b/config/config.exs @@ -69,7 +69,7 @@ config :microwaveprop, Oban, crontab: [ {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, - {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker} diff --git a/config/dev.exs b/config/dev.exs index c1394f64..ce9724ef 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -87,7 +87,7 @@ config :microwaveprop, Oban, {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 30)}, {Oban.Plugins.Cron, crontab: [ - {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, diff --git a/config/runtime.exs b/config/runtime.exs index 1a7f42a1..8dcd9aa4 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -182,7 +182,10 @@ if config_env() == :prod do crontab: [ {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, - {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + # Every 3 hours: one full run does f00–f18 (~5 min/hour × 19 ≈ 95 min). + # Hourly cron caused overlapping/failed runs because one full sweep + # exceeds a 60-minute window. + {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"*/30 * * * *", Microwaveprop.Workers.BackfillEnqueueWorker, args: %{"limit" => 500, "types" => ["hrrr", "weather", "terrain", "iemre"]}} diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index fb5b5c73..90289b76 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -13,6 +13,7 @@ defmodule Microwaveprop.Weather do alias Microwaveprop.Weather.RtmaObservation alias Microwaveprop.Weather.SolarIndex alias Microwaveprop.Weather.Sounding + alias Microwaveprop.Weather.SoundingParams alias Microwaveprop.Weather.Station alias Microwaveprop.Weather.SurfaceObservation alias Microwaveprop.Weather.WeatherLayers @@ -481,15 +482,16 @@ defmodule Microwaveprop.Weather do duct_characteristics: h.duct_characteristics } ) - |> Repo.all() + |> Repo.all(timeout: 60_000) |> Enum.map(&derive_and_clean/1) end @doc """ Eagerly populate the `GridCache` with the full CONUS weather grid for - `valid_time` and broadcast it to every node in the cluster. Called from - `PropagationGridWorker` after each hourly HRRR upsert so connected clients - see zero-DB pan/zoom latency on the `/weather` map. + `valid_time` and broadcast it to every node in the cluster. Used by the cold + cache fill path (`kickoff_async_grid_fill/1`) — prefer + `build_grid_cache_rows/2` inside `PropagationGridWorker`, which already has + the in-memory grid data and avoids the ~20s JSONB round trip. """ @spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok def warm_grid_cache_and_broadcast(valid_time) do @@ -498,6 +500,65 @@ defmodule Microwaveprop.Weather do :ok end + @doc """ + Build derived weather grid cache rows directly from the in-memory HRRR + `grid_data` that `PropagationGridWorker` already has after fetching a + forecast hour. Avoids the 15s+ DB round trip that `load_weather_grid_from_db/1` + pays to SELECT + JSONB-decode 92k hrrr_profiles rows — the source of the + crash loop that stopped forecast hours from being written. + + `grid_data` is `%{{lat, lon} => profile_map}` as produced by + `HrrrClient.fetch_grid/3` (optionally enriched via native duct merge). The + output matches the shape of `load_weather_grid_from_db/1` after + `derive_and_clean/1`. + """ + @spec build_grid_cache_rows(%{{float(), float()} => map()}, DateTime.t()) :: [map()] + def build_grid_cache_rows(grid_data, valid_time) do + Enum.flat_map(grid_data, fn {{lat, lon}, profile} -> + build_grid_cache_row(lat, lon, profile, valid_time) + end) + end + + defp build_grid_cache_row(lat, lon, profile, valid_time) do + temp_c = profile[:surface_temp_c] + + if is_nil(temp_c) or temp_c < -80 or temp_c > 60 do + [] + else + sounding = derive_sounding(profile[:profile]) + + row = %{ + lat: lat, + lon: lon, + valid_time: valid_time, + temperature: temp_c, + dewpoint_depression: depression(temp_c, profile[:surface_dewpoint_c]), + bl_height: profile[:hpbl_m], + pwat: profile[:pwat_mm], + refractivity_gradient: sounding[:min_refractivity_gradient], + ducting: sounding[:ducting_detected], + surface_pressure_mb: profile[:surface_pressure_mb], + surface_temp_c: temp_c, + surface_dewpoint_c: profile[:surface_dewpoint_c], + surface_refractivity: sounding[:surface_refractivity], + profile: profile[:profile] || [], + duct_characteristics: sounding[:duct_characteristics] + } + + [derive_and_clean(row)] + end + end + + defp derive_sounding(profile) when is_list(profile) and length(profile) >= 3 do + SoundingParams.derive(profile) || %{} + end + + defp derive_sounding(_), do: %{} + + defp depression(nil, _), do: nil + defp depression(_, nil), do: nil + defp depression(t, d), do: t - d + @spec weather_point_detail(float(), float(), DateTime.t()) :: map() | nil def weather_point_detail(lat, lon, valid_time) do step = 0.125 diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index ba6fd19f..1b89bf30 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -15,6 +15,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do alias Microwaveprop.Propagation.Grid alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Weather + alias Microwaveprop.Weather.GridCache alias Microwaveprop.Weather.HrrrClient alias Microwaveprop.Weather.HrrrNativeClient alias Microwaveprop.Weather.SoundingParams @@ -27,11 +28,6 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do def perform(%Oban.Job{}) do t_start = System.monotonic_time(:millisecond) - # Safety net: prune before compute so a row of failed runs doesn't let - # stale scores accumulate indefinitely. Pruning also runs on its own cron - # (`PropagationPruneWorker`) independent of this worker's success. - Propagation.prune_old_scores() - # 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) @@ -85,7 +81,10 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do grid_data = merge_native_duct_data(grid_data, run_time, forecast_hour) :erlang.garbage_collect() - Weather.warm_grid_cache_and_broadcast(valid_time) + # Build weather cache rows from in-memory grid_data — avoids the ~20s + # JSONB round trip that was crashing the worker before f01–f18 could run. + rows = Weather.build_grid_cache_rows(grid_data, valid_time) + GridCache.broadcast_put(valid_time, rows) Phoenix.PubSub.broadcast( Microwaveprop.PubSub, diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index 47bb545b..d409f949 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -178,6 +178,158 @@ defmodule Microwaveprop.WeatherGridTest do end end + describe "build_grid_cache_rows/2" do + test "returns an empty list for empty grid_data" do + now = DateTime.truncate(DateTime.utc_now(), :second) + assert Weather.build_grid_cache_rows(%{}, now) == [] + end + + test "builds a derived row for each in-memory grid point without hitting the database" do + now = DateTime.truncate(DateTime.utc_now(), :second) + + profile = [ + %{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0}, + %{"pres" => 925.0, "tmpc" => 20.0, "dwpc" => 14.0, "hght" => 800.0}, + %{"pres" => 850.0, "tmpc" => 15.0, "dwpc" => 10.0, "hght" => 1500.0}, + %{"pres" => 700.0, "tmpc" => 5.0, "dwpc" => -2.0, "hght" => 3100.0}, + %{"pres" => 500.0, "tmpc" => -15.0, "dwpc" => -25.0, "hght" => 5600.0} + ] + + grid_data = %{ + {32.125, -97.375} => %{ + profile: profile, + surface_temp_c: 25.0, + surface_dewpoint_c: 18.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0 + }, + {32.250, -97.375} => %{ + profile: profile, + surface_temp_c: 26.0, + surface_dewpoint_c: 19.0, + surface_pressure_mb: 1012.0, + hpbl_m: 600.0, + pwat_mm: 32.0 + } + } + + result = Weather.build_grid_cache_rows(grid_data, now) + + assert length(result) == 2 + + point = Enum.find(result, &(&1.lat == 32.125)) + assert point.lon == -97.375 + assert point.valid_time == now + assert point.temperature == 25.0 + assert point.dewpoint_depression == 7.0 + assert point.bl_height == 500.0 + assert point.pwat == 30.0 + assert point.surface_pressure_mb == 1013.0 + end + + test "computes surface_refractivity and min_refractivity_gradient from the profile" do + now = DateTime.truncate(DateTime.utc_now(), :second) + + profile = [ + %{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0}, + %{"pres" => 925.0, "tmpc" => 20.0, "dwpc" => 14.0, "hght" => 800.0}, + %{"pres" => 850.0, "tmpc" => 15.0, "dwpc" => 10.0, "hght" => 1500.0}, + %{"pres" => 700.0, "tmpc" => 5.0, "dwpc" => -2.0, "hght" => 3100.0}, + %{"pres" => 500.0, "tmpc" => -15.0, "dwpc" => -25.0, "hght" => 5600.0} + ] + + grid_data = %{ + {32.125, -97.375} => %{ + profile: profile, + surface_temp_c: 25.0, + surface_dewpoint_c: 18.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0 + } + } + + [point] = Weather.build_grid_cache_rows(grid_data, now) + + assert is_float(point.surface_refractivity) + assert is_float(point.refractivity_gradient) + assert is_float(point.surface_rh) + assert point.temp_850mb == 15.0 + assert point.dewpoint_850mb == 10.0 + end + + test "skips points with missing or physically impossible surface temperature" do + now = DateTime.truncate(DateTime.utc_now(), :second) + + profile = [ + %{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0}, + %{"pres" => 925.0, "tmpc" => 20.0, "dwpc" => 14.0, "hght" => 800.0}, + %{"pres" => 850.0, "tmpc" => 15.0, "dwpc" => 10.0, "hght" => 1500.0} + ] + + grid_data = %{ + {32.125, -97.375} => %{ + profile: profile, + surface_temp_c: nil, + surface_dewpoint_c: nil, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0 + }, + {32.250, -97.375} => %{ + profile: profile, + surface_temp_c: 250.0, + surface_dewpoint_c: 200.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0 + }, + {32.375, -97.375} => %{ + profile: profile, + surface_temp_c: 25.0, + surface_dewpoint_c: 18.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0 + } + } + + result = Weather.build_grid_cache_rows(grid_data, now) + + assert length(result) == 1 + assert hd(result).lat == 32.375 + end + + test "strips raw profile/duct_characteristics/surface_temp_c/surface_dewpoint_c fields" do + now = DateTime.truncate(DateTime.utc_now(), :second) + + profile = [ + %{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0}, + %{"pres" => 925.0, "tmpc" => 20.0, "dwpc" => 14.0, "hght" => 800.0}, + %{"pres" => 850.0, "tmpc" => 15.0, "dwpc" => 10.0, "hght" => 1500.0} + ] + + grid_data = %{ + {32.125, -97.375} => %{ + profile: profile, + surface_temp_c: 25.0, + surface_dewpoint_c: 18.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0 + } + } + + [point] = Weather.build_grid_cache_rows(grid_data, now) + + refute Map.has_key?(point, :profile) + refute Map.has_key?(point, :duct_characteristics) + refute Map.has_key?(point, :surface_temp_c) + refute Map.has_key?(point, :surface_dewpoint_c) + end + end + describe "weather_point_detail/3" do test "returns all fields for a single grid point" do now = DateTime.truncate(DateTime.utc_now(), :second)