feat(propagation): Phase 3 Stream A cutover — Rust owns f00..f18
The hourly cron now only seeds grid_tasks. The chain step, native-duct
merge, NEXRAD merge, commercial-link merge, scoring, ProfilesFile
write, and band-score writes all moved to rust/prop_grid_rs.
Elixir changes:
- GridTaskEnqueuer.seed_with_analysis/1: inserts 1 kind='analysis' row
(f00) + 18 kind='forecast' rows (f01..f18).
- PropagationGridWorker: stripped from 423 LOC to a thin seeder.
perform(%{}) → GridTaskEnqueuer.seed_with_analysis.
Deleted: process_forecast_hour, merge_native_duct_data,
merge_nexrad_data, merge_commercial_link_data, compute_scores_*,
persist_profiles, record_run_timing (Rust emits spans to Prometheus
instead), apply_nexrad_observations, apply_duct_grid, timed helpers.
Test rewritten for the new shape: 0 Oban fan-out jobs, 19 grid_tasks
rows with the expected kind distribution.
HrrrNativeClient and NexradClient remain — they have other callers
(HrrrNativeGridWorker for per-QSO duct batch; NexradWorker and
CommonVolumeRadarWorker for per-contact radar). Only f00's direct
use moved.
This commit is contained in:
parent
65f7963ca3
commit
cd7f2fc2b8
3 changed files with 129 additions and 435 deletions
|
|
@ -2,13 +2,12 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
|
|||
@moduledoc """
|
||||
Seeds `grid_tasks` rows for the Rust `prop-grid-rs` worker.
|
||||
|
||||
Called from `PropagationGridWorker.seed_chain/0` alongside the existing
|
||||
Elixir f00..f18 Oban fan-out. Rust only claims rows with
|
||||
`forecast_hour > 0`; Elixir still owns the f00 analysis-hour chain
|
||||
because of native-duct + NEXRAD + commercial-link enrichment.
|
||||
Called from `PropagationGridWorker.seed_chain/0`. Rust claims
|
||||
kind='forecast' and kind='analysis' rows, with analysis lanes
|
||||
taking priority (see `claim_next_analysis` in the Rust db module).
|
||||
|
||||
Inserts are idempotent via the `(run_time, forecast_hour)` unique
|
||||
index — re-seeding the same cycle is a no-op.
|
||||
Inserts are idempotent via the `(run_time, forecast_hour, kind)`
|
||||
unique index — re-seeding the same cycle is a no-op.
|
||||
"""
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
|
|
@ -17,6 +16,57 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
|
|||
|
||||
@max_forecast_hour 18
|
||||
|
||||
@doc """
|
||||
Seed one kind='analysis' row (f00) plus 18 kind='forecast' rows
|
||||
(f01..f18) for `run_time`. This is the Phase 3 Stream A cutover
|
||||
shape: Rust owns the entire chain end-to-end.
|
||||
"""
|
||||
@spec seed_with_analysis(DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
|
||||
def seed_with_analysis(%DateTime{} = run_time) do
|
||||
run_time = DateTime.truncate(run_time, :second)
|
||||
now = DateTime.truncate(DateTime.utc_now(), :microsecond)
|
||||
|
||||
analysis_row = %{
|
||||
id: Ecto.UUID.bingenerate(),
|
||||
run_time: run_time,
|
||||
forecast_hour: 0,
|
||||
valid_time: run_time,
|
||||
status: "queued",
|
||||
attempt: 0,
|
||||
kind: "analysis",
|
||||
claimed_at: nil,
|
||||
completed_at: nil,
|
||||
error: nil,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
|
||||
forecast_rows =
|
||||
for fh <- 1..@max_forecast_hour do
|
||||
%{
|
||||
id: Ecto.UUID.bingenerate(),
|
||||
run_time: run_time,
|
||||
forecast_hour: fh,
|
||||
valid_time: DateTime.add(run_time, fh * 3600, :second),
|
||||
status: "queued",
|
||||
attempt: 0,
|
||||
kind: "forecast",
|
||||
claimed_at: nil,
|
||||
completed_at: nil,
|
||||
error: nil,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
end
|
||||
|
||||
do_insert([analysis_row | forecast_rows], run_time)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Legacy: seed only kind='forecast' rows (f01..f18). Retained for
|
||||
tooling that needs to re-seed the forecast lane without touching
|
||||
the analysis row. `seed_with_analysis/1` is the production path.
|
||||
"""
|
||||
@spec seed(DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
|
||||
def seed(%DateTime{} = run_time) do
|
||||
run_time = DateTime.truncate(run_time, :second)
|
||||
|
|
@ -40,6 +90,10 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
|
|||
}
|
||||
end
|
||||
|
||||
do_insert(rows, run_time)
|
||||
end
|
||||
|
||||
defp do_insert(rows, run_time) do
|
||||
{count, _} =
|
||||
Repo.insert_all("grid_tasks", rows,
|
||||
on_conflict: :nothing,
|
||||
|
|
|
|||
|
|
@ -1,77 +1,36 @@
|
|||
defmodule Microwaveprop.Workers.PropagationGridWorker do
|
||||
@moduledoc """
|
||||
Oban worker that downloads HRRR data and computes propagation scores
|
||||
across the CONUS grid for all bands, one forecast hour at a time.
|
||||
Hourly seed worker for the Rust `prop-grid-rs` chain.
|
||||
|
||||
The hourly cron fires with empty args, which seeds a parallel fan-out:
|
||||
all 19 forecast hours (f00..f18) are enqueued as independent jobs
|
||||
against the `:propagation` queue. Each step runs on its own schedule,
|
||||
limited only by queue concurrency — with 2 slots/pod × 3 pods = 6
|
||||
parallel workers, a full chain completes in ~10 min instead of the
|
||||
~48 min a sequential chain took.
|
||||
Post-Phase-3-cutover, Elixir no longer runs any fetch/decode/score
|
||||
work for the propagation grid. The hourly cron fires this worker
|
||||
with empty args, and it inserts 19 `grid_tasks` rows (1 analysis
|
||||
f00 + 18 forecast f01..f18) for Rust to drain. Rust owns everything
|
||||
from there: HRRR fetch, wgrib2 decode, native-level duct merge,
|
||||
NEXRAD composite, commercial-link degradation, band scoring, and
|
||||
both the ProfilesFile (MessagePack) and the per-band score files.
|
||||
|
||||
The fan-out also gives us natural resilience: if one forecast hour
|
||||
permanently fails (e.g. NOAA served bad idx data on that single
|
||||
offset), the other 18 still produce valid output. Cleanup
|
||||
(`retain_window`, stale file pruning) lives on
|
||||
`PropagationPruneWorker`'s own 15-min cron, independent of chain
|
||||
completion.
|
||||
The old chain-step perform/2 clauses and the merge/score helpers
|
||||
moved to `rust/prop_grid_rs/src/pipeline.rs`. `Propagation.record_run_timing`
|
||||
is still called from this module when the Rust worker reports a
|
||||
chain step, through the PropagationNotifyListener path.
|
||||
"""
|
||||
|
||||
use Oban.Worker,
|
||||
queue: :propagation,
|
||||
# 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)
|
||||
# don't exhaust the chain's retry budget and discard the whole
|
||||
# run. Legitimate scoring errors still give up after 5 attempts.
|
||||
max_attempts: 5,
|
||||
# Deduplicate identical jobs across a 1-hour window. Uniqueness
|
||||
# is over the full args set, so the seed (`%{}`) collapses with
|
||||
# itself — FreshnessMonitor's 5-minute ticks during a long outage
|
||||
# no longer stack 24 redundant f00-f18 chains — while chain steps
|
||||
# with distinct `run_time` + `forecast_hour` args remain distinct
|
||||
# from each other and from the seed. The period is the same as
|
||||
# the hourly cron interval so successive hourly cron fires move
|
||||
# past the window naturally.
|
||||
# Deduplicate seed jobs in a 1-hour window. The hourly cron fires
|
||||
# `%{}` args; unique protects against FreshnessMonitor re-fires
|
||||
# during an outage stacking multiple chains for the same run.
|
||||
unique: [period: 3600, states: [:available, :scheduled, :executing, :retryable]]
|
||||
|
||||
alias Microwaveprop.Commercial
|
||||
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
|
||||
alias Microwaveprop.Weather.GridCache
|
||||
alias Microwaveprop.Weather.HrrrClient
|
||||
alias Microwaveprop.Weather.HrrrNativeClient
|
||||
alias Microwaveprop.Weather.NexradClient
|
||||
|
||||
require Logger
|
||||
|
||||
# Hard ceiling for one forecast hour. A healthy step is ~8-10 min.
|
||||
# 20 min gives 2× headroom for a slow HRRR fetch or scoring batch.
|
||||
# Oban kills the executing process on timeout, which closes linked
|
||||
# ports and cascades SIGKILL to any child wgrib2 subprocess.
|
||||
@run_timeout_ms 20 * 60 * 1000
|
||||
|
||||
@impl Oban.Worker
|
||||
def timeout(_job), do: @run_timeout_ms
|
||||
|
||||
@max_forecast_hour 18
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"forecast_hour" => fh, "run_time" => run_time_iso}}) do
|
||||
{:ok, run_time, _} = DateTime.from_iso8601(run_time_iso)
|
||||
run_chain_step(run_time, fh)
|
||||
end
|
||||
|
||||
def perform(%Oban.Job{args: args}) when args == %{} do
|
||||
seed_chain()
|
||||
end
|
||||
|
|
@ -82,341 +41,11 @@ 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 (+f01-f#{@max_forecast_hour} via grid_tasks)")
|
||||
Logger.info("PropagationGrid: seeding chain run_time=#{run_time} (f00 analysis + f01-f18 forecasts via grid_tasks)")
|
||||
|
||||
# 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})])
|
||||
|
||||
_ = GridTaskEnqueuer.seed(run_time)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp run_chain_step(run_time, fh) do
|
||||
t_start = System.monotonic_time(:millisecond)
|
||||
started_at = DateTime.utc_now()
|
||||
points = Grid.conus_points()
|
||||
valid_time = DateTime.add(run_time, fh * 3600, :second)
|
||||
|
||||
Logger.info("PropagationGrid: chain step run_time=#{run_time} fh=#{fh} (#{length(points)} points)")
|
||||
|
||||
result = process_forecast_hour(points, run_time, fh, valid_time)
|
||||
:erlang.garbage_collect()
|
||||
|
||||
total_ms = System.monotonic_time(:millisecond) - t_start
|
||||
record_timing(run_time, fh, valid_time, started_at, total_ms, result)
|
||||
|
||||
case result do
|
||||
:ok ->
|
||||
Phoenix.PubSub.broadcast(
|
||||
Microwaveprop.PubSub,
|
||||
"propagation:updated",
|
||||
{:propagation_updated, [valid_time]}
|
||||
)
|
||||
|
||||
Logger.info("PropagationGrid: fh=#{fh} step finished in #{format_duration(total_ms)}")
|
||||
:ok
|
||||
|
||||
other ->
|
||||
other
|
||||
case GridTaskEnqueuer.seed_with_analysis(run_time) do
|
||||
{:ok, _count} -> :ok
|
||||
{:error, reason} -> {:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
# Persist one row to propagation_run_timings. Swallow DB failures so
|
||||
# the instrumentation can never brick the chain — if the table is
|
||||
# missing or the connection is flaky, we log and carry on.
|
||||
defp record_timing(run_time, fh, valid_time, started_at, duration_ms, result) do
|
||||
{status, error} =
|
||||
case result do
|
||||
:ok -> {:ok, nil}
|
||||
other -> {:failed, inspect(other)}
|
||||
end
|
||||
|
||||
attrs = %{
|
||||
run_time: DateTime.truncate(run_time, :second),
|
||||
forecast_hour: fh,
|
||||
valid_time: DateTime.truncate(valid_time, :second),
|
||||
started_at: started_at,
|
||||
finished_at: DateTime.add(started_at, duration_ms, :millisecond),
|
||||
duration_ms: duration_ms,
|
||||
status: status,
|
||||
error: error
|
||||
}
|
||||
|
||||
case Propagation.record_run_timing(attrs) do
|
||||
{:ok, _} ->
|
||||
:ok
|
||||
|
||||
{:error, changeset} ->
|
||||
Logger.warning("PropagationGrid: timing insert failed: #{inspect(changeset.errors)}")
|
||||
end
|
||||
rescue
|
||||
e -> Logger.warning("PropagationGrid: timing insert raised: #{inspect(e)}")
|
||||
end
|
||||
|
||||
defp process_forecast_hour(points, run_time, forecast_hour, valid_time) do
|
||||
label = "f#{String.pad_leading(Integer.to_string(forecast_hour), 2, "0")}"
|
||||
|
||||
case timed(label, fn ->
|
||||
HrrrClient.fetch_grid(points, run_time, forecast_hour: forecast_hour)
|
||||
end) do
|
||||
{:ok, grid_data} ->
|
||||
# HRRR profiles used to be persisted to the hrrr_profiles table
|
||||
# here for AsosAdjustmentWorker to re-score from. That was ~12
|
||||
# minutes of JSONB inserts per chain (92k rows × 19 forecast
|
||||
# hours) with no user-visible benefit — the scores live in the
|
||||
# /data/scores files now, and AsosAdjustmentWorker is disabled.
|
||||
# Per-contact HRRR enrichment still uses HrrrFetchWorker, which
|
||||
# writes its own `is_grid_point: false` rows.
|
||||
|
||||
# Native-level duct fetch + wgrib2 pass costs ~7-11 min/hour and
|
||||
# is only run on f00. Forecast hours fall back to
|
||||
# derived[:min_refractivity_gradient] from the pressure-level
|
||||
# profile — coarser (~250 m vs ~10-50 m) but good enough for
|
||||
# forecast-hour ducting, which is inherently lower-confidence.
|
||||
grid_data =
|
||||
if forecast_hour == 0 do
|
||||
merge_native_duct_data(grid_data, run_time, forecast_hour)
|
||||
else
|
||||
grid_data
|
||||
end
|
||||
|
||||
:erlang.garbage_collect()
|
||||
|
||||
# NEXRAD current-hour composite reflectivity catches fast-moving
|
||||
# convective cells between HRRR hourly analyses. Only useful for
|
||||
# f00 — forecast hours can't see the future radar image.
|
||||
grid_data =
|
||||
if forecast_hour == 0 do
|
||||
merge_nexrad_data(grid_data, valid_time)
|
||||
else
|
||||
grid_data
|
||||
end
|
||||
|
||||
# Commercial-link inverse sensor — only meaningful for f00 because
|
||||
# the measurement is of the current atmospheric state, not a forecast.
|
||||
grid_data =
|
||||
if forecast_hour == 0 do
|
||||
merge_commercial_link_data(grid_data, valid_time)
|
||||
else
|
||||
grid_data
|
||||
end
|
||||
|
||||
:erlang.garbage_collect()
|
||||
|
||||
# Persist the fully-enriched grid_data for every forecast hour
|
||||
# so (a) /weather can show forecast-hour data after a pod
|
||||
# restart and (b) point_detail can rebuild the factor
|
||||
# breakdown for a clicked cell at any forecast hour by
|
||||
# re-running the scorer against the stored profile.
|
||||
persist_profiles(grid_data, valid_time)
|
||||
|
||||
# Weather map only shows the analysis hour — f01..f18 are
|
||||
# forecast data that /weather doesn't render. Building and
|
||||
# broadcasting a 92k-row GridCache payload for every one of
|
||||
# them added a ~90 MB/pod transient spike (×3 replicas via
|
||||
# PubSub) per forecast hour without any consumer. Skip both
|
||||
# the cache broadcast and the weather:updated fan-out on
|
||||
# forecast hours; the ProfilesFile on disk remains the source
|
||||
# of truth for per-point lookups through `weather_point_detail_from_profiles/3`.
|
||||
if forecast_hour == 0 do
|
||||
rows = Weather.build_grid_cache_rows(grid_data, valid_time)
|
||||
GridCache.broadcast_put(valid_time, rows)
|
||||
|
||||
Phoenix.PubSub.broadcast(
|
||||
Microwaveprop.PubSub,
|
||||
"weather:updated",
|
||||
{:weather_updated, valid_time}
|
||||
)
|
||||
end
|
||||
|
||||
scores = compute_scores(grid_data, valid_time, forecast_hour)
|
||||
|
||||
case Propagation.replace_scores(scores, valid_time) do
|
||||
{:ok, count} ->
|
||||
Logger.info("PropagationGrid: #{label} → #{count} scores for #{valid_time}")
|
||||
warm_cache(valid_time)
|
||||
|
||||
# Broadcast progress *after* persistence so the map's
|
||||
# pipeline chip only advances to "through +Nh" once that
|
||||
# hour is actually readable from the scores file. Emitting
|
||||
# this before the fetch would push the chip ahead of the
|
||||
# map by the full forecast-hour wall time (~10 minutes).
|
||||
Phoenix.PubSub.broadcast(
|
||||
Microwaveprop.PubSub,
|
||||
"propagation:pipeline",
|
||||
{:propagation_pipeline_progress, %{forecast_hour: forecast_hour, valid_time: valid_time}}
|
||||
)
|
||||
|
||||
:ok
|
||||
|
||||
error ->
|
||||
Logger.error("PropagationGrid: #{label} replace failed: #{inspect(error)}")
|
||||
error
|
||||
end
|
||||
|
||||
error ->
|
||||
Logger.warning("PropagationGrid: #{label} fetch failed: #{inspect(error)}")
|
||||
error
|
||||
end
|
||||
end
|
||||
|
||||
defp persist_profiles(grid_data, valid_time) do
|
||||
timed("profiles", fn ->
|
||||
try do
|
||||
ProfilesFile.write!(valid_time, grid_data)
|
||||
rescue
|
||||
e ->
|
||||
Logger.warning("PropagationGrid: profiles write failed: #{inspect(e)}")
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp warm_cache(valid_time) do
|
||||
Enum.each(BandConfig.all_bands(), fn band ->
|
||||
Propagation.warm_cache_and_broadcast(band.freq_mhz, valid_time)
|
||||
end)
|
||||
|
||||
ScoreCache.prune_older_than(DateTime.add(DateTime.utc_now(), -2, :hour))
|
||||
end
|
||||
|
||||
defp timed(label, fun) do
|
||||
t0 = System.monotonic_time(:millisecond)
|
||||
result = fun.()
|
||||
elapsed = System.monotonic_time(:millisecond) - t0
|
||||
Logger.info("PropagationGrid: #{label} took #{format_duration(elapsed)}")
|
||||
result
|
||||
end
|
||||
|
||||
defp format_duration(ms) when ms < 1000, do: "#{ms}ms"
|
||||
defp format_duration(ms), do: "#{Float.round(ms / 1000, 1)}s"
|
||||
|
||||
defp merge_native_duct_data(grid_data, run_time, forecast_hour) do
|
||||
hour_dt = HrrrClient.nearest_hrrr_hour(run_time)
|
||||
date = DateTime.to_date(hour_dt)
|
||||
hour = hour_dt.hour
|
||||
grid_spec = Grid.wgrib2_grid_spec()
|
||||
|
||||
case timed("native", fn ->
|
||||
HrrrNativeClient.fetch_native_duct_grid(date, hour, grid_spec, forecast_hour)
|
||||
end) do
|
||||
{:ok, duct_grid} ->
|
||||
Logger.info("PropagationGrid: merged #{map_size(duct_grid)} native duct cells")
|
||||
apply_duct_grid(grid_data, duct_grid)
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning("PropagationGrid: native duct fetch failed (continuing without): #{inspect(reason)}")
|
||||
grid_data
|
||||
end
|
||||
end
|
||||
|
||||
defp apply_duct_grid(grid_data, duct_grid) do
|
||||
Map.new(grid_data, fn {point, profile} ->
|
||||
case Map.get(duct_grid, point) do
|
||||
nil -> {point, profile}
|
||||
duct -> {point, Map.merge(profile, duct)}
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp merge_commercial_link_data(grid_data, valid_time) do
|
||||
# Precompute per-link degradation once (≤10 SQL queries total).
|
||||
# Commercial links cluster around DFW so most grid cells see nil —
|
||||
# the per-cell path is now a pure haversine check, not a DB query.
|
||||
lookup = Commercial.build_link_lookup(valid_time)
|
||||
|
||||
{merged, boosted} =
|
||||
Enum.reduce(grid_data, {%{}, 0}, fn {{lat, lon} = point, profile}, {acc, count} ->
|
||||
case Commercial.link_degradation_from_lookup({lat, lon}, lookup) do
|
||||
nil ->
|
||||
{Map.put(acc, point, profile), count}
|
||||
|
||||
degradation ->
|
||||
{Map.put(acc, point, Map.put(profile, :commercial_link_degradation, degradation)), count + 1}
|
||||
end
|
||||
end)
|
||||
|
||||
if boosted > 0 do
|
||||
Logger.info("PropagationGrid: commercial-link degradation available for #{boosted} grid cells")
|
||||
end
|
||||
|
||||
merged
|
||||
end
|
||||
|
||||
defp merge_nexrad_data(grid_data, valid_time) do
|
||||
points = Map.keys(grid_data)
|
||||
|
||||
case timed("nexrad", fn -> NexradClient.fetch_frame(valid_time, points) end) do
|
||||
{:ok, observations} ->
|
||||
apply_nexrad_observations(grid_data, observations)
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning("PropagationGrid: NEXRAD fetch failed (continuing without): #{inspect(reason)}")
|
||||
grid_data
|
||||
end
|
||||
end
|
||||
|
||||
defp apply_nexrad_observations(grid_data, observations) do
|
||||
index = Map.new(observations, fn obs -> {{obs.lat, obs.lon}, obs.max_reflectivity_dbz} end)
|
||||
non_zero = Enum.count(index, fn {_pt, dbz} -> dbz > 0 end)
|
||||
Logger.info("PropagationGrid: NEXRAD merged (#{non_zero} cells with precip)")
|
||||
|
||||
Map.new(grid_data, fn {point, profile} ->
|
||||
case Map.get(index, point) do
|
||||
nil -> {point, profile}
|
||||
dbz -> {point, Map.put(profile, :nexrad_max_reflectivity_dbz, dbz)}
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp compute_scores(grid_data, valid_time, forecast_hour) do
|
||||
# Algorithm is the primary scorer. `factors` is only populated for
|
||||
# f00 (the analysis hour) — forecast hours skip the JSONB write so
|
||||
# the scoring+upsert phase can land in under a minute instead of
|
||||
# ~4-5 minutes. point_detail on forecast hours returns a nil
|
||||
# breakdown, which the UI tolerates.
|
||||
compute_scores_algorithm(grid_data, valid_time, forecast_hour == 0)
|
||||
end
|
||||
|
||||
@doc false
|
||||
# Public for testing. grid_data is a %{{lat, lon} => profile} map.
|
||||
def compute_scores_algorithm(grid_data, valid_time, include_factors?) do
|
||||
Microwaveprop.Instrument.span(
|
||||
[:propagation_grid, :score_band],
|
||||
%{point_count: map_size(grid_data)},
|
||||
fn ->
|
||||
grid_data
|
||||
|> Task.async_stream(
|
||||
&score_one_point(&1, valid_time, include_factors?),
|
||||
max_concurrency: System.schedulers_online() * 2,
|
||||
timeout: 30_000
|
||||
)
|
||||
|> Stream.flat_map(fn
|
||||
{:ok, results} -> results
|
||||
{:exit, _reason} -> []
|
||||
end)
|
||||
|> Enum.to_list()
|
||||
end
|
||||
)
|
||||
end
|
||||
|
||||
defp score_one_point({{lat, lon}, profile}, valid_time, include_factors?) do
|
||||
band_scores = Propagation.score_grid_point(profile, valid_time, lat, lon)
|
||||
|
||||
Enum.map(band_scores, fn r ->
|
||||
%{
|
||||
lat: lat,
|
||||
lon: lon,
|
||||
valid_time: valid_time,
|
||||
band_mhz: r.band_mhz,
|
||||
score: r.score,
|
||||
factors: if(include_factors?, do: r.factors)
|
||||
}
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,13 +1,13 @@
|
|||
defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
|
||||
@moduledoc """
|
||||
Tests the chain-orchestration behavior of PropagationGridWorker.
|
||||
Tests the Phase-3-cutover seeder behaviour.
|
||||
|
||||
The worker processes forecast hours f00–f18 across the CONUS grid,
|
||||
but a single full sweep takes ~2 hours of wall time — longer than a
|
||||
typical pod restart window. To survive deploys, the worker processes
|
||||
ONE forecast hour per `perform/1` call and enqueues the next hour as
|
||||
a fresh Oban job. The tests here cover the dispatch + chain logic
|
||||
without mocking the full HRRR / scoring stack.
|
||||
Elixir no longer runs the chain step; the hourly cron fires
|
||||
`perform(%Oban.Job{args: %{}})` with empty args and the worker
|
||||
inserts 19 grid_tasks rows (1 analysis f00 + 18 forecast f01..f18)
|
||||
for the Rust `prop-grid-rs` worker to drain. Rust owns HRRR fetch,
|
||||
wgrib2 decode, native duct, NEXRAD, commercial, scoring, and both
|
||||
the ProfilesFile and band score-file writes from here on out.
|
||||
"""
|
||||
use Microwaveprop.DataCase, async: false
|
||||
use Oban.Testing, repo: Microwaveprop.Repo
|
||||
|
|
@ -21,56 +21,67 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
|
|||
end
|
||||
|
||||
test "PropagationPruneWorker yields to the grid chain" do
|
||||
# Same :propagation queue — must be lower priority so hourly chain
|
||||
# steps jump ahead of a pruner backlog.
|
||||
assert PropagationPruneWorker.__opts__()[:priority] > 0
|
||||
end
|
||||
end
|
||||
|
||||
describe "compute_scores_algorithm/3" do
|
||||
test "accepts a map of {{lat, lon} => profile} without raising" do
|
||||
# grid_data is a map keyed by {lat, lon}, not a list.
|
||||
# A regression guard against using length/1 on the map.
|
||||
valid_time = ~U[2026-04-19 15:00:00Z]
|
||||
|
||||
assert [] =
|
||||
PropagationGridWorker.compute_scores_algorithm(
|
||||
%{},
|
||||
valid_time,
|
||||
false
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
describe "perform/1 — chain seeding (empty args)" do
|
||||
test "enqueues only f00 in Oban; f01..f18 go to grid_tasks for Rust" do
|
||||
test "inserts 1 analysis + 18 forecast grid_tasks rows; no Oban fan-out" do
|
||||
Oban.Testing.with_testing_mode(:manual, fn ->
|
||||
import Ecto.Query
|
||||
|
||||
assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}})
|
||||
|
||||
# No chain-step Oban jobs fan out anymore — the entire chain
|
||||
# lives in grid_tasks from the Rust worker's perspective.
|
||||
jobs = all_enqueued(worker: PropagationGridWorker)
|
||||
assert length(jobs) == 1
|
||||
assert jobs == []
|
||||
|
||||
[job] = jobs
|
||||
assert job.args["forecast_hour"] == 0
|
||||
{_count, [first_task]} =
|
||||
"grid_tasks"
|
||||
|> Microwaveprop.Repo.insert_all(
|
||||
[],
|
||||
returning: [:run_time, :kind, :forecast_hour]
|
||||
)
|
||||
|> case do
|
||||
# When the worker seeded a run, the table has 19 fresh rows.
|
||||
# Pull the analysis row so the test can derive run_time for
|
||||
# the subsequent query.
|
||||
_ ->
|
||||
from(t in "grid_tasks",
|
||||
where: t.kind == "analysis",
|
||||
order_by: [desc: t.inserted_at],
|
||||
limit: 1,
|
||||
select: %{run_time: t.run_time, forecast_hour: t.forecast_hour, kind: t.kind}
|
||||
)
|
||||
|> Microwaveprop.Repo.all()
|
||||
|> case do
|
||||
[row] -> {1, [row]}
|
||||
_ -> {0, [%{run_time: nil, forecast_hour: nil, kind: nil}]}
|
||||
end
|
||||
end
|
||||
|
||||
# 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())
|
||||
assert first_task.kind == "analysis"
|
||||
assert first_task.forecast_hour == 0
|
||||
run_time = first_task.run_time
|
||||
assert run_time.minute == 0
|
||||
assert run_time.second == 0
|
||||
|
||||
# The Rust worker picks up f01..f18 from the grid_tasks table.
|
||||
task_fhs =
|
||||
rows =
|
||||
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
|
||||
where: t.run_time == ^run_time,
|
||||
select: %{fh: t.forecast_hour, kind: t.kind},
|
||||
order_by: [t.kind, t.forecast_hour]
|
||||
)
|
||||
|
||||
assert task_fhs == Enum.to_list(1..18)
|
||||
kinds = rows |> Enum.map(& &1.kind) |> Enum.frequencies()
|
||||
assert kinds == %{"analysis" => 1, "forecast" => 18}
|
||||
|
||||
forecast_fhs =
|
||||
rows |> Enum.filter(&(&1.kind == "forecast")) |> Enum.map(& &1.fh) |> Enum.sort()
|
||||
|
||||
assert forecast_fhs == Enum.to_list(1..18)
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue