From 48708621c50c97c19a278553c3e56dacfd2f53e3 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 17:25:51 -0500 Subject: [PATCH] =?UTF-8?q?chore(mrms):=20delete=20pipeline=20=E2=80=94=20?= =?UTF-8?q?consumer=20AsosAdjustmentWorker=20already=20disabled?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MrmsFetchWorker fired every 2 min on the :propagation queue, decoding an MRMS PrecipRate GRIB2 into MrmsCache. Its only consumer, AsosAdjustmentWorker, is disabled in all configs (dev / config / runtime). Net effect: 30 GRIB2 decodes/hour of dead work on hot pods. Removed: - MrmsFetchWorker, MrmsClient, MrmsCache and their tests - cron entries in config.exs, dev.exs, runtime.exs - mrms_req_options stub in test.exs - MrmsCache from supervision tree in application.ex - stale MRMS references in PropagationGridWorker docstring and test Also adds docs/plans/2026-04-19-rust-migration-phase3.md which tracks the broader Stream A (f00 port) and Stream C (HrrrFetchWorker port) follow-on work. --- config/config.exs | 3 - config/dev.exs | 3 - config/runtime.exs | 11 - config/test.exs | 1 - .../plans/2026-04-19-rust-migration-phase3.md | 607 ++++++++++++++++++ lib/microwaveprop/application.ex | 1 - lib/microwaveprop/weather/mrms_cache.ex | 80 --- lib/microwaveprop/weather/mrms_client.ex | 190 ------ .../workers/mrms_fetch_worker.ex | 46 -- .../workers/propagation_grid_worker.ex | 9 +- .../microwaveprop/weather/mrms_cache_test.exs | 97 --- .../weather/mrms_client_test.exs | 72 --- .../workers/mrms_fetch_worker_test.exs | 137 ---- .../workers/propagation_grid_worker_test.exs | 6 +- 14 files changed, 613 insertions(+), 650 deletions(-) create mode 100644 docs/plans/2026-04-19-rust-migration-phase3.md delete mode 100644 lib/microwaveprop/weather/mrms_cache.ex delete mode 100644 lib/microwaveprop/weather/mrms_client.ex delete mode 100644 lib/microwaveprop/workers/mrms_fetch_worker.ex delete mode 100644 test/microwaveprop/weather/mrms_cache_test.exs delete mode 100644 test/microwaveprop/weather/mrms_client_test.exs delete mode 100644 test/microwaveprop/workers/mrms_fetch_worker_test.exs diff --git a/config/config.exs b/config/config.exs index d72cfa72..97f6f21a 100644 --- a/config/config.exs +++ b/config/config.exs @@ -98,9 +98,6 @@ config :microwaveprop, Oban, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, - {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, - # AsosAdjustmentWorker disabled — see runtime.exs for rationale. - # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, # UWYO publishes the 00Z/12Z Canadian radiosondes ~90 minutes after launch {"30 1 * * *", CanadianSoundingFetchWorker}, diff --git a/config/dev.exs b/config/dev.exs index 95161cdf..682a8355 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -97,9 +97,6 @@ config :microwaveprop, Oban, {Oban.Plugins.Cron, crontab: [ {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, - {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, - # AsosAdjustmentWorker disabled — see runtime.exs for rationale. - # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} ]} diff --git a/config/runtime.exs b/config/runtime.exs index 37bab24e..196dee74 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -203,17 +203,6 @@ if config_env() == :prod do # publication lag. Seeder enqueues f024-f168 at 6-hour cadence # (25 jobs per run, 100/day) into the :gefs queue. {"30 5,11,17,23 * * *", Microwaveprop.Workers.GefsFetchWorker}, - # MRMS refreshes the PrecipRate cache that AsosAdjustmentWorker - # overlays onto the score grid. Both must run together — an - # ASOS nudge without a fresh MRMS frame reverts to HRRR-only - # rain. Requires wgrib2 built with PNG decoder support (DRT - # 5.41), which the Dockerfile now enables via NCEPLIBS-g2c. - {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, - # AsosAdjustmentWorker disabled: depends on hrrr_profiles rows - # that PropagationGridWorker no longer persists. The 10-min - # ASOS nudge is a nice-to-have refinement we're not paying the - # full grid-side JSONB write for. - # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, # The :narr type is virtual: it targets contacts with hrrr_status = # :unavailable (pre-2014, missing from the HRRR archive) and dispatches diff --git a/config/test.exs b/config/test.exs index d2b1b0a8..027fbbd0 100644 --- a/config/test.exs +++ b/config/test.exs @@ -63,7 +63,6 @@ config :microwaveprop, giro_req_options: [plug: {Req.Test, Microwaveprop.Ionosph config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false] config :microwaveprop, iem_rate_limiter_interval_ms: 0 config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false] -config :microwaveprop, mrms_req_options: [plug: {Req.Test, Microwaveprop.Weather.MrmsClient}, retry: false] config :microwaveprop, narr_req_options: [plug: {Req.Test, Microwaveprop.Weather.NarrClient}, retry: false] # Route HTTP requests through Req.Test stubs diff --git a/docs/plans/2026-04-19-rust-migration-phase3.md b/docs/plans/2026-04-19-rust-migration-phase3.md new file mode 100644 index 00000000..d4efb054 --- /dev/null +++ b/docs/plans/2026-04-19-rust-migration-phase3.md @@ -0,0 +1,607 @@ +# Rust Migration Phase 3 — f00 Port + MRMS Deletion + HRRR Per-QSO Port + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Remove every remaining GRIB2 decode from BEAM so hot pods can shrink to 1 Gi, and delete one dead pipeline (MRMS) along the way. + +**Architecture:** Three independent work streams against the same shared `prop-grid-rs` binary. Stream A ports f00's native-duct + NEXRAD + commercial-link merge from Elixir into Rust via a new `grid_tasks.kind` column so the same work queue handles both forecast hours and analysis-hour enrichment. Stream B deletes the MRMS worker/cache/client entirely (the only consumer `AsosAdjustmentWorker` is already disabled). Stream C ports the per-QSO `HrrrFetchWorker` into a new Rust binary sharing the same crate, taking the `:hrrr` queue off BEAM. + +**Tech Stack:** Rust (tokio, sqlx, reqwest, wgrib2 subprocess), Elixir (Oban, Postgrex.Notifications), Postgres (grid_tasks work queue + LISTEN/NOTIFY), Kubernetes/flux. + +--- + +## Work Streams Overview + +| Stream | Priority | What | Payoff | +|---|---|---|---| +| **A: Port f00** | 1 | Move f00 chain step (native duct + NEXRAD + commercial + ProfilesFile write) from Elixir to Rust. | Finishes hourly-chain memory story. Hot pod can drop from 2Gi/replica × N replicas to 1Gi. | +| **B: Delete MRMS** | 2 | Remove MrmsFetchWorker, MrmsClient, MrmsCache, and the `*/2 * * * *` cron. | Removes 30 GRIB2 decodes/hour of dead work. Zero risk — AsosAdjustmentWorker (only consumer) is already disabled. | +| **C: Port HrrrFetchWorker** | 3 | Move per-QSO HRRR profile fetch to Rust. New `hrrr_point_tasks` queue. Elixir enqueues, Rust fulfills. | Takes the `:hrrr` queue off BEAM. Removes per-QSO refc-binary churn and the explicit `:erlang.garbage_collect` that's papering over it. | + +Each stream is independently mergeable and reversible. Recommended order: **B → A → C**. B first because it's a pure delete (simplest; shrinks the attack surface before A touches more code). A second because it's the real memory win. C last because it's optional polish. + +--- + +## Stream B: Delete MRMS (start here — warm-up) + +**Context:** `MrmsFetchWorker` runs every 2 minutes on the `:propagation` queue. It decodes an MRMS PrecipRate GRIB2 and caches the regridded result in `MrmsCache`. The only consumer that reads that cache, `AsosAdjustmentWorker`, is disabled in all three configs (see `config/runtime.exs:212-216`). Net effect: 30 GRIB2 decodes/hour of dead work on the hot pods. + +### Task B1: Confirm no live consumers + +**Files to check:** +- Search for `MrmsCache` and `Mrms\.` across `lib/` and `test/` — confirm only the worker/client/cache/tests reference it. +- Search for `AsosAdjustmentWorker` — confirm the only references are disabled cron entries + the worker file itself. + +**Step 1: Run the searches** + +```bash +grep -rn "MrmsCache\|Mrms\." lib/ config/ | grep -v "#" +grep -rn "AsosAdjustmentWorker" config/ lib/ +``` + +**Expected output:** References only in `workers/mrms_fetch_worker.ex`, `weather/mrms_client.ex`, `weather/mrms_cache.ex`, `application.ex` (supervisor), and commented-out cron lines. No LiveView, context, or worker depends on it. + +**If you find a live consumer, STOP** — add that consumer to the removal list before proceeding. + +### Task B2: Remove MRMS from supervisor + cron + +**Files:** +- Modify: `lib/microwaveprop/application.ex` (drop `MrmsCache` from the supervision tree) +- Modify: `config/runtime.exs:206-211` (drop the `MrmsFetchWorker` cron line and its comment block) +- Modify: `config/config.exs:101` (same removal in dev-carrying config) +- Modify: `config/dev.exs:100` (same) + +**Step 1: Write the failing test** + +Add to `test/microwaveprop/application_test.exs` (create if missing): + +```elixir +defmodule Microwaveprop.ApplicationTest do + use ExUnit.Case, async: true + + test "MrmsCache is not in the supervision tree" do + children = Microwaveprop.Application.children_for_env(:test) + refute Enum.any?(children, &match?({Microwaveprop.Weather.MrmsCache, _}, &1)) + refute Enum.any?(children, fn + Microwaveprop.Weather.MrmsCache -> true + _ -> false + end) + end +end +``` + +If `children_for_env/1` doesn't exist, either extract it from `start/2` or inline the list in the test. + +**Step 2: Run test to verify it fails** + +```bash +mix test test/microwaveprop/application_test.exs +``` + +Expected: FAIL — MrmsCache still in supervision tree. + +**Step 3: Remove MrmsCache from supervision tree** + +Edit `lib/microwaveprop/application.ex`, delete the line that adds `MrmsCache`. + +**Step 4: Run test to verify it passes** + +```bash +mix test test/microwaveprop/application_test.exs +``` + +Expected: PASS. + +**Step 5: Remove cron entries** + +Delete the `MrmsFetchWorker` entry + its leading comment block from all three config files. + +**Step 6: Commit** + +```bash +git add lib/microwaveprop/application.ex config/ test/microwaveprop/application_test.exs +git commit -m "chore(mrms): remove from supervision tree and cron (consumer disabled)" +``` + +### Task B3: Delete MRMS source files and tests + +**Files:** +- Delete: `lib/microwaveprop/workers/mrms_fetch_worker.ex` +- Delete: `lib/microwaveprop/weather/mrms_client.ex` +- Delete: `lib/microwaveprop/weather/mrms_cache.ex` +- Delete: `test/microwaveprop/workers/mrms_fetch_worker_test.exs` +- Delete: `test/microwaveprop/weather/mrms_cache_test.exs` +- Modify: `config/test.exs:66` (drop `mrms_req_options`) + +**Step 1: Delete files** + +```bash +rm lib/microwaveprop/workers/mrms_fetch_worker.ex \ + lib/microwaveprop/weather/mrms_client.ex \ + lib/microwaveprop/weather/mrms_cache.ex \ + test/microwaveprop/workers/mrms_fetch_worker_test.exs \ + test/microwaveprop/weather/mrms_cache_test.exs +``` + +**Step 2: Drop the test config line** + +Edit `config/test.exs:66`, remove the `mrms_req_options` line. + +**Step 3: Run full test suite** + +```bash +mix precommit +``` + +Expected: clean compile (warnings-as-errors), all tests pass. If `mix xref` surfaces a caller, fix it. + +**Step 4: Commit** + +```bash +git add -A +git commit -m "chore(mrms): delete worker, client, cache, and tests" +``` + +### Task B4: Drop MRMS k8s resources (if any) + +**Files to inspect:** +- `k8s/kustomization.yaml`, any `mrms_*` resource + +**Step 1: Check** + +```bash +grep -r "mrms\|Mrms" k8s/ +``` + +**Step 2: If found, remove. Otherwise skip.** + +**Step 3: Commit if changed.** + +--- + +## Stream A: Port f00 to Rust + +**Context:** Today, Elixir's `PropagationGridWorker.perform/1` handles f00 only — the analysis-hour step that does three things Rust doesn't: native-level HRRR duct merge (the ~530 MB native GRIB2), NEXRAD composite reflectivity overlay, and commercial-link degradation lookup. It also writes the `ProfilesFile` that `/weather` reads. Forecast hours f01–f18 already run in Rust via the `grid_tasks` queue. + +**Strategy:** Add a `kind` column to `grid_tasks` so the same queue serves both forecast hours (`kind='forecast'`) and analysis-hour enrichment (`kind='analysis'`). Rust's worker dispatches on `kind`. Elixir's seeder switches from `Oban.insert_all` for f00 to seeding a `kind='analysis'` row alongside the 18 `kind='forecast'` rows. When the analysis task completes, Rust writes the same `/data/scores//.ntms` + `/data/profiles/.ntms` files and NOTIFYs `propagation_ready` with the existing payload. + +### Task A1: Add `kind` column to grid_tasks + +**Files:** +- Create: `priv/repo/migrations/YYYYMMDDHHMMSS_add_kind_to_grid_tasks.exs` + +**Step 1: Generate migration** + +```bash +mix ecto.gen.migration add_kind_to_grid_tasks +``` + +**Step 2: Write the migration** + +```elixir +defmodule Microwaveprop.Repo.Migrations.AddKindToGridTasks do + use Ecto.Migration + + def change do + alter table(:grid_tasks) do + add :kind, :string, null: false, default: "forecast" + end + + create index(:grid_tasks, [:status, :kind]) + + # Replace the old unique index with one that includes kind so an + # analysis task and a forecast task for the same (run_time, f00) + # can coexist without conflict. + drop_if_exists unique_index(:grid_tasks, [:run_time, :forecast_hour]) + create unique_index(:grid_tasks, [:run_time, :forecast_hour, :kind]) + end +end +``` + +**Step 3: Run migration** + +```bash +mix ecto.migrate +``` + +**Step 4: Commit** + +```bash +git add priv/repo/migrations/ +git commit -m "feat(grid_tasks): add kind column to separate forecast vs analysis work" +``` + +### Task A2: Rust reads `kind` and routes + +**Files:** +- Modify: `rust/prop_grid_rs/src/db.rs` — add `kind: String` to the task struct, select it in `claim_next` +- Modify: `rust/prop_grid_rs/src/bin/worker.rs` — match on `kind` and dispatch to forecast vs. analysis pipeline +- Test: `rust/prop_grid_rs/src/db.rs` test for `claim_next` returning the `kind` + +**Step 1: Write failing test** + +In `rust/prop_grid_rs/src/db.rs`: + +```rust +#[tokio::test] +async fn claim_next_returns_kind() { + // seed one task with kind='analysis', claim it, assert kind round-trips + // ... +} +``` + +**Step 2: Run test** + +```bash +cd rust/prop_grid_rs && cargo test claim_next_returns_kind +``` + +Expected: FAIL (field doesn't exist). + +**Step 3: Add `kind` field to the claim struct + SELECT list** + +Minimal change to `claim_next` to read and return `kind`. Default to `"forecast"` if unmapped so the compile stays green during rollout. + +**Step 4: Run test** + +Expected: PASS. + +**Step 5: Commit** + +```bash +git add rust/prop_grid_rs/src/db.rs +git commit -m "feat(prop-grid-rs): surface grid_tasks.kind in claim_next" +``` + +### Task A3: Port `HrrrNativeClient.fetch_native_duct_grid` to Rust + +This is the biggest subtask — the 530 MB GRIB2 native-level pull + duct-metric computation. + +**Files:** +- Create: `rust/prop_grid_rs/src/native_duct.rs` +- Test: `rust/prop_grid_rs/src/native_duct.rs` (inline `#[cfg(test)] mod tests`) + +**Step 1: Write golden-fixture test** + +Follow the same pattern Phase 1 used for surface/pressure: a Mix task dumps the Elixir-computed duct grid for a known `(date, hour)` to a bincode file in `rust/prop_grid_rs/tests/golden/`. Rust loads it and asserts exact match on every `{lat, lon}` key and per-field values. + +Create `lib/mix/tasks/rust.golden_duct.ex`: + +```elixir +defmodule Mix.Tasks.Rust.GoldenDuct do + use Mix.Task + + @shortdoc "Dump Elixir-computed native duct grid for Rust comparison" + + def run([date_str, hour_str]) do + Mix.Task.run("app.start") + date = Date.from_iso8601!(date_str) + hour = String.to_integer(hour_str) + grid_spec = Microwaveprop.Propagation.Grid.wgrib2_grid_spec() + + {:ok, duct_grid} = + Microwaveprop.Weather.HrrrNativeClient.fetch_native_duct_grid(date, hour, grid_spec, 0) + + path = "rust/prop_grid_rs/tests/golden/duct_#{date_str}_t#{hour_str}z.bincode" + File.mkdir_p!(Path.dirname(path)) + File.write!(path, :erlang.term_to_binary(duct_grid)) + IO.puts("Wrote #{map_size(duct_grid)} cells to #{path}") + end +end +``` + +**Step 2: Run the golden dump against a recent hour** + +```bash +mix rust.golden_duct 2026-04-19 14 +``` + +Expected: writes a bincode file with ~92k cells. + +**Step 3: Write the Rust failing test** + +In `rust/prop_grid_rs/src/native_duct.rs`: + +```rust +#[tokio::test] +async fn matches_elixir_golden() { + let golden = load_golden("tests/golden/duct_2026-04-19_t14z.bincode"); + let rust = fetch_native_duct_grid(Date::from_ymd(2026, 4, 19), 14, grid_spec(), 0).await.unwrap(); + + assert_eq!(rust.len(), golden.len()); + for (pt, expected) in &golden { + let got = rust.get(pt).expect(&format!("missing {:?}", pt)); + assert!((got.native_min_gradient - expected.native_min_gradient).abs() < 1e-3); + assert_eq!(got.duct_count, expected.duct_count); + // ... other fields + } +} +``` + +**Step 4: Run test** + +```bash +cargo test matches_elixir_golden +``` + +Expected: FAIL — `fetch_native_duct_grid` doesn't exist. + +**Step 5: Implement the Rust port** + +Port these pieces in order, one commit each: + +1. **Byte-range fetch for duct variables.** Port `duct_byte_ranges/1` (selects TMP/SPFH/HGT/PRES on all 50 hybrid levels from the idx). Reuse the existing `fetcher::download_ranges_to_file` that Phase 1 built. +2. **wgrib2 subprocess for `extract_grid_from_file_mapped`.** Port the `-lola` + match-pattern invocation. Parse the flat output into a `HashMap<(lat, lon), HashMap>`. Stream cell-by-cell rather than buffering all values — matches Elixir's "peak memory ~86 MB" design. +3. **`compute_duct_metrics` per cell.** Port `build_native_profile/1`, `min_m_gradient/1`, and `Duct.analyze/1`. `Duct.analyze/1` is in `lib/microwaveprop/propagation/duct.ex` — that's its own port. +4. **Wire into `native_duct.rs::fetch_native_duct_grid`** as the public entry point. + +**Step 6: Run test to verify it passes** + +```bash +cargo test matches_elixir_golden +``` + +Expected: PASS with max-per-cell delta under 1e-3. + +**Step 7: Commit** + +```bash +git add rust/prop_grid_rs/src/native_duct.rs rust/prop_grid_rs/tests/golden/ lib/mix/tasks/rust.golden_duct.ex +git commit -m "feat(prop-grid-rs): port HrrrNativeClient.fetch_native_duct_grid" +``` + +### Task A4: Port NEXRAD composite reflectivity fetch + +**Context:** `NexradClient.fetch_frame/2` (in `lib/microwaveprop/weather/nexrad_client.ex`) pulls the current MRMS composite reflectivity mosaic and returns `[%{lat, lon, max_reflectivity_dbz}]`. The merge into `grid_data` is line-level (`apply_nexrad_observations`). It's another GRIB2 decode but much smaller than the native-level file. + +**Files:** +- Create: `rust/prop_grid_rs/src/nexrad.rs` +- Test: inline fixture test + +**Step 1: Golden fixture** + +Same pattern: `mix rust.golden_nexrad 2026-04-19T14:00:00Z` dumps the Elixir observations list as bincode. + +**Step 2: Write failing test** + +```rust +#[tokio::test] +async fn matches_elixir_golden() { /* ... */ } +``` + +**Step 3: Implement `fetch_frame`** by porting the HTTP GET + GRIB2 decode from `NexradClient`. + +**Step 4: Run test.** Expected PASS. + +**Step 5: Commit.** + +### Task A5: Port commercial-link degradation lookup + +**Context:** `Commercial.build_link_lookup/1` (in `lib/microwaveprop/commercial.ex`) returns a `[{{lat, lon}, degradation}]` list from Postgres. `link_degradation_from_lookup/2` is a pure haversine match. Only 7 links cluster around DFW so most cells return nil. + +**Files:** +- Create: `rust/prop_grid_rs/src/commercial.rs` +- Test: inline SQL + haversine round-trip + +**Step 1: Golden fixture (simpler — just dump the 7 link rows)** + +**Step 2: Failing test** + +```rust +#[tokio::test] +async fn haversine_matches_elixir() { /* 7 links × 10 sample points */ } +``` + +**Step 3: Port `build_link_lookup` (one SELECT) + haversine matcher.** + +**Step 4: Commit.** + +### Task A6: Port ProfilesFile write + +**Context:** `Microwaveprop.Propagation.ProfilesFile.write!/2` serializes the fully-enriched `grid_data` to `/data/profiles/.ntms` using `:erlang.term_to_binary/1`. This is the tricky one — the file format is Erlang ETF. + +**Decision point:** Two options. + +1. **Port the reader to Rust + ETF encoder.** The `eetf` crate exists but is not heavily maintained. Risk: ETF subtleties (ref counting, atom tables) could produce bit-different output that breaks Elixir's reader. +2. **Change the format.** Replace ETF with a simple framed bincode/cbor/msgpack file. Rust writes, Elixir reads via a new decoder. Clean break. + +**Recommendation: Option 2.** The profiles file is read from exactly one place (`/weather` and point-detail LiveView in `lib/microwaveprop/weather.ex`). Porting the reader to a new format is ~50 lines. Fighting with eetf is a tarpit. + +**Files:** +- Create: `rust/prop_grid_rs/src/profiles_file.rs` (writes new format) +- Modify: `lib/microwaveprop/propagation/profiles_file.ex` (reads old format; add reader for new format; write-path deleted once Rust owns f00) +- Modify: `lib/microwaveprop/weather.ex` (call sites that read profiles) + +**Step 1: Design the format.** Header: magic `PROF\x00`, version u8, record count u32, `{lat: f32, lon: f32, field_count: u8, [{key: varint-len-prefixed string, value: tagged-union}]...}`. Or — simpler — MessagePack. `rmp-serde` in Rust, `Msgpax` in Elixir. Both are well-maintained. + +**Step 2: Write Elixir reader for MessagePack variant.** Keep the old ETF reader around under a `read_etf!/1` alias for backward compatibility during rollout. + +**Step 3: Failing test in Rust.** Round-trip: write a sample `grid_data`, read it back, assert equal. + +**Step 4: Implement writer.** + +**Step 5: Dual-read test in Elixir.** Old ETF files still read; new MessagePack files read identically. + +**Step 6: Commit.** + +### Task A7: Wire analysis pipeline in Rust worker + +**Files:** +- Modify: `rust/prop_grid_rs/src/pipeline.rs` — add `run_analysis_task` alongside the existing forecast pipeline +- Modify: `rust/prop_grid_rs/src/bin/worker.rs` — dispatch on `task.kind` + +**Step 1: Write failing integration test** + +In `rust/prop_grid_rs/tests/analysis_pipeline_test.rs`: + +```rust +#[tokio::test] +async fn analysis_task_writes_scores_and_profiles() { + // seed one kind='analysis' task, run pipeline, assert + // /data/scores/10000/.ntms exists + // /data/profiles/.ntms exists + // NOTIFY propagation_ready fired +} +``` + +**Step 2: Run test.** Expected FAIL. + +**Step 3: Implement `run_analysis_task`.** Same shape as forecast pipeline but: +- Calls `native_duct::fetch_native_duct_grid` after the regular surface+pressure merge. +- Calls `nexrad::fetch_frame` and merges. +- Calls `commercial::build_link_lookup` and merges. +- Writes `ProfilesFile` to `/data/profiles/.ntms`. +- Scores all bands as before. + +**Step 4: Run test.** Expected PASS. + +**Step 5: Commit.** + +### Task A8: Flip Elixir seeder to enqueue analysis as grid_task + +**Files:** +- Modify: `lib/microwaveprop/workers/propagation_grid_worker.ex:80-99` (seed_chain) +- Modify: `lib/microwaveprop/propagation/grid_task_enqueuer.ex` (add `seed_with_analysis/1`) + +**Step 1: Write failing test** + +In `test/microwaveprop/workers/propagation_grid_worker_test.exs`, update the seed test to assert: +- **Zero** Oban jobs enqueued (f00 is no longer an Oban job). +- **19** grid_tasks rows: 1 with `kind='analysis' AND forecast_hour=0`, 18 with `kind='forecast' AND forecast_hour IN (1..18)`. + +**Step 2: Run test.** Expected FAIL. + +**Step 3: Modify `seed_chain` to call `GridTaskEnqueuer.seed_with_analysis(run_time)` and return `:ok`.** Delete the `Oban.insert_all` line. Delete `run_chain_step`, `process_forecast_hour`, and all the merge helpers from `propagation_grid_worker.ex`. The module becomes a thin cron handler. + +**Step 4: Run test.** Expected PASS. + +**Step 5: Delete now-unused Elixir modules** after `mix xref callers` confirms no callers remain: +- `lib/microwaveprop/weather/hrrr_native_client.ex` +- `lib/microwaveprop/weather/nexrad_client.ex` (assuming no other caller) +- The `merge_native_duct_data`, `merge_nexrad_data`, `merge_commercial_link_data`, `compute_scores_*` helpers + +**Step 6: Commit.** + +### Task A9: Shrink hot pod memory limit + +**Files:** +- Modify: `k8s/deployment.yaml` + +**Step 1: Wait for 24h of clean post-cutover runs.** + +**Step 2: Reduce memory limit from 2Gi to 1Gi, requests from 512Mi to 256Mi.** + +**Step 3: Watch `kubectl -n prop top pod` for OOM signals over 24h.** If any pod OOMKills, revert immediately and investigate before retrying. + +**Step 4: Commit.** + +--- + +## Stream C: Port HrrrFetchWorker (optional polish) + +**Context:** Each QSO submission enqueues an `HrrrFetchWorker` job that fetches a HRRR profile at the contact location. Per-job payload is small (one point, ~100 kB of GRIB2 bytes) but the job calls `HrrrClient` which on BEAM holds refc binaries that the worker explicitly `:erlang.garbage_collect`s at the end. That GC is a signal the binary pressure is real. + +**Strategy:** New Rust binary `hrrr-point-worker` shares the existing `prop_grid_rs` crate (adds a second bin). A new `hrrr_point_tasks` table mirrors `grid_tasks`. Elixir's ingestion switches from `HrrrFetchWorker` (Oban) to inserting into `hrrr_point_tasks`. + +### Task C1: Migration for `hrrr_point_tasks` + +**Files:** +- Create: `priv/repo/migrations/YYYYMMDDHHMMSS_create_hrrr_point_tasks.exs` + +**Columns:** `id uuid PK`, `qso_id uuid`, `valid_time timestamp`, `lat double precision`, `lon double precision`, `status text`, `claimed_at`, `completed_at`, `error text`, `attempt smallint`. Unique on `(qso_id, valid_time)`. + +**Step 1: Generate + write migration.** +**Step 2: Run migration.** +**Step 3: Commit.** + +### Task C2: Rust `hrrr-point-worker` binary + +**Files:** +- Create: `rust/prop_grid_rs/src/bin/hrrr_point_worker.rs` +- Create: `rust/prop_grid_rs/src/point_fetcher.rs` (per-point HRRR profile extract) + +**Step 1: Golden fixture for HrrrClient.fetch_profile.** +**Step 2: Failing test.** +**Step 3: Implement point fetcher + worker loop.** Reuse the existing `fetcher` module for byte-range downloads; add a new `extract_point_profile` that takes a single `(lat, lon)` instead of a grid spec. +**Step 4: Commit.** + +### Task C3: k8s deployment for hrrr-point-worker + +**Files:** +- Create: `k8s/deployment-hrrr-point-rs.yaml` (same image, different ENTRYPOINT) +- Modify: `k8s/kustomization.yaml` + +**Step 1: Write manifest. 1 replica on talos5 initially, 256 Mi memory.** +**Step 2: Apply.** +**Step 3: Commit.** + +### Task C4: Switch Elixir ingestion to enqueue into `hrrr_point_tasks` + +**Files:** +- Modify: wherever `HrrrFetchWorker.new(...)` is called (check `lib/microwaveprop/qsos.ex` or `lib/microwaveprop/radio.ex` via `grep -rn "HrrrFetchWorker.new"`) +- Delete: `lib/microwaveprop/workers/hrrr_fetch_worker.ex` (after `mix xref callers` is clean) +- Delete: corresponding test + +**Step 1: Failing test.** QSO submission inserts a row into `hrrr_point_tasks` and does NOT enqueue an Oban job. +**Step 2: Implement.** +**Step 3: Commit.** + +--- + +## Critical Files Reference + +**Ported out (deleted at end of Stream A):** +- `lib/microwaveprop/weather/hrrr_native_client.ex` +- `lib/microwaveprop/weather/nexrad_client.ex` (confirm no per-QSO callers first) +- `lib/microwaveprop/weather/mrms_client.ex` (Stream B) +- `lib/microwaveprop/weather/mrms_cache.ex` (Stream B) +- `lib/microwaveprop/workers/mrms_fetch_worker.ex` (Stream B) +- `lib/microwaveprop/workers/hrrr_fetch_worker.ex` (Stream C) +- Most of `lib/microwaveprop/workers/propagation_grid_worker.ex` — becomes a thin seeder + +**Kept as-is:** +- `lib/microwaveprop/commercial.ex` — Elixir still needs this for the SNMP poller and LiveView sensor widget. Rust just adds a parallel reader over the same `commercial_samples` table. +- `lib/microwaveprop/propagation/duct.ex` — ported to Rust, original stays because other callers (per-contact analysis) still use it. + +**New Rust modules (Stream A):** +- `rust/prop_grid_rs/src/native_duct.rs` +- `rust/prop_grid_rs/src/nexrad.rs` +- `rust/prop_grid_rs/src/commercial.rs` +- `rust/prop_grid_rs/src/profiles_file.rs` +- `rust/prop_grid_rs/src/duct.rs` (ported from `lib/microwaveprop/propagation/duct.ex`) + +**New Rust module (Stream C):** +- `rust/prop_grid_rs/src/point_fetcher.rs` +- `rust/prop_grid_rs/src/bin/hrrr_point_worker.rs` + +--- + +## Verification + +**Stream B:** +- `grep -r "Mrms" lib/ test/ config/` returns nothing. +- `mix precommit` green. +- 24h later: `kubectl -n prop get events` shows no MrmsFetchWorker executions. + +**Stream A:** +- `cargo test -p prop_grid_rs` green with golden-duct, golden-nexrad, golden-commercial fixtures. +- `mix test` green. +- One full post-cutover hourly cycle shows `kind='analysis'` task `status='done'` in `grid_tasks`, `/data/profiles/.ntms` written, `/weather` page loads correctly. +- Hot pod `beam_memory_total` < 700 MiB peak over 24 h (down from ~1 GiB under Phase 2). +- Hot pod memory limit reduced to 1 Gi with zero OOMKilled events over 7 days. + +**Stream C:** +- QSO submission inserts a `hrrr_point_tasks` row within 1 s. +- Rust `hrrr-point-worker` logs show claim→complete within 30 s of insert. +- HRRR profile appears on the QSO detail page within 1 min. +- `:hrrr` Oban queue is empty and can be removed from the Oban config. + +--- + +## Rollback Plans + +**Stream B:** `git revert` restores MRMS. Supervision tree re-adds MrmsCache on next deploy. No data loss — MrmsCache was in-memory only. + +**Stream A:** `git revert` restores the Elixir f00 chain step. The `grid_tasks.kind` column can stay (default `'forecast'` keeps existing Rust behaviour). Bump hot pod memory back to 2Gi if Phase 3 shrink was already deployed. + +**Stream C:** `git revert` restores `HrrrFetchWorker`. `hrrr_point_tasks` rows become orphan; a one-off Mix task migrates them back to Oban jobs. diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 7e7d7eb6..2dc2acdf 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -22,7 +22,6 @@ defmodule Microwaveprop.Application do Microwaveprop.Weather.GridCache, {Microwaveprop.Weather.IemRateLimiter, interval_ms: Application.get_env(:microwaveprop, :iem_rate_limiter_interval_ms, 700)}, - Microwaveprop.Weather.MrmsCache, Microwaveprop.Weather.NexradCache, Microwaveprop.Qrz.Client, {Oban, Application.fetch_env!(:microwaveprop, Oban)}, diff --git a/lib/microwaveprop/weather/mrms_cache.ex b/lib/microwaveprop/weather/mrms_cache.ex deleted file mode 100644 index a144c737..00000000 --- a/lib/microwaveprop/weather/mrms_cache.ex +++ /dev/null @@ -1,80 +0,0 @@ -defmodule Microwaveprop.Weather.MrmsCache do - @moduledoc """ - Node-local ETS cache of the latest MRMS PrecipRate grid, regridded onto - the 0.125° propagation grid. Mirrors the pattern used by - `Microwaveprop.Propagation.ScoreCache` and - `Microwaveprop.Weather.GridCache` — a single GenServer owns the table, - callers read directly from ETS, and `broadcast_put/2` fans updates out - to peer nodes via PubSub. - - Only one node runs `Microwaveprop.Workers.MrmsFetchWorker` per cron - tick (Oban Pro Smart engine picks a leader), so the other nodes get - the grid via the `"mrms:cache"` topic and stay in sync without each - one paying the 1 MB fetch + wgrib2 regrid cost. - """ - use GenServer - - alias Phoenix.PubSub - - @table :mrms_cache - @topic "mrms:cache" - @pubsub Microwaveprop.PubSub - - @type rain_grid :: %{{float(), float()} => float()} - - @spec start_link(keyword()) :: GenServer.on_start() - def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) - - @doc "Return the cached `{valid_time, rain_grid}` or `:miss`." - @spec fetch() :: {:ok, DateTime.t(), rain_grid()} | :miss - def fetch do - case :ets.lookup(@table, :current) do - [{_, valid_time, grid}] -> {:ok, valid_time, grid} - [] -> :miss - end - end - - @doc "Current MRMS valid_time if anything is cached." - @spec valid_time() :: DateTime.t() | nil - def valid_time do - case fetch() do - {:ok, vt, _} -> vt - :miss -> nil - end - end - - @spec put(DateTime.t(), rain_grid()) :: :ok - def put(valid_time, grid) do - :ets.insert(@table, {:current, valid_time, grid}) - :ok - end - - @doc "Insert locally and broadcast to peer nodes." - @spec broadcast_put(DateTime.t(), rain_grid()) :: :ok - def broadcast_put(valid_time, grid) do - put(valid_time, grid) - PubSub.broadcast(@pubsub, @topic, {:mrms_cache_refresh, valid_time, grid}) - :ok - end - - @spec clear() :: :ok - def clear do - :ets.delete_all_objects(@table) - :ok - end - - @impl true - def init(_opts) do - :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true]) - PubSub.subscribe(@pubsub, @topic) - {:ok, %{}} - end - - @impl true - def handle_info({:mrms_cache_refresh, valid_time, grid}, state) do - put(valid_time, grid) - {:noreply, state} - end - - def handle_info(_msg, state), do: {:noreply, state} -end diff --git a/lib/microwaveprop/weather/mrms_client.ex b/lib/microwaveprop/weather/mrms_client.ex deleted file mode 100644 index 553bcfde..00000000 --- a/lib/microwaveprop/weather/mrms_client.ex +++ /dev/null @@ -1,190 +0,0 @@ -defmodule Microwaveprop.Weather.MrmsClient do - @moduledoc """ - Fetch NOAA MRMS `PrecipRate` surface rain-rate mosaics and interpolate - them onto the same 0.125° propagation grid we score against. MRMS is - ~1 km resolution and updates every 2 minutes, so between hourly HRRR runs - it's the freshest rain signal we can get over CONUS. - - The module is pure transport + decode: fetch, gunzip, run wgrib2 to - regrid, return `%{{lat, lon} => rain_rate_mmhr}`. Caching and cron - scheduling live in `Microwaveprop.Weather.MrmsCache` and - `Microwaveprop.Workers.MrmsFetchWorker`. - """ - - alias Microwaveprop.Propagation.Grid - alias Microwaveprop.Weather.Grib2.Wgrib2 - - require Logger - - @base_url "https://mrms.ncep.noaa.gov/2D/PrecipRate/" - @match_pattern ":PrecipRate:" - @missing_sentinel -3.0 - - @type rain_grid :: %{{float(), float()} => float()} - @type listing_entry :: %{filename: String.t(), valid_time: DateTime.t()} - - @doc """ - List the most recent MRMS PrecipRate files published on the NCEP index. - Returns entries sorted newest-first so callers can just take the head. - """ - @spec list_latest(non_neg_integer()) :: {:ok, [listing_entry()]} | {:error, term()} - def list_latest(limit \\ 10) do - Microwaveprop.Instrument.span([:mrms, :list_latest], %{}, fn -> do_list_latest(limit) end) - end - - defp do_list_latest(limit) do - case Req.get(@base_url, req_options()) do - {:ok, %{status: 200, body: body}} -> - entries = - body - |> parse_listing() - |> Enum.sort_by(& &1.valid_time, {:desc, DateTime}) - |> Enum.take(limit) - - {:ok, entries} - - {:ok, %{status: status}} -> - {:error, {:http_status, status}} - - {:error, reason} -> - {:error, reason} - end - end - - @doc """ - Fetch a single MRMS file by filename, decompress, regrid to the 0.125° - propagation grid, and return `{valid_time, %{{lat, lon} => rain_rate_mmhr}}`. - - Missing-data sentinels (-3) are dropped from the returned grid so callers - can test membership directly. - """ - @spec fetch_and_decode(String.t()) :: {:ok, DateTime.t(), rain_grid()} | {:error, term()} - def fetch_and_decode(filename) do - with {:ok, valid_time} <- parse_filename(filename), - {:ok, grib_bytes} <- download(filename), - {:ok, tmp_path} <- write_temp(grib_bytes), - {:ok, grid} <- regrid(tmp_path) do - File.rm(tmp_path) - {:ok, valid_time, grid} - end - end - - @doc """ - Convenience: list the latest files, grab the newest that isn't already - cached, and return the decoded grid. `known_valid_time` is whatever the - caller already has — if the latest file matches it we short-circuit. - """ - @spec fetch_latest(DateTime.t() | nil) :: - {:ok, DateTime.t(), rain_grid()} | {:up_to_date, DateTime.t()} | {:error, term()} - def fetch_latest(known_valid_time \\ nil) do - case list_latest(1) do - {:ok, [latest | _]} -> - if known_valid_time && DateTime.compare(latest.valid_time, known_valid_time) != :gt do - {:up_to_date, known_valid_time} - else - fetch_and_decode(latest.filename) - end - - {:ok, []} -> - {:error, :no_files_listed} - - other -> - other - end - end - - # -- internals (public-but-undocumented for tests) -------------------- - - @doc false - def parse_listing(html) do - ~r/MRMS_PrecipRate_00\.00_(\d{8}-\d{6})\.grib2\.gz/ - |> Regex.scan(html) - |> Enum.uniq() - |> Enum.flat_map(fn [full, stamp] -> - case parse_stamp(stamp) do - {:ok, dt} -> [%{filename: full, valid_time: dt}] - _ -> [] - end - end) - end - - defp parse_filename(filename) do - case Regex.run(~r/MRMS_PrecipRate_00\.00_(\d{8}-\d{6})\.grib2\.gz/, filename) do - [_, stamp] -> parse_stamp(stamp) - _ -> {:error, :bad_filename} - end - end - - defp parse_stamp(stamp) do - with <> <- - stamp, - {:ok, naive} <- - NaiveDateTime.new(si(y), si(m), si(d), si(h), si(mi), si(s)) do - {:ok, DateTime.from_naive!(naive, "Etc/UTC")} - else - _ -> {:error, :bad_stamp} - end - end - - defp si(s), do: String.to_integer(s) - - defp download(filename) do - # Req auto-decompresses gzip responses, so the body we receive is - # already the raw GRIB2 binary (starts with "GRIB" magic) even though - # the URL ends in .gz. No gunzip step needed. - url = @base_url <> filename - - Microwaveprop.Instrument.span([:mrms, :download], %{}, fn -> - case Req.get(url, req_options()) do - {:ok, %{status: 200, body: body}} when is_binary(body) -> - {:ok, body} - - {:ok, %{status: status}} -> - {:error, {:http_status, status}} - - {:error, reason} -> - {:error, reason} - end - end) - end - - defp write_temp(grib_bytes) do - path = Path.join(System.tmp_dir!(), "mrms_preciprate_#{System.unique_integer([:positive])}.grib2") - File.write!(path, grib_bytes) - {:ok, path} - end - - defp regrid(tmp_path) do - case Wgrib2.extract_grid_from_file(tmp_path, @match_pattern, Grid.wgrib2_grid_spec()) do - {:ok, cells} -> {:ok, flatten_rain_grid(cells)} - error -> error - end - end - - @doc false - def flatten_rain_grid(cells) do - Enum.reduce(cells, %{}, fn {{lat, lon}, fields}, acc -> - case extract_preciprate(fields) do - nil -> acc - rate -> Map.put(acc, {lat, lon}, rate) - end - end) - end - - defp extract_preciprate(fields) do - Enum.find_value(fields, fn - {"PrecipRate" <> _, value} when is_number(value) and value > @missing_sentinel and value >= 0 -> - value - - _ -> - nil - end) - end - - defp req_options do - Keyword.merge( - [receive_timeout: 120_000, retry: :safe_transient, max_retries: 3], - Application.get_env(:microwaveprop, :mrms_req_options, []) - ) - end -end diff --git a/lib/microwaveprop/workers/mrms_fetch_worker.ex b/lib/microwaveprop/workers/mrms_fetch_worker.ex deleted file mode 100644 index ab42fa7e..00000000 --- a/lib/microwaveprop/workers/mrms_fetch_worker.ex +++ /dev/null @@ -1,46 +0,0 @@ -defmodule Microwaveprop.Workers.MrmsFetchWorker do - @moduledoc """ - Every ~2 minutes, pull the newest NOAA MRMS PrecipRate grid and cache - the regridded version (0.125° CONUS) in `Microwaveprop.Weather.MrmsCache`. - `Microwaveprop.Workers.AsosAdjustmentWorker` reads that cache on its - own tick to overlay real radar rain rates onto the score grid. - - Runs on the `propagation` queue so it shares the scoring node's crontab - and isn't competing with the heavier HRRR pipeline for slots. - """ - # Lower priority than PropagationGridWorker so an MRMS backlog - # (cron fires every 2 minutes) doesn't starve the hourly chain - # on the shared :propagation queue. - use Oban.Worker, queue: :propagation, priority: 5, max_attempts: 2 - - alias Microwaveprop.Weather.MrmsCache - alias Microwaveprop.Weather.MrmsClient - - require Logger - - @impl Oban.Worker - def perform(%Oban.Job{}) do - Microwaveprop.Instrument.span([:worker, :mrms_fetch], %{}, fn -> - case MrmsClient.fetch_latest(MrmsCache.valid_time()) do - {:ok, valid_time, grid} -> - MrmsCache.broadcast_put(valid_time, grid) - - Logger.info("MrmsFetch: cached #{map_size(grid)} cells at #{valid_time} (#{count_rainy(grid)} with rain)") - - :ok - - {:up_to_date, valid_time} -> - Logger.debug("MrmsFetch: cache already has #{valid_time}") - :ok - - {:error, reason} -> - Logger.warning("MrmsFetch: skipped — #{inspect(reason)}") - :ok - end - end) - end - - defp count_rainy(grid) do - Enum.count(grid, fn {_, v} -> v > 0 end) - end -end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index d26d2af3..057c4a7c 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -20,11 +20,10 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do use Oban.Worker, queue: :propagation, - # Highest priority on the shared :propagation queue. MrmsFetchWorker - # and PropagationPruneWorker also live here; a backlog of those - # (e.g. after a deploy, or MRMS firing every 2 minutes) would - # otherwise starve the hourly chain because Oban dispatches - # priority-then-FIFO. Explicit so the invariant is visible. + # Highest priority on the shared :propagation queue. PropagationPruneWorker + # also lives here; a backlog would otherwise starve the hourly chain + # because Oban dispatches priority-then-FIFO. Explicit so the invariant + # is visible. priority: 0, # Higher than the default 3 so a few DynamicLifeline rescues # (e.g., a rolling deploy that kills a mid-flight chain step) diff --git a/test/microwaveprop/weather/mrms_cache_test.exs b/test/microwaveprop/weather/mrms_cache_test.exs deleted file mode 100644 index 905d9106..00000000 --- a/test/microwaveprop/weather/mrms_cache_test.exs +++ /dev/null @@ -1,97 +0,0 @@ -defmodule Microwaveprop.Weather.MrmsCacheTest do - # async: false — MrmsCache is started at the application level and owns a - # named ETS table. Tests mutate shared state and must not race. - use ExUnit.Case, async: false - - alias Microwaveprop.Weather.MrmsCache - - @topic "mrms:cache" - @pubsub Microwaveprop.PubSub - - setup do - MrmsCache.clear() - :ok - end - - describe "fetch/0" do - test "returns :miss when nothing is cached" do - assert MrmsCache.fetch() == :miss - end - - test "returns {:ok, valid_time, grid} after put/2" do - valid_time = ~U[2026-04-12 19:24:00Z] - grid = %{{32.0, -97.0} => 1.2, {32.125, -97.0} => 0.0} - - assert :ok = MrmsCache.put(valid_time, grid) - assert {:ok, ^valid_time, ^grid} = MrmsCache.fetch() - end - - test "put/2 overwrites the single current entry" do - first_vt = ~U[2026-04-12 19:24:00Z] - second_vt = ~U[2026-04-12 19:26:00Z] - - MrmsCache.put(first_vt, %{{32.0, -97.0} => 1.0}) - MrmsCache.put(second_vt, %{{32.0, -97.0} => 2.0}) - - assert {:ok, ^second_vt, %{{32.0, -97.0} => 2.0}} = MrmsCache.fetch() - end - end - - describe "valid_time/0" do - test "returns nil when cache is empty" do - assert MrmsCache.valid_time() == nil - end - - test "returns the cached valid_time when present" do - vt = ~U[2026-04-12 19:24:00Z] - MrmsCache.put(vt, %{{32.0, -97.0} => 5.4}) - - assert MrmsCache.valid_time() == vt - end - end - - describe "broadcast_put/2" do - test "inserts locally and broadcasts on the mrms:cache topic" do - Phoenix.PubSub.subscribe(@pubsub, @topic) - - vt = ~U[2026-04-12 19:28:00Z] - grid = %{{32.0, -97.0} => 3.3} - - assert :ok = MrmsCache.broadcast_put(vt, grid) - assert {:ok, ^vt, ^grid} = MrmsCache.fetch() - - # The cache subscribes to its own topic, so we should observe the - # message via our own subscription too. - assert_receive {:mrms_cache_refresh, ^vt, ^grid} - end - end - - describe "clear/0" do - test "empties the cache" do - MrmsCache.put(~U[2026-04-12 19:24:00Z], %{{32.0, -97.0} => 1.0}) - assert {:ok, _, _} = MrmsCache.fetch() - - assert :ok = MrmsCache.clear() - assert MrmsCache.fetch() == :miss - assert MrmsCache.valid_time() == nil - end - end - - describe "peer broadcast handling" do - test "applies refreshes received from peers on the PubSub topic" do - # Simulate a peer node pushing a refresh. The cache GenServer is - # subscribed to @topic in init/1; broadcasting directly exercises the - # handle_info({:mrms_cache_refresh, ...}) clause. - vt = ~U[2026-04-12 19:30:00Z] - grid = %{{32.0, -97.0} => 7.7} - - Phoenix.PubSub.broadcast(@pubsub, @topic, {:mrms_cache_refresh, vt, grid}) - - # Round-trip through the MrmsCache GenServer mailbox: send it a sync - # call so we know the previous broadcast has been processed. - _ = :sys.get_state(MrmsCache) - - assert {:ok, ^vt, ^grid} = MrmsCache.fetch() - end - end -end diff --git a/test/microwaveprop/weather/mrms_client_test.exs b/test/microwaveprop/weather/mrms_client_test.exs deleted file mode 100644 index 4be8e536..00000000 --- a/test/microwaveprop/weather/mrms_client_test.exs +++ /dev/null @@ -1,72 +0,0 @@ -defmodule Microwaveprop.Weather.MrmsClientTest do - use ExUnit.Case, async: true - - alias Microwaveprop.Weather.MrmsClient - - describe "parse_listing/1" do - test "extracts all unique filenames from a listing fragment" do - html = """ - - ... - ... - ... - - """ - - entries = - html - |> MrmsClient.parse_listing() - |> Enum.sort_by(& &1.valid_time, DateTime) - - assert [ - %{valid_time: ~U[2026-04-12 19:24:00Z]}, - %{valid_time: ~U[2026-04-12 19:26:00Z]}, - %{valid_time: ~U[2026-04-12 19:28:00Z]} - ] = entries - end - - test "drops filenames with malformed timestamps" do - html = """ - ... - ... - """ - - assert [%{valid_time: ~U[2026-04-12 19:30:00Z]}] = MrmsClient.parse_listing(html) - end - - test "deduplicates identical filenames (directory indexes often list twice)" do - html = """ - file - file - """ - - assert [%{}] = MrmsClient.parse_listing(html) - end - end - - describe "flatten_rain_grid/1" do - test "keeps positive rain rates keyed by lat/lon" do - cells = %{ - {32.0, -97.0} => %{"PrecipRate:0 m above mean sea level" => 5.4}, - {32.125, -97.0} => %{"PrecipRate:0 m above mean sea level" => 0.0} - } - - assert MrmsClient.flatten_rain_grid(cells) == - %{{32.0, -97.0} => 5.4, {32.125, -97.0} => 0.0} - end - - test "drops missing-value sentinels (-3 or less)" do - cells = %{ - {32.0, -97.0} => %{"PrecipRate:0 m above mean sea level" => -3.0}, - {32.125, -97.0} => %{"PrecipRate:0 m above mean sea level" => 12.5} - } - - assert MrmsClient.flatten_rain_grid(cells) == %{{32.125, -97.0} => 12.5} - end - - test "drops entries with no PrecipRate field at all" do - cells = %{{32.0, -97.0} => %{"OTHER:sfc" => 1.0}} - assert MrmsClient.flatten_rain_grid(cells) == %{} - end - end -end diff --git a/test/microwaveprop/workers/mrms_fetch_worker_test.exs b/test/microwaveprop/workers/mrms_fetch_worker_test.exs deleted file mode 100644 index c1cb4951..00000000 --- a/test/microwaveprop/workers/mrms_fetch_worker_test.exs +++ /dev/null @@ -1,137 +0,0 @@ -defmodule Microwaveprop.Workers.MrmsFetchWorkerTest do - # async: false because the MrmsCache is an application-wide singleton with - # shared ETS state. Concurrent tests would race on cache contents. - use Microwaveprop.DataCase, async: false - use Oban.Testing, repo: Microwaveprop.Repo - - alias Microwaveprop.Weather.MrmsCache - alias Microwaveprop.Weather.MrmsClient - alias Microwaveprop.Workers.MrmsFetchWorker - - @topic "mrms:cache" - @pubsub Microwaveprop.PubSub - - setup do - MrmsCache.clear() - :ok - end - - describe "perform/1" do - test "returns :ok and leaves cache empty when the listing is empty" do - Req.Test.stub(MrmsClient, fn conn -> - # Empty directory index — no filenames matched by the parser. - Plug.Conn.send_resp(conn, 200, "no files here") - end) - - assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}}) - assert MrmsCache.fetch() == :miss - end - - test "returns :ok and leaves cache empty when the listing request 500s" do - Req.Test.stub(MrmsClient, fn conn -> - Plug.Conn.send_resp(conn, 500, "upstream boom") - end) - - assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}}) - assert MrmsCache.fetch() == :miss - end - - test "short-circuits with :up_to_date when cache already has the latest valid_time" do - latest_vt = ~U[2026-04-12 19:28:00Z] - # Pre-populate the cache so fetch_latest/1 sees known_valid_time >= latest. - MrmsCache.put(latest_vt, %{{32.0, -97.0} => 4.2}) - - Req.Test.stub(MrmsClient, fn conn -> - # Only the listing URL should be hit — never a .grib2.gz download. - assert String.ends_with?(conn.request_path, "/PrecipRate/") or - String.ends_with?(conn.request_path, "/PrecipRate") - - html = """ - f - """ - - Plug.Conn.send_resp(conn, 200, html) - end) - - assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}}) - - # Cache is untouched. - assert {:ok, ^latest_vt, %{{32.0, -97.0} => 4.2}} = MrmsCache.fetch() - end - - test "returns :ok and leaves cache empty when the download path 404s" do - # Listing succeeds with one file; download 404s. The worker must log - # and swallow the error rather than crash. - # - # The .grib2.gz URL triggers Req's URL-extension-based gzip decode step - # regardless of response status, so even a 404 body must be valid gzip. - gzipped_404 = :zlib.gzip("gone") - - Req.Test.stub(MrmsClient, fn conn -> - if String.ends_with?(conn.request_path, ".grib2.gz") do - Plug.Conn.send_resp(conn, 404, gzipped_404) - else - html = """ - f - """ - - Plug.Conn.send_resp(conn, 200, html) - end - end) - - assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}}) - assert MrmsCache.fetch() == :miss - end - - # The full happy path — successful listing + download + real GRIB2 - # decode into a populated cache grid — requires a real MRMS PrecipRate - # payload that we don't have a hermetic fixture for. We defer that to - # the wgrib2/GRIB2 coverage task. - # - # What we can still pin down here is the worker's delivery contract - # when both listing and download return 200: if fetch_latest/1 returns - # {:ok, vt, grid}, the worker MUST broadcast_put/2 to the mrms:cache - # topic. We stub garbage gzipped bytes; wgrib2 (when installed) will - # yield an empty grid rather than an error on "no matching messages", - # and the worker broadcasts that through. When wgrib2 is NOT installed - # the client returns :wgrib2_not_available, the worker logs a warning - # and still returns :ok — no broadcast. Both arms are valid behavior - # and both are exercised here. - test "returns :ok on a 200 download (broadcasts iff wgrib2 yields :ok)" do - Phoenix.PubSub.subscribe(@pubsub, @topic) - - # The .grib2.gz URL triggers Req's URL-extension-based gzip decoding, - # so the response body must be real gzip bytes. - gzipped_garbage = :zlib.gzip(:binary.copy(<<0>>, 1024)) - - Req.Test.stub(MrmsClient, fn conn -> - if String.ends_with?(conn.request_path, ".grib2.gz") do - Plug.Conn.send_resp(conn, 200, gzipped_garbage) - else - html = """ - f - """ - - Plug.Conn.send_resp(conn, 200, html) - end - end) - - assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}}) - - # One of two valid outcomes: - # 1. wgrib2 installed → fetch_latest returns {:ok, vt, %{}} → - # worker calls broadcast_put and the subscriber sees a refresh. - # 2. wgrib2 missing → fetch_latest returns {:error, ...} → no - # broadcast, cache unchanged. - receive do - {:mrms_cache_refresh, vt, grid} -> - assert %DateTime{} = vt - assert is_map(grid) - assert {:ok, ^vt, ^grid} = MrmsCache.fetch() - after - 200 -> - assert MrmsCache.fetch() == :miss - end - end - end -end diff --git a/test/microwaveprop/workers/propagation_grid_worker_test.exs b/test/microwaveprop/workers/propagation_grid_worker_test.exs index 2782a239..c71308ea 100644 --- a/test/microwaveprop/workers/propagation_grid_worker_test.exs +++ b/test/microwaveprop/workers/propagation_grid_worker_test.exs @@ -12,7 +12,6 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do use Microwaveprop.DataCase, async: false use Oban.Testing, repo: Microwaveprop.Repo - alias Microwaveprop.Workers.MrmsFetchWorker alias Microwaveprop.Workers.PropagationGridWorker alias Microwaveprop.Workers.PropagationPruneWorker @@ -21,10 +20,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do assert PropagationGridWorker.__opts__()[:priority] == 0 end - test "MrmsFetchWorker and PropagationPruneWorker yield to the grid chain" do + test "PropagationPruneWorker yields to the grid chain" do # Same :propagation queue — must be lower priority so hourly chain - # steps jump ahead of a MRMS or pruner backlog. - assert MrmsFetchWorker.__opts__()[:priority] > 0 + # steps jump ahead of a pruner backlog. assert PropagationPruneWorker.__opts__()[:priority] > 0 end end