graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(pskr): self-healing follow-up sampler runs for hours with missing HRRR

Browse source 
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.
This commit is contained in:
Graham McIntire 2026-05-05 18:00:10 -05:00
parent 7ee19c5b6a
commit fe2ad2df15
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
4 changed files with 179 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
lib/microwaveprop/pskr/calibration_sampler.ex
View file

@ -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

64
lib/microwaveprop/workers/pskr_calibration_worker.ex
View file

@ -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}}

12
test/microwaveprop/pskr/calibration_sampler_test.exs
View file

@ -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).

78
test/microwaveprop/workers/pskr_calibration_worker_test.exs
View file

@ -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