From fe2ad2df15874067327e8c7da9bf19056c9c6adf Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 5 May 2026 18:00:10 -0500 Subject: [PATCH] fix(pskr): self-healing follow-up sampler runs for hours with missing HRRR The producer-only path left older hours stuck at 0% HRRR coverage forever once their fetch tasks drained: the rolling-window cron only re-passes the prior hour, so manual lookback runs (or any out-of-window hour) wrote NULL samples + enqueued tasks, then never re-ran to UPSERT the now-landed HRRR data. Fix: split the responsibility cleanly. * CalibrationSampler.build_for_hour/1 now returns %{upserted: int, missing_hrrr_cells: int} so callers know whether the producer enqueued anything (i.e. whether a follow-up makes sense). Spec-tightened with a new @type result. * PskrCalibrationWorker, after each hour's sampler pass, schedules a follow-up of itself for that exact hour 10 min later when missing > 0. The follow-up carries args %{"hour_utc" => ..., "_follow_up" => true}; the sentinel prevents follow-ups from chaining further follow-ups (rolling-window cron is the safety net). Tests: * Updated 4 existing sampler tests to the new map return shape. * Moved the follow-up assertions from the sampler suite into the worker suite where they belong (sampler is now scheduling-free). * Added 3 worker tests: schedules-when-missing, no-schedule-when- covered, follow-up-doesn't-chain. 3313 tests + 228 properties, 0 failures. --- lib/microwaveprop/pskr/calibration_sampler.ex | 55 ++++++++----- .../workers/pskr_calibration_worker.ex | 64 +++++++++++++-- .../pskr/calibration_sampler_test.exs | 12 +-- .../workers/pskr_calibration_worker_test.exs | 78 +++++++++++++++++++ 4 files changed, 179 insertions(+), 30 deletions(-) diff --git a/lib/microwaveprop/pskr/calibration_sampler.ex b/lib/microwaveprop/pskr/calibration_sampler.ex index 37c8d4f5..331d4744 100644 --- a/lib/microwaveprop/pskr/calibration_sampler.ex +++ b/lib/microwaveprop/pskr/calibration_sampler.ex @@ -46,19 +46,41 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do @typep cell_key :: {band :: String.t(), lat :: float(), lon :: float()} - @doc """ - Build calibration samples for one hour. Returns the upsert count. - Returns `0` when no PSKR spots were captured for that hour — for - example, during PSKR client downtime. + @typedoc """ + Result of a single `build_for_hour/1` pass: + + * `upserted` — number of `pskr_calibration_samples` rows + inserted-or-updated. + * `missing_hrrr_cells` — number of distinct cell midpoints whose + HRRR profile was unavailable. The caller (typically + `Microwaveprop.Workers.PskrCalibrationWorker`) uses this to + decide whether to schedule a follow-up sampler pass after the + Rust hrrr-point-worker has had time to drain the just-enqueued + fetch tasks. """ - @spec build_for_hour(DateTime.t()) :: non_neg_integer() + @type result :: %{upserted: non_neg_integer(), missing_hrrr_cells: non_neg_integer()} + + @doc """ + Build calibration samples for one hour. Returns + `%{upserted: count, missing_hrrr_cells: count}`. Both counts are 0 + when no PSKR spots were captured for that hour — for example, + during PSKR client downtime. + + The `missing_hrrr_cells` count tells the caller whether the Rust + hrrr-point-worker has work to drain after this fire. The + `PskrCalibrationWorker` uses it to decide whether to schedule a + follow-up sampler pass for the same hour ~10 min later, so the + newly-landed HRRR data gets UPSERTed onto the samples we just + wrote with NULL fields. + """ + @spec build_for_hour(DateTime.t()) :: result() def build_for_hour(%DateTime{} = hour_utc) do hour_utc = align_to_hour(hour_utc) spots = fetch_spots(hour_utc) if spots == [] do Logger.info("Pskr.CalibrationSampler: no spots for #{hour_utc} — skipping") - 0 + %{upserted: 0, missing_hrrr_cells: 0} else hrrr_index = build_hrrr_index(hour_utc) kp = fetch_kp(hour_utc) @@ -74,17 +96,10 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do {key, nearest_hrrr(lat, lon, hrrr_index)} end) - # Producer side of the two-pass HRRR join: any cell whose - # midpoint has no profile in the lookahead window gets a - # `hrrr_fetch_tasks` row enqueued for the Rust hrrr-point-worker - # to drain. The next rolling-window cron pass (every 5 min) will - # find the landed profile and UPSERT the sample with non-NULL - # weather fields. Existing cells with HRRR data already in - # memory are skipped — no churn against the queue table. - enqueue_missing_hrrr(cell_hrrr, hour_utc) + missing = enqueue_missing_hrrr(cell_hrrr, hour_utc) samples = Enum.map(cells, &build_sample(&1, hour_utc, cell_hrrr, kp)) - upsert(samples) + %{upserted: upsert(samples), missing_hrrr_cells: missing} end end @@ -162,7 +177,11 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do # a midpoint enqueue once, and let `HrrrPointEnqueuer.enqueue/1` # union the points into the existing `hrrr_fetch_tasks` row for # this hour rather than racing to insert duplicates. - @spec enqueue_missing_hrrr(%{cell_key() => map() | nil}, DateTime.t()) :: :ok + # + # Returns the number of unique missing-HRRR cells. The caller uses + # that to decide whether a follow-up sampler pass is needed (see + # `Microwaveprop.Workers.PskrCalibrationWorker`). + @spec enqueue_missing_hrrr(%{cell_key() => map() | nil}, DateTime.t()) :: non_neg_integer() defp enqueue_missing_hrrr(cell_hrrr, hour_utc) do points = cell_hrrr @@ -174,12 +193,12 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do case points do [] -> - :ok + 0 pts -> {:ok, _} = HrrrPointEnqueuer.enqueue(%{hour_utc => pts}) - Logger.info("Pskr.CalibrationSampler: enqueued #{length(pts)} HRRR points for #{hour_utc}") + length(pts) end end diff --git a/lib/microwaveprop/workers/pskr_calibration_worker.ex b/lib/microwaveprop/workers/pskr_calibration_worker.ex index 40b5469d..21a7b9ff 100644 --- a/lib/microwaveprop/workers/pskr_calibration_worker.ex +++ b/lib/microwaveprop/workers/pskr_calibration_worker.ex @@ -29,6 +29,22 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorker do - `lookback_hours` (integer) — override the default window size. Useful after longer outages: `%{"lookback_hours" => 24}` will sweep the last day. + + ## Self-healing follow-ups + + When the sampler enqueues `hrrr_fetch_tasks` (cells without HRRR + data), this worker schedules a follow-up of itself for that exact + hour ~10 min in the future via the `hour_utc` args path. The + follow-up's job is to UPSERT the now-landed HRRR data onto the + same calibration_samples rows. + + The chain terminates naturally: once HRRR data is fully landed for + an hour, no fetch tasks are enqueued and no follow-up is scheduled. + Args carry a `_follow_up: true` sentinel on scheduled re-runs so a + follow-up that *itself* still finds missing data won't infinitely + schedule another (the Rust worker sometimes fails on HRRR publish- + lag for the current hour); the rolling-window cron remains the + safety net. """ use Oban.Worker, queue: :backfill_enqueue, @@ -52,10 +68,15 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorker do # complete. Use `lookback_hours` to widen for catch-up. @default_lookback_hours 1 + # Delay before the self-scheduled follow-up sampler pass fires. 10 + # min covers the typical Rust hrrr-point-worker drain latency for + # ~1k point batches. + @follow_up_delay_seconds 600 + @impl Oban.Worker - def perform(%Oban.Job{args: %{"hour_utc" => iso}}) when is_binary(iso) do + def perform(%Oban.Job{args: %{"hour_utc" => iso} = args}) when is_binary(iso) do {:ok, hour, _} = DateTime.from_iso8601(iso) - process_hour(hour) + process_hour(hour, args) :ok end @@ -64,7 +85,8 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorker do now = truncate_to_hour(DateTime.utc_now()) Enum.each(0..lookback, fn offset -> - now |> DateTime.add(-offset * 3600, :second) |> process_hour() + hour = DateTime.add(now, -offset * 3600, :second) + process_hour(hour, args) end) :ok @@ -73,9 +95,39 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorker do defp lookback_hours(%{"lookback_hours" => n}) when is_integer(n) and n >= 0, do: n defp lookback_hours(_), do: @default_lookback_hours - defp process_hour(hour) do - count = CalibrationSampler.build_for_hour(hour) - if count > 0, do: Logger.info("PskrCalibrationWorker: upserted #{count} samples for #{hour}") + defp process_hour(hour, args) do + %{upserted: upserted, missing_hrrr_cells: missing} = CalibrationSampler.build_for_hour(hour) + + if upserted > 0 do + Logger.info("PskrCalibrationWorker: upserted #{upserted} samples for #{hour}") + end + + if missing > 0, do: maybe_schedule_follow_up(hour, args) + end + + # Schedule one follow-up for this hour ~10 min out unless we're + # already on a follow-up. The `_follow_up: true` sentinel + the + # `hour_utc`-keyed args together make a unique fingerprint for the + # Oban unique constraint, so duplicate follow-ups for the same + # hour within 240 s are deduped automatically. + defp maybe_schedule_follow_up(_hour, %{"_follow_up" => true}), do: :ok + + defp maybe_schedule_follow_up(hour, _args) do + %{"hour_utc" => DateTime.to_iso8601(hour), "_follow_up" => true} + |> __MODULE__.new(schedule_in: @follow_up_delay_seconds) + |> Oban.insert() + |> case do + {:ok, _job} -> + :ok + + {:error, reason} -> + # Unique-conflict (same hour follow-up already pending) or + # transient DB error — either way the rolling-window cron is + # the safety net, so log and move on. + Logger.warning("PskrCalibrationWorker: failed to schedule follow-up for #{hour}: #{inspect(reason)}") + + :ok + end end defp truncate_to_hour(dt), do: %{dt | minute: 0, second: 0, microsecond: {0, 0}} diff --git a/test/microwaveprop/pskr/calibration_sampler_test.exs b/test/microwaveprop/pskr/calibration_sampler_test.exs index 8872d818..7a1b42da 100644 --- a/test/microwaveprop/pskr/calibration_sampler_test.exs +++ b/test/microwaveprop/pskr/calibration_sampler_test.exs @@ -70,8 +70,8 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do end describe "build_for_hour/1" do - test "skips and returns 0 when no spots exist for the hour" do - assert CalibrationSampler.build_for_hour(@hour) == 0 + test "skips and returns zeros when no spots exist for the hour" do + assert %{upserted: 0, missing_hrrr_cells: 0} = CalibrationSampler.build_for_hour(@hour) assert Repo.aggregate(CalibrationSample, :count) == 0 end @@ -82,7 +82,7 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do insert_spot!(%{midpoint_lat: 35.0, midpoint_lon: -97.0, sender_grid: "EM15"}) insert_hrrr!(%{lat: 33.0, lon: -97.0}) - assert CalibrationSampler.build_for_hour(@hour) == 2 + assert %{upserted: 2} = CalibrationSampler.build_for_hour(@hour) [a, b] = CalibrationSample |> Repo.all() |> Enum.sort_by(& &1.midpoint_lat) # Two near-midpoints collapsed; one distant cell stayed separate. assert a.spot_count == 4 @@ -138,7 +138,7 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do test "enqueues an hrrr_fetch_tasks row when a cell has no nearby HRRR profile" do insert_spot!(%{midpoint_lat: 33.0, midpoint_lon: -97.0}) - assert CalibrationSampler.build_for_hour(@hour) == 1 + assert %{upserted: 1, missing_hrrr_cells: 1} = CalibrationSampler.build_for_hour(@hour) [task] = Repo.all(from(t in "hrrr_fetch_tasks", select: %{points: t.points, status: t.status, valid_time: t.valid_time})) @@ -215,7 +215,7 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do insert_spot!(%{midpoint_lat: 33.0, midpoint_lon: -97.0}) # ── Pass 1 ───────────────────────────────────────────────── - assert CalibrationSampler.build_for_hour(@hour) == 1 + assert %{upserted: 1, missing_hrrr_cells: 1} = CalibrationSampler.build_for_hour(@hour) [first_sample] = Repo.all(CalibrationSample) assert first_sample.surface_temp_c == nil assert first_sample.hpbl_m == nil @@ -234,7 +234,7 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do }) # ── Pass 2 ───────────────────────────────────────────────── - assert CalibrationSampler.build_for_hour(@hour) == 1 + assert %{upserted: 1, missing_hrrr_cells: 0} = CalibrationSampler.build_for_hour(@hour) [second_sample] = Repo.all(CalibrationSample) # Same row, UPSERTed (id preserved, weather populated). diff --git a/test/microwaveprop/workers/pskr_calibration_worker_test.exs b/test/microwaveprop/workers/pskr_calibration_worker_test.exs index a8f8681e..40c51073 100644 --- a/test/microwaveprop/workers/pskr_calibration_worker_test.exs +++ b/test/microwaveprop/workers/pskr_calibration_worker_test.exs @@ -5,6 +5,7 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorkerTest do alias Microwaveprop.Pskr.CalibrationSample alias Microwaveprop.Pskr.SpotHourly alias Microwaveprop.Repo + alias Microwaveprop.Weather.HrrrProfile alias Microwaveprop.Workers.PskrCalibrationWorker defp insert_spot!(hour_utc, opts \\ []) do @@ -98,4 +99,81 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorkerTest do assert Repo.aggregate(CalibrationSample, :count) == 4 end end + + describe "follow-up scheduling" do + # Self-healing pipeline: when the sampler reports cells whose HRRR + # data isn't yet in the DB (so it just enqueued hrrr_fetch_tasks + # for the Rust hrrr-point-worker to drain), the worker schedules + # a re-run of itself for that exact hour ~10 min later. By that + # point the Rust worker has typically drained the batch, and the + # follow-up re-run UPSERTs the samples with non-NULL HRRR. + + test "schedules a follow-up for hours with missing HRRR data" do + Oban.Testing.with_testing_mode(:manual, fn -> + hour = ~U[2026-05-04 18:00:00Z] + insert_spot!(hour) + + assert :ok = + perform_job(PskrCalibrationWorker, %{"hour_utc" => DateTime.to_iso8601(hour)}) + + assert_enqueued( + worker: PskrCalibrationWorker, + args: %{ + "hour_utc" => DateTime.to_iso8601(hour), + "_follow_up" => true + } + ) + end) + end + + test "does not schedule a follow-up when no cells are missing HRRR" do + # Plant the HRRR profile so the sampler finds it on first pass. + hour = ~U[2026-05-04 18:00:00Z] + insert_spot!(hour) + + {:ok, _} = + %HrrrProfile{} + |> HrrrProfile.changeset(%{ + valid_time: hour, + lat: 33.0, + lon: -97.0, + run_time: hour, + surface_temp_c: 20.0, + surface_dewpoint_c: 15.0, + pwat_mm: 25.0, + surface_pressure_mb: 1013.0, + min_refractivity_gradient: -100.0, + hpbl_m: 500.0, + ducting_detected: false, + profile: [] + }) + |> Repo.insert() + + Oban.Testing.with_testing_mode(:manual, fn -> + assert :ok = + perform_job(PskrCalibrationWorker, %{"hour_utc" => DateTime.to_iso8601(hour)}) + + refute_enqueued(worker: PskrCalibrationWorker) + end) + end + + test "a follow-up re-run does not chain another follow-up (loop terminator)" do + # Even if the follow-up re-run STILL finds missing HRRR (e.g. + # the Rust worker is slow), the `_follow_up: true` sentinel + # prevents chaining a second follow-up. The rolling-window + # cron is the safety net for that case. + Oban.Testing.with_testing_mode(:manual, fn -> + hour = ~U[2026-05-04 18:00:00Z] + insert_spot!(hour) + + assert :ok = + perform_job(PskrCalibrationWorker, %{ + "hour_utc" => DateTime.to_iso8601(hour), + "_follow_up" => true + }) + + refute_enqueued(worker: PskrCalibrationWorker) + end) + end + end end