feat(prop): Phase 2 cutover — Rust owns f01–f18

Rust `prop-grid-rs` has been validated end-to-end on talos5 on the
streamed-bands image (main-1776635096-6d91461): real chain steps
complete in ~7 s each and the peak-heap refactor landed.

Changes:
- `PropagationGridWorker.seed_chain` now enqueues a single f00 Oban
  job instead of f00–f18. The f00 analysis-hour keeps its enrichment
  (native-level duct, NEXRAD composite, commercial-link boost,
  ProfilesFile write) and the GridCache broadcast.
- `GridTaskEnqueuer.seed/1` is called again so grid_tasks gets the
  18 forecast-hour rows the Rust worker claims.
- Test asserts only one Oban job is enqueued and the matching 18
  grid_tasks rows land in the DB.
- `deployment-grid-rs.yaml` flips `PROP_SCORES_DIR` from
  `/data/scores_shadow` to `/data/scores` so Rust writes to the live
  score tree. No file collision — Elixir only writes the f00 files.

The hot pod memory shrink (6 Gi → 1 Gi) is deferred until one full
hourly cycle runs clean under the new split.

k8s/deployment-grid-rs.yaml

@@ -3,9 +3,11 @@ kind: Deployment
 metadata:
   name: prop-grid-rs
   namespace: prop
-  # Phase 1 (shadow) default: write to /data/scores_shadow so Elixir keeps
+  # Phase 2 cutover: Rust writes directly to /data/scores. Elixir only
-  # owning the live /data/scores tree. Flip PROP_SCORES_DIR to /data/scores
+  # runs the f00 analysis-hour step (which carries enrichment Rust
-  # at cutover after ShadowComparator shows < 0.5 mean delta for 72h.
+  # doesn't own yet — native-level duct merge, NEXRAD composite,
+  # commercial-link degradation, ProfilesFile write). f01..f18 land
+  # here from the Rust worker.
 spec:
   replicas: 1
   minReadySeconds: 5
@@ -56,7 +58,7 @@ spec:
             - name: HRRR_BASE_URL
               value: "http://skippy.w5isp.com:8080"
             - name: PROP_SCORES_DIR
-              value: "/data/scores_shadow"
+              value: "/data/scores"
             - name: RUST_LOG
               value: "info"
             - name: PROP_GRID_RS_PG_CONNS

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -45,6 +45,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   alias Microwaveprop.Propagation
   alias Microwaveprop.Propagation.BandConfig
   alias Microwaveprop.Propagation.Grid
+  alias Microwaveprop.Propagation.GridTaskEnqueuer
   alias Microwaveprop.Propagation.ProfilesFile
   alias Microwaveprop.Propagation.ScoreCache
   alias Microwaveprop.Weather
@@ -82,25 +83,17 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
     two_hours_ago = DateTime.add(DateTime.utc_now(), -2, :hour)
     run_time = two_hours_ago |> HrrrClient.nearest_hrrr_hour() |> DateTime.truncate(:second)
 
-    Logger.info("PropagationGrid: seeding chain run_time=#{run_time}, f00-f#{@max_forecast_hour} (parallel)")
+    Logger.info("PropagationGrid: seeding chain run_time=#{run_time}, f00 (+f01-f#{@max_forecast_hour} via grid_tasks)")
 
-    # Fan out all 19 forecast hours at once. Each step is independent
+    # Post-cutover: Elixir only runs the f00 analysis-hour step. f00 carries
-    # (different HRRR URL, different valid_time file output) so there's
+    # the expensive enrichment that Rust doesn't cover yet (native-level
-    # no dependency that requires the old sequential chain pattern.
+    # duct merge, NEXRAD composite, commercial-link degradation) and writes
-    # With 2 slots/pod × 3 pods = 6 concurrent workers the chain wall
+    # the ProfilesFile that /weather reads. f01..f18 go through the
-    # time drops from ~48 min to ~10 min. Cleanup (retain_window,
+    # `grid_tasks` handoff queue to the Rust `prop-grid-rs` worker, which
-    # prune) runs on PropagationPruneWorker's own 15-min cron.
+    # fetches + decodes + scores each forecast hour independently.
-    jobs =
+    Oban.insert_all([new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => 0})])
-      for fh <- 0..@max_forecast_hour do
-        new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh})
-      end
 
-    Oban.insert_all(jobs)
+    _ = GridTaskEnqueuer.seed(run_time)
-
-    # `GridTaskEnqueuer.seed/1` feeds the Rust `prop-grid-rs` worker via
-    # grid_tasks. Disabled until the Rust image is validated end-to-end
-    # on talos5 — once cutover is approved, the Elixir fan-out above
-    # shrinks to just fh=0 and this line comes back.
 
     :ok
   end

test/microwaveprop/workers/propagation_grid_worker_test.exs

@@ -45,24 +45,34 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
   end
 
   describe "perform/1 — chain seeding (empty args)" do
-    test "enqueues all 19 forecast-hour jobs at once, all pointing at the same run_time" do
+    test "enqueues only f00 in Oban; f01..f18 go to grid_tasks for Rust" do
       Oban.Testing.with_testing_mode(:manual, fn ->
+        import Ecto.Query
+
         assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}})
 
         jobs = all_enqueued(worker: PropagationGridWorker)
-        assert length(jobs) == 19
+        assert length(jobs) == 1
 
-        fhs = jobs |> Enum.map(& &1.args["forecast_hour"]) |> Enum.sort()
+        [job] = jobs
-        assert fhs == Enum.to_list(0..18)
+        assert job.args["forecast_hour"] == 0
 
-        # Every child job shares the same run_time and it's normalized
+        # run_time is normalized to top-of-hour UTC.
-        # to top-of-hour UTC.
+        {:ok, dt, _} = DateTime.from_iso8601(job.args["run_time"])
-        run_times = jobs |> Enum.map(& &1.args["run_time"]) |> Enum.uniq()
-        assert [rt] = run_times
-        {:ok, dt, _} = DateTime.from_iso8601(rt)
         assert dt.minute == 0
         assert dt.second == 0
         assert DateTime.before?(dt, DateTime.utc_now())
+
+        # The Rust worker picks up f01..f18 from the grid_tasks table.
+        task_fhs =
+          Microwaveprop.Repo.all(
+            from t in "grid_tasks",
+              where: t.run_time == ^DateTime.truncate(dt, :second),
+              select: t.forecast_hour,
+              order_by: t.forecast_hour
+          )
+
+        assert task_fhs == Enum.to_list(1..18)
       end)
     end
   end