chore(mrms): delete pipeline — consumer AsosAdjustmentWorker already disabled

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.
2026-04-19 17:25:51 -05:00 · 2026-04-19 17:25:51 -05:00 · 48708621c5
commit 48708621c5
parent 676999606b
14 changed files with 613 additions and 650 deletions
--- 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},
--- 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}
     ]}
--- 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
--- 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
--- a/docs/plans/2026-04-19-rust-migration-phase3.md
+++ 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/<band>/<iso>.ntms` + `/data/profiles/<iso>.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<String, f32>>`. 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/<iso>.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/<iso>.ntms exists
    //   /data/profiles/<iso>.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/<iso>.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/<iso>.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.
--- 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)},
--- a/lib/microwaveprop/weather/mrms_cache.ex
+++ b/lib/microwaveprop/weather/mrms_cache.ex
@ -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
--- a/lib/microwaveprop/weather/mrms_client.ex
+++ b/lib/microwaveprop/weather/mrms_client.ex
@ -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 <<y::binary-4, m::binary-2, d::binary-2, "-", h::binary-2, mi::binary-2, s::binary-2>> <-
           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
--- a/lib/microwaveprop/workers/mrms_fetch_worker.ex
+++ b/lib/microwaveprop/workers/mrms_fetch_worker.ex
@ -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
--- 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
+    # Highest priority on the shared :propagation queue. PropagationPruneWorker
-    # and PropagationPruneWorker also live here; a backlog of those
+    # also lives here; a backlog would otherwise starve the hourly chain
-    # (e.g. after a deploy, or MRMS firing every 2 minutes) would
+    # because Oban dispatches priority-then-FIFO. Explicit so the invariant
-    # otherwise starve the hourly chain because Oban dispatches
+    # is visible.
    # 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)
--- a/test/microwaveprop/weather/mrms_cache_test.exs
+++ b/test/microwaveprop/weather/mrms_cache_test.exs
@ -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
--- a/test/microwaveprop/weather/mrms_client_test.exs
+++ b/test/microwaveprop/weather/mrms_client_test.exs
@ -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 = """
      <html><body>
      <a href="MRMS_PrecipRate_00.00_20260412-192400.grib2.gz">...</a>
      <a href="MRMS_PrecipRate_00.00_20260412-192600.grib2.gz">...</a>
      <a href="MRMS_PrecipRate_00.00_20260412-192800.grib2.gz">...</a>
      </body></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 = """
      <a href="MRMS_PrecipRate_00.00_bogusstamp.grib2.gz">...</a>
      <a href="MRMS_PrecipRate_00.00_20260412-193000.grib2.gz">...</a>
      """
      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 = """
      <a href="MRMS_PrecipRate_00.00_20260412-193000.grib2.gz">file</a>
      <a href="MRMS_PrecipRate_00.00_20260412-193000.grib2.gz">file</a>
      """
      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
--- a/test/microwaveprop/workers/mrms_fetch_worker_test.exs
+++ b/test/microwaveprop/workers/mrms_fetch_worker_test.exs
@ -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, "<html><body>no files here</body></html>")
      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 = """
        <a href="MRMS_PrecipRate_00.00_20260412-192800.grib2.gz">f</a>
        """
        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 = """
          <a href="MRMS_PrecipRate_00.00_20260412-193000.grib2.gz">f</a>
          """
          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 = """
          <a href="MRMS_PrecipRate_00.00_20260412-193200.grib2.gz">f</a>
          """
          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
--- 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.
+      # steps jump ahead of a pruner backlog.
      assert MrmsFetchWorker.__opts__()[:priority] > 0
      assert PropagationPruneWorker.__opts__()[:priority] > 0
    end
  end