From 65693ed415b67ba4ae2de34e8a7d56d8afd9513d Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 16:51:13 -0500 Subject: [PATCH] =?UTF-8?q?feat(prop):=20Phase=202=20cutover=20=E2=80=94?= =?UTF-8?q?=20Rust=20owns=20f01=E2=80=93f18?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 10 ++++--- .../workers/propagation_grid_worker.ex | 27 +++++++----------- .../workers/propagation_grid_worker_test.exs | 28 +++++++++++++------ 3 files changed, 35 insertions(+), 30 deletions(-) diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 7ae44e4b..9e5ff98c 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/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 - # owning the live /data/scores tree. Flip PROP_SCORES_DIR to /data/scores - # at cutover after ShadowComparator shows < 0.5 mean delta for 72h. + # Phase 2 cutover: Rust writes directly to /data/scores. Elixir only + # runs the f00 analysis-hour step (which carries enrichment Rust + # 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 diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 6e48b9f2..d26d2af3 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/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 - # (different HRRR URL, different valid_time file output) so there's - # no dependency that requires the old sequential chain pattern. - # With 2 slots/pod × 3 pods = 6 concurrent workers the chain wall - # time drops from ~48 min to ~10 min. Cleanup (retain_window, - # prune) runs on PropagationPruneWorker's own 15-min cron. - jobs = - for fh <- 0..@max_forecast_hour do - new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh}) - end + # Post-cutover: Elixir only runs the f00 analysis-hour step. f00 carries + # the expensive enrichment that Rust doesn't cover yet (native-level + # duct merge, NEXRAD composite, commercial-link degradation) and writes + # the ProfilesFile that /weather reads. f01..f18 go through the + # `grid_tasks` handoff queue to the Rust `prop-grid-rs` worker, which + # fetches + decodes + scores each forecast hour independently. + Oban.insert_all([new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => 0})]) - Oban.insert_all(jobs) - - # `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. + _ = GridTaskEnqueuer.seed(run_time) :ok end diff --git a/test/microwaveprop/workers/propagation_grid_worker_test.exs b/test/microwaveprop/workers/propagation_grid_worker_test.exs index 0a2febb4..2782a239 100644 --- a/test/microwaveprop/workers/propagation_grid_worker_test.exs +++ b/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() - assert fhs == Enum.to_list(0..18) + [job] = jobs + assert job.args["forecast_hour"] == 0 - # Every child job shares the same run_time and it's normalized - # to top-of-hour UTC. - run_times = jobs |> Enum.map(& &1.args["run_time"]) |> Enum.uniq() - assert [rt] = run_times - {:ok, dt, _} = DateTime.from_iso8601(rt) + # run_time is normalized to top-of-hour UTC. + {:ok, dt, _} = DateTime.from_iso8601(job.args["run_time"]) 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