test(pskr): two-pass HRRR pipeline integration test + tighter specs

Adds the test that verifies the full producer→drain→re-pass loop:
sample written with NULL HRRR fields + fetch task enqueued, then once
an HRRR profile lands, the next sampler pass UPSERTs the same
calibration_samples row (id preserved) with non-NULL weather data.

Also tightens dialyzer surface:

* PartitionManager defines a `result()` typedoc and uses it on both
  public functions instead of inlined tuple types.
* PartitionMaintenanceWorker.perform/1 + the lookahead/1 helper get
  explicit specs.
* CalibrationSampler.enqueue_missing_hrrr/3 gets a spec covering its
  group_by-shape input map and HRRR profile list.

`mix precommit` clean (3310 tests, 0 failures). Local `mix dialyzer`
hits an OTP 28.4 vs 28.5 toolchain mismatch unrelated to this change.
This commit is contained in:
Graham McIntire 2026-05-05 17:31:17 -05:00
parent a6a2c859b4
commit f78d4ccb9f
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
4 changed files with 56 additions and 6 deletions

View file

@ -40,6 +40,14 @@ defmodule Microwaveprop.PartitionManager do
@default_parents ~w(hrrr_profiles hrdps_profiles)
@typedoc """
Per-partition result. `parent` is the parent table name, `name` is
the resolved child partition name, and the status flags whether
this call performed the CREATE or found an existing partition that
already covered the range.
"""
@type result :: {parent :: String.t(), name :: String.t(), :created | :exists}
@doc """
For each configured parent table, ensure partitions covering the
current quarter and the next `lookahead_quarters` quarters exist.
@ -47,8 +55,7 @@ defmodule Microwaveprop.PartitionManager do
Returns a list of `{parent, partition_name, :created | :exists}`
entries so the worker can log a single summary line.
"""
@spec ensure_quarterly_partitions(non_neg_integer(), [String.t()]) ::
[{String.t(), String.t(), :created | :exists}]
@spec ensure_quarterly_partitions(non_neg_integer(), [String.t()]) :: [result()]
def ensure_quarterly_partitions(lookahead_quarters \\ 4, parents \\ @default_parents)
when is_integer(lookahead_quarters) and lookahead_quarters >= 0 and is_list(parents) do
today = Date.utc_today()
@ -59,10 +66,8 @@ defmodule Microwaveprop.PartitionManager do
end
end
# Public for direct access from tests / mix tasks. Returns
# `{parent, partition_name, :created | :exists}`.
@spec ensure_partition(String.t(), Date.t(), Date.t()) ::
{String.t(), String.t(), :created | :exists}
# Public for direct access from tests / mix tasks.
@spec ensure_partition(String.t(), Date.t(), Date.t()) :: result()
def ensure_partition(parent, %Date{} = q_start, %Date{} = q_end) do
name = partition_name(parent, q_start)

View file

@ -150,6 +150,7 @@ 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(map(), DateTime.t(), [map()]) :: :ok
defp enqueue_missing_hrrr(cells, hour_utc, hrrr_index) do
points =
cells

View file

@ -28,6 +28,7 @@ defmodule Microwaveprop.Workers.PartitionMaintenanceWorker do
@lookahead_quarters 4
@impl Oban.Worker
@spec perform(Oban.Job.t()) :: :ok
def perform(%Oban.Job{args: args}) do
lookahead = lookahead(args)
results = PartitionManager.ensure_quarterly_partitions(lookahead)
@ -46,6 +47,7 @@ defmodule Microwaveprop.Workers.PartitionMaintenanceWorker do
:ok
end
@spec lookahead(map()) :: non_neg_integer()
defp lookahead(%{"lookahead_quarters" => n}) when is_integer(n) and n >= 0, do: n
defp lookahead(_), do: @lookahead_quarters
end

View file

@ -205,6 +205,48 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do
assert Enum.sort(sample.modes) == ["FT4", "FT8"]
end
test "two-pass pipeline: first run NULL+enqueue, second run UPSERTs with HRRR" do
# End-to-end behaviour the producer was added for: a cell with no
# nearby HRRR profile gets a sample row written with NULL weather
# fields *and* a fetch task enqueued; once the Rust hrrr-point-worker
# delivers the profile (simulated here by directly inserting an
# hrrr_profiles row), the next sampler pass UPSERTs the same
# calibration_samples row, this time with non-NULL HRRR fields.
insert_spot!(%{midpoint_lat: 33.0, midpoint_lon: -97.0})
# ── Pass 1 ─────────────────────────────────────────────────
assert CalibrationSampler.build_for_hour(@hour) == 1
[first_sample] = Repo.all(CalibrationSample)
assert first_sample.surface_temp_c == nil
assert first_sample.hpbl_m == nil
assert Repo.aggregate(from(t in "hrrr_fetch_tasks"), :count) == 1
# ── Rust worker drains the task (simulated) ────────────────
insert_hrrr!(%{
lat: 33.0,
lon: -97.0,
surface_temp_c: 24.0,
surface_dewpoint_c: 18.0,
pwat_mm: 31.0,
min_refractivity_gradient: -120.0,
hpbl_m: 480.0,
ducting_detected: true
})
# ── Pass 2 ─────────────────────────────────────────────────
assert CalibrationSampler.build_for_hour(@hour) == 1
[second_sample] = Repo.all(CalibrationSample)
# Same row, UPSERTed (id preserved, weather populated).
assert second_sample.id == first_sample.id
assert second_sample.surface_temp_c == 24.0
assert second_sample.surface_dewpoint_c == 18.0
assert second_sample.pwat_mm == 31.0
assert second_sample.min_refractivity_gradient == -120.0
assert second_sample.hpbl_m == 480.0
assert second_sample.hrrr_ducting_detected == true
end
test "is idempotent — re-running the same hour upserts the row" do
insert_spot!(%{})
insert_hrrr!(%{lat: 33.0, lon: -97.0})