From b9847945714cc97c1a59d9ca937e1435d1c06628 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 15:01:12 -0500 Subject: [PATCH] Hoist commercial link query + tie pipeline chip to ScoresFile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PropagationGridWorker.merge_commercial_link_data/2 called Commercial.link_degradation_at twice per grid cell (once to count, once to merge), and each call re-ran enabled_links plus two sample queries per in-range link. That was ~500k SQL queries per forecast hour — fine for the DB but catastrophic for log volume in dev iex sessions trying to watch a chain step run. Split into two new functions: * build_link_lookup/2 does all the DB work up front — one enabled_links query + one link_degradation per enabled link (~10 queries total). * link_degradation_from_lookup/3 is pure: takes a (lat, lon) and the precomputed lookup, returns the aggregate or nil. The worker now calls build_link_lookup once per forecast hour and the per-cell path is haversine-only. Net: 500k queries → ~10. PipelineStatus freshness detection moves off oban_jobs.completed_at onto ScoresFile.latest_valid_time(). The on-disk files ARE the data the map renders from, so the chip's "Up to date · Nm ago" and the 2h stale threshold now track actual data presence. Running-state detection still comes from oban_jobs since the chain step is still an Oban row. Tests updated to seed a ScoresFile instead of a completed Oban row for the idle/stale cases. --- lib/microwaveprop/commercial.ex | 79 ++++++++++++------- .../propagation/pipeline_status.ex | 40 ++++------ .../workers/propagation_grid_worker.ex | 28 ++++--- .../propagation/pipeline_status_test.exs | 46 +++++------ 4 files changed, 103 insertions(+), 90 deletions(-) diff --git a/lib/microwaveprop/commercial.ex b/lib/microwaveprop/commercial.ex index fd0e2031..1d4e6000 100644 --- a/lib/microwaveprop/commercial.ex +++ b/lib/microwaveprop/commercial.ex @@ -45,43 +45,68 @@ defmodule Microwaveprop.Commercial do """ @spec link_degradation_at({float(), float()}, DateTime.t(), keyword()) :: map() | nil def link_degradation_at({lat, lon}, valid_time, opts \\ []) do - radius_km = Keyword.get(opts, :radius_km, @default_radius_km) + lookup = build_link_lookup(valid_time, opts) + link_degradation_from_lookup({lat, lon}, lookup, opts) + end + + @doc """ + Precompute every enabled link's degradation once so a bulk caller + (like PropagationGridWorker's per-cell merge) can reuse the same + `(link, endpoint, degradation)` list across all 95k grid points + instead of re-running `enabled_links/0` and the per-link sample + queries per call. Result is fed back into + `link_degradation_from_lookup/3`. + """ + @spec build_link_lookup(DateTime.t(), keyword()) :: [ + {Link.t(), {float(), float()}, map() | nil} + ] + def build_link_lookup(valid_time, opts \\ []) do baseline_days = Keyword.get(opts, :baseline_days, @default_baseline_days) current_window = Keyword.get(opts, :current_window_seconds, @default_current_window_seconds) - candidates = - enabled_links() - |> Enum.map(fn link -> {link, link_endpoint(link)} end) - |> Enum.reject(fn {_link, endpoint} -> is_nil(endpoint) end) - |> Enum.filter(fn {_link, {elat, elon}} -> - haversine_km(lat, lon, elat, elon) <= radius_km - end) - baseline_cutoff = DateTime.add(valid_time, -baseline_days * 24 * 3600, :second) current_cutoff = DateTime.add(valid_time, -current_window, :second) + enabled_links() + |> Enum.map(fn link -> {link, link_endpoint(link)} end) + |> Enum.reject(fn {_link, endpoint} -> is_nil(endpoint) end) + |> Enum.map(fn {link, endpoint} -> + degradation = link_degradation(link.id, baseline_cutoff, current_cutoff, valid_time) + {link, endpoint, degradation} + end) + end + + @doc """ + Pure per-cell aggregator — given a `(lat, lon)` point and a + precomputed lookup from `build_link_lookup/2`, returns the + aggregate degradation map or `nil` when no in-range link is + reporting usable data. + """ + @spec link_degradation_from_lookup({float(), float()}, [tuple()], keyword()) :: map() | nil + def link_degradation_from_lookup({lat, lon}, lookup, opts \\ []) do + radius_km = Keyword.get(opts, :radius_km, @default_radius_km) + results = - candidates - |> Enum.map(fn {link, _endpoint} -> - link_degradation(link.id, baseline_cutoff, current_cutoff, valid_time) - end) - |> Enum.reject(&is_nil/1) + for {_link, {elat, elon}, %{} = degradation} <- lookup, + haversine_km(lat, lon, elat, elon) <= radius_km do + degradation + end - case results do - [] -> - nil + aggregate_degradation(results) + end - list -> - baseline_avg = average(Enum.map(list, & &1.baseline_dbm)) - current_avg = average(Enum.map(list, & &1.current_dbm)) + defp aggregate_degradation([]), do: nil - %{ - degradation_db: Float.round(baseline_avg - current_avg, 2), - baseline_dbm: Float.round(baseline_avg, 2), - current_dbm: Float.round(current_avg, 2), - n_links: length(list) - } - end + defp aggregate_degradation(list) do + baseline_avg = average(Enum.map(list, & &1.baseline_dbm)) + current_avg = average(Enum.map(list, & &1.current_dbm)) + + %{ + degradation_db: Float.round(baseline_avg - current_avg, 2), + baseline_dbm: Float.round(baseline_avg, 2), + current_dbm: Float.round(current_avg, 2), + n_links: length(list) + } end defp link_degradation(link_id, baseline_cutoff, current_cutoff, valid_time) do diff --git a/lib/microwaveprop/propagation/pipeline_status.ex b/lib/microwaveprop/propagation/pipeline_status.ex index d554c8fa..2314245b 100644 --- a/lib/microwaveprop/propagation/pipeline_status.ex +++ b/lib/microwaveprop/propagation/pipeline_status.ex @@ -2,20 +2,21 @@ defmodule Microwaveprop.Propagation.PipelineStatus do @moduledoc """ Aggregated status for the propagation update pipeline. - The pipeline is driven by two Oban workers: + Two inputs drive the chip: - * `PropagationGridWorker` — hourly, fetches HRRR f00–f18, scores - the CONUS grid, and broadcasts `propagation:updated`. - * `AsosAdjustmentWorker` — every 10 minutes, nudges scores with - recent ASOS surface observations. - - `current/0` returns one struct so the map page can render a single - status chip ("Updating…", "Up to date · 8m ago", "Stale · 4h ago") - that reflects the whole pipeline rather than tailing a single job. + * **Running detection** — queries `oban_jobs` for executing + PropagationGridWorker / AsosAdjustmentWorker rows so the chip + knows when a chain step is in flight. + * **Freshness detection** — reads the newest valid_time from + `Microwaveprop.Propagation.ScoresFile` on disk. The on-disk + files ARE the propagation data now, so the chip's "Up to date · + Nm ago" and the 2h stale threshold track actual data presence + instead of Oban's `completed_at` timestamp. """ import Ecto.Query + alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Repo @grid_worker "Microwaveprop.Workers.PropagationGridWorker" @@ -58,14 +59,14 @@ defmodule Microwaveprop.Propagation.PipelineStatus do def current do case running_workers() do [] -> - build_idle_or_stale(latest_completed_at()) + build_idle_or_stale(latest_data_at()) details -> %{ state: :running, label: "Updating propagation", details: details, - last_update_at: latest_completed_at() + last_update_at: latest_data_at() } end end @@ -91,18 +92,11 @@ defmodule Microwaveprop.Propagation.PipelineStatus do defp worker_sort_key(@asos_worker), do: 1 defp worker_sort_key(_), do: 99 - defp latest_completed_at do - case Repo.one( - from j in "oban_jobs", - where: j.state == "completed" and j.worker in ^@workers, - order_by: [desc: j.completed_at], - limit: 1, - select: j.completed_at - ) do - nil -> nil - %NaiveDateTime{} = naive -> DateTime.from_naive!(naive, "Etc/UTC") - %DateTime{} = dt -> dt - end + # Freshness is derived from the newest ScoresFile on disk. That file + # is the thing the map actually renders from, so its valid_time is + # the most honest answer to "when was the data last updated?". + defp latest_data_at do + ScoresFile.latest_valid_time() end defp build_idle_or_stale(nil) do diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 391fcd56..6914ed50 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -254,25 +254,27 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do end defp merge_commercial_link_data(grid_data, valid_time) do - # Compute link degradation once per distinct link cluster and cache by - # point. Commercial links only cluster around DFW so most grid points see - # nil — cheap no-op path dominates. - boosted = - Enum.count(grid_data, fn {{lat, lon}, _profile} -> - degradation = Commercial.link_degradation_at({lat, lon}, valid_time) - not is_nil(degradation) + # Precompute per-link degradation once (≤10 SQL queries total). + # Commercial links cluster around DFW so most grid cells see nil — + # the per-cell path is now a pure haversine check, not a DB query. + lookup = Commercial.build_link_lookup(valid_time) + + {merged, boosted} = + Enum.reduce(grid_data, {%{}, 0}, fn {{lat, lon} = point, profile}, {acc, count} -> + case Commercial.link_degradation_from_lookup({lat, lon}, lookup) do + nil -> + {Map.put(acc, point, profile), count} + + degradation -> + {Map.put(acc, point, Map.put(profile, :commercial_link_degradation, degradation)), count + 1} + end end) if boosted > 0 do Logger.info("PropagationGrid: commercial-link degradation available for #{boosted} grid cells") end - Map.new(grid_data, fn {{lat, lon} = point, profile} -> - case Commercial.link_degradation_at({lat, lon}, valid_time) do - nil -> {point, profile} - degradation -> {point, Map.put(profile, :commercial_link_degradation, degradation)} - end - end) + merged end defp merge_nexrad_data(grid_data, valid_time) do diff --git a/test/microwaveprop/propagation/pipeline_status_test.exs b/test/microwaveprop/propagation/pipeline_status_test.exs index 4b43f917..855c3646 100644 --- a/test/microwaveprop/propagation/pipeline_status_test.exs +++ b/test/microwaveprop/propagation/pipeline_status_test.exs @@ -2,6 +2,7 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do use Microwaveprop.DataCase, async: false alias Microwaveprop.Propagation.PipelineStatus + alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Repo @grid_worker "Microwaveprop.Workers.PropagationGridWorker" @@ -99,15 +100,12 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do assert asos.label =~ "ASOS" end - test "returns :idle with Up to date label when last grid run completed recently" do - completed = minutes_ago(15) - - insert_oban_job(%{ - state: "completed", - worker: @grid_worker, - attempted_at: completed, - completed_at: completed - }) + test "returns :idle with Up to date label when a recent ScoresFile exists" do + ScoresFile.write!( + 10_000, + minutes_ago(15), + [%{lat: 25.0, lon: -125.0, score: 50}] + ) status = PipelineStatus.current() @@ -116,15 +114,12 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do assert status.label =~ "Up to date" end - test "returns :stale when last completed was more than 120 minutes ago" do - stale = minutes_ago(180) - - insert_oban_job(%{ - state: "completed", - worker: @grid_worker, - attempted_at: stale, - completed_at: stale - }) + test "returns :stale when the newest ScoresFile is more than 120 minutes old" do + ScoresFile.write!( + 10_000, + minutes_ago(180), + [%{lat: 25.0, lon: -125.0, score: 50}] + ) status = PipelineStatus.current() @@ -132,15 +127,12 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do assert status.label =~ "stale" end - test "prefers :running over :idle when a job is executing even if a recent job completed" do - recent = minutes_ago(5) - - insert_oban_job(%{ - state: "completed", - worker: @grid_worker, - attempted_at: recent, - completed_at: recent - }) + test "prefers :running over :idle when a job is executing even with a recent ScoresFile" do + ScoresFile.write!( + 10_000, + minutes_ago(5), + [%{lat: 25.0, lon: -125.0, score: 50}] + ) insert_oban_job(%{ state: "executing",