From b1fe077863a9cd70fe7686557c1ce1152121d5f6 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 25 Apr 2026 16:56:55 -0500 Subject: [PATCH] fix(path): single grid decode + zero-downtime rust-worker rollouts --- k8s/deployment-grid-rs.yaml | 16 ++++- k8s/deployment-hrrr-point-rs.yaml | 13 +++- .../propagation/profiles_file.ex | 8 ++- lib/microwaveprop_web/live/path_live.ex | 71 ++++++++++++++----- 4 files changed, 85 insertions(+), 23 deletions(-) diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 55c8fd2b..0fb40d49 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/k8s/deployment-grid-rs.yaml @@ -15,11 +15,18 @@ spec: # workers still draining. # Replica count is HPA-managed (hpa.yaml, min 1 / max 4). minReadySeconds: 5 + # Zero-downtime rollout for the propagation chain. With HPA min=1 + # and the previous maxSurge:0/maxUnavailable:1 config, every deploy + # tore the single replica down before bringing the new one up, + # leaving a multi-minute window where the :05 hourly cron had no + # worker to claim tasks. Surge-first means the new pod is ready + # before the old one drains; the two briefly overlap and race for + # `grid_tasks` rows via FOR UPDATE SKIP LOCKED, which is safe. strategy: type: RollingUpdate rollingUpdate: - maxSurge: 0 - maxUnavailable: 1 + maxSurge: 1 + maxUnavailable: 0 selector: matchLabels: app: prop-grid-rs @@ -39,6 +46,11 @@ spec: # removed so HPA can stack multiple replicas on the same node if # that's where the scheduler wants them; grid-rs is CPU-bound but # not memory-bound so colocating a couple of replicas is fine. + # Wide window for an in-flight chain step (typically 1–3 min) to + # finish before SIGKILL. The worker handles SIGTERM by stopping + # task pickup; releasing the current claim is the cron's job + # (visibility timeout reclaims abandoned `grid_tasks` rows). + terminationGracePeriodSeconds: 300 imagePullSecrets: - name: forgejo-registry securityContext: diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml index a9192604..dbec70ae 100644 --- a/k8s/deployment-hrrr-point-rs.yaml +++ b/k8s/deployment-hrrr-point-rs.yaml @@ -12,11 +12,16 @@ spec: # claiming rows from hrrr_fetch_tasks via FOR UPDATE SKIP LOCKED — # no coordination needed between workers. # Replica count is HPA-managed (hpa.yaml, min 1 / max 4). + # Zero-downtime rollout: surge first so the new pod is serving + # before the old one drains, otherwise a deploy during a QSO-submit + # burst leaves the per-QSO fetch lane empty until the new pod is + # ready. Both pods race for hrrr_fetch_tasks via FOR UPDATE SKIP + # LOCKED, so brief overlap is safe. strategy: type: RollingUpdate rollingUpdate: - maxSurge: 0 - maxUnavailable: 1 + maxSurge: 1 + maxUnavailable: 0 selector: matchLabels: app: hrrr-point-rs @@ -29,6 +34,10 @@ spec: # Anti-affinity against prop-grid-rs was removed; the scheduler # is free to colocate. Contention on wgrib2 + NFS hasn't been an # observed bottleneck in practice. + # Per-QSO point fetches are short (~5-15 s), but a stack of + # claimed rows can take a couple of minutes to drain. 120 s lets + # the worker finish in-flight tasks before SIGKILL. + terminationGracePeriodSeconds: 120 imagePullSecrets: - name: forgejo-registry securityContext: diff --git a/lib/microwaveprop/propagation/profiles_file.ex b/lib/microwaveprop/propagation/profiles_file.ex index ff738b68..17774ad1 100644 --- a/lib/microwaveprop/propagation/profiles_file.ex +++ b/lib/microwaveprop/propagation/profiles_file.ex @@ -323,7 +323,13 @@ defmodule Microwaveprop.Propagation.ProfilesFile do :ok end - defp snap(lat, lon) do + @doc """ + Snap a lat/lon pair to the nearest grid cell key used in the decoded + profile map. Public so callers that have already loaded the full grid + via `read/1` can do their own keyed lookups without re-fetching. + """ + @spec snap(float(), float()) :: {float(), float()} + def snap(lat, lon) do step = Grid.step() snapped_lat = Float.round(Float.round(lat / step) * step, 3) snapped_lon = Float.round(Float.round(lon / step) * step, 3) diff --git a/lib/microwaveprop_web/live/path_live.ex b/lib/microwaveprop_web/live/path_live.ex index c01666ee..ef8f446a 100644 --- a/lib/microwaveprop_web/live/path_live.ex +++ b/lib/microwaveprop_web/live/path_live.ex @@ -296,17 +296,23 @@ defmodule MicrowavepropWeb.PathLive do midlon = (src.lon + dst.lon) / 2 profile_valid_time = latest_profile_valid_time(now) + # Load the full profile grid ONCE (single NFS read + gunzip + decode) + # instead of letting 9 async_stream workers each re-decode the same + # ~30 MB file. Each per-point lookup becomes a Map.get. + profile_grid = profile_grid_for(profile_valid_time) sample_points = path_sample_points(src, dst, 9) - hrrr_points = - sample_points - |> Task.async_stream(&label_hrrr_point(&1, now, profile_valid_time), - max_concurrency: 9, - timeout: 10_000, - on_timeout: :kill_task - ) - |> Enum.zip(sample_points) + # Fast path: in-memory Map.get against the single decoded grid. + # Only points that miss the grid (e.g. just-rolled-over chain) take + # the slow DB-fallback path concurrently below. + {grid_hits, misses} = partition_grid_hits(sample_points, profile_grid, profile_valid_time) + + fallback_hits = + misses + |> Enum.reverse() + |> Task.async_stream(&fallback_hrrr_point(&1, now), max_concurrency: 4, timeout: 5_000, on_timeout: :kill_task) + |> Enum.zip(Enum.reverse(misses)) |> Enum.flat_map(fn {{:ok, nil}, _} -> [] @@ -316,12 +322,14 @@ defmodule MicrowavepropWeb.PathLive do {{:exit, reason}, {label, lat, lon}} -> Logger.error( - "PathLive HRRR point lookup failed: label=#{inspect(label)} lat=#{lat} lon=#{lon} valid_time=#{inspect(profile_valid_time)} reason=#{inspect(reason)}" + "PathLive HRRR fallback lookup failed: label=#{inspect(label)} lat=#{lat} lon=#{lon} reason=#{inspect(reason)}" ) [] end) + hrrr_points = Enum.reverse(grid_hits, fallback_hits) + hrrr_profiles = Enum.map(hrrr_points, & &1.profile) # Independent duct signal: the nearest RAOB within 300 km / ±3 h of @@ -455,16 +463,43 @@ defmodule MicrowavepropWeb.PathLive do defp humidity_effect_string(%{humidity_effect: :harmful}), do: "harmful" defp humidity_effect_string(_), do: "neutral" - defp label_hrrr_point({label, lat, lon}, now, profile_valid_time) do - case profile_valid_time && ProfilesFile.read_point(profile_valid_time, lat, lon) do - %{} = cell -> - %{label: label, profile: profile_from_cell(cell, lat, lon, profile_valid_time)} + # The grid is the single decoded output of ProfilesFile.read/1 — keys are + # `{snapped_lat, snapped_lon}` tuples produced by ProfilesFile.snap/2. + # We mirror the same snapping locally instead of round-tripping through + # ProfilesFile.read_point/3 (which would re-fetch the cached grid). + defp profile_grid_for(nil), do: nil - _ -> - case Weather.find_nearest_hrrr(lat, lon, now) do - nil -> nil - profile -> %{label: label, profile: profile} - end + defp profile_grid_for(valid_time) do + case ProfilesFile.read(valid_time) do + {:ok, grid} -> grid + _ -> nil + end + end + + defp lookup_profile_grid(grid, lat, lon) do + {snapped_lat, snapped_lon} = ProfilesFile.snap(lat, lon) + Map.get(grid, {snapped_lat, snapped_lon}) + end + + defp partition_grid_hits(sample_points, profile_grid, profile_valid_time) do + Enum.reduce(sample_points, {[], []}, fn pt, acc -> + classify_grid_hit(pt, profile_grid, profile_valid_time, acc) + end) + end + + defp classify_grid_hit(pt, nil, _vt, {hits, mss}), do: {hits, [pt | mss]} + + defp classify_grid_hit({label, lat, lon} = pt, grid, vt, {hits, mss}) do + case lookup_profile_grid(grid, lat, lon) do + nil -> {hits, [pt | mss]} + cell -> {[%{label: label, profile: profile_from_cell(cell, lat, lon, vt)} | hits], mss} + end + end + + defp fallback_hrrr_point({label, lat, lon}, now) do + case Weather.find_nearest_hrrr(lat, lon, now) do + nil -> nil + profile -> %{label: label, profile: profile} end end