diff --git a/config/runtime.exs b/config/runtime.exs index 8eb460a3..a3aa50f1 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -310,13 +310,15 @@ if config_env() == :prod do # deploy clears it or new profiles need to fold into the means. # 02:30 UTC sits inside the quiet pre-NARR/HRRR ingestion gap. {"30 2 * * *", Microwaveprop.Workers.AdminTaskWorker, args: %{"task" => "climatology"}}, - # Build the PSKR calibration corpus one hour at a time. HH:25 - # is past the HRRR analysis publish window (~HH:50 of the - # previous hour, surfaces by HH:05) and past PSKR's 60s - # aggregator flush, so the previous-hour sample has the - # atmospheric and signal data it needs. Idempotent UPSERT, so - # missed fires are safe to backfill manually. - {"25 * * * *", Microwaveprop.Workers.PskrCalibrationWorker}, + # Build the PSKR calibration corpus *early and often*: every + # 5 min, process the current hour + the prior hour. Each pass + # spreads the work across the hour as spots accumulate so we + # never bind a giant batch at the boundary, and the prior- + # hour pass guarantees the boundary is covered without + # waiting for the next fire. UPSERT is idempotent so the + # ~12 redundant passes per hour cost only the read+merge. + # Longer outages can be backfilled with `lookback_hours`. + {"*/5 * * * *", Microwaveprop.Workers.PskrCalibrationWorker}, # Weekly recalibration analysis pass. The recalibrator # checks corpus density and bins each scoring feature × # band when the corpus is large enough; otherwise records diff --git a/lib/microwaveprop/workers/pskr_calibration_worker.ex b/lib/microwaveprop/workers/pskr_calibration_worker.ex index 246c286e..40b5469d 100644 --- a/lib/microwaveprop/workers/pskr_calibration_worker.ex +++ b/lib/microwaveprop/workers/pskr_calibration_worker.ex @@ -1,29 +1,44 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorker do @moduledoc """ - Hourly cron that builds `Pskr.CalibrationSample` rows by joining + Cron worker that builds `Pskr.CalibrationSample` rows by joining PSKR spot density with HRRR atmospheric state and SWPC space - weather at a given hour. + weather. - Default behaviour fires for the previous full hour — by HH:25 UTC - the HRRR cycle has analysis available and PSKR's hourly aggregator - has flushed its accumulator. Pass `"hour_utc"` in the args map to - rerun a specific hour: + ## Why frequent + rolling window - Microwaveprop.Workers.PskrCalibrationWorker.new(%{"hour_utc" => "2026-05-04T18:00:00Z"}) + The naive design — one cron fire per hour, processing only the + previous hour — has two failure modes: - The sampler is idempotent (UPSERT on `(hour_utc, band, midpoint_lat, - midpoint_lon)`) so reruns are safe. + 1. The whole hour's worth of cells is built in a single batch. + During peaks (50k+ spots/hour) one batch holds the DB pool + long enough to look like a backlog. + 2. A pod restart between fires drops a full hour because the + next cron only catches the most-recent-prev hour. - Lives on the `:backfill_enqueue` queue rather than spawning a new - one — the workload is small (~1 query window, in-memory join, - bulk upsert) and cluster-wide single-slot is plenty for one fire - per hour. + Instead we fire every 5 minutes and process the **current hour + plus one prior hour** by default. Idempotent UPSERT on + `(hour_utc, band, midpoint_lat, midpoint_lon)` means each pass + merges new spots into existing rows; the work spreads across the + hour as spots arrive, and the boundary always has at least one + cron pass covering both sides without waiting for an hourly fire. + + ## Manual reruns + + - `hour_utc` (ISO 8601) — process exactly that one hour. Uniqueness + is keyed on args, so manual reruns slot in alongside cron fires. + - `lookback_hours` (integer) — override the default window size. + Useful after longer outages: `%{"lookback_hours" => 24}` will + sweep the last day. """ use Oban.Worker, queue: :backfill_enqueue, max_attempts: 3, unique: [ - period: 3600, + # Cron is every 5 min (300 s); 240 s window means the unique + # slot has expired by the time the next fire arrives. Manual + # reruns with explicit args use a different args fingerprint + # and bypass this check entirely. + period: 240, states: [:available, :scheduled, :executing, :retryable] ] @@ -31,26 +46,37 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorker do require Logger + # Default rolling window: current hour + 1 prior. Wide enough to + # cover a 5–10 min cron interval crossing an hour boundary; not + # so wide that every fire burns DB time on hours that are already + # complete. Use `lookback_hours` to widen for catch-up. + @default_lookback_hours 1 + @impl Oban.Worker - def perform(%Oban.Job{args: args}) do - hour = parse_hour(args) - Logger.info("PskrCalibrationWorker: building samples for #{hour}") - count = CalibrationSampler.build_for_hour(hour) - Logger.info("PskrCalibrationWorker: upserted #{count} samples for #{hour}") + def perform(%Oban.Job{args: %{"hour_utc" => iso}}) when is_binary(iso) do + {:ok, hour, _} = DateTime.from_iso8601(iso) + process_hour(hour) :ok end - defp parse_hour(%{"hour_utc" => iso}) when is_binary(iso) do - {:ok, dt, _} = DateTime.from_iso8601(iso) - dt + def perform(%Oban.Job{args: args}) do + lookback = lookback_hours(args) + now = truncate_to_hour(DateTime.utc_now()) + + Enum.each(0..lookback, fn offset -> + now |> DateTime.add(-offset * 3600, :second) |> process_hour() + end) + + :ok end - defp parse_hour(_) do - # Default: the previous fully-completed hour. - DateTime.utc_now() - |> DateTime.add(-3600, :second) - |> Map.put(:minute, 0) - |> Map.put(:second, 0) - |> Map.put(:microsecond, {0, 0}) + 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}") end + + defp truncate_to_hour(dt), do: %{dt | minute: 0, second: 0, microsecond: {0, 0}} end diff --git a/test/microwaveprop/workers/pskr_calibration_worker_test.exs b/test/microwaveprop/workers/pskr_calibration_worker_test.exs index b5bc6dc2..a8f8681e 100644 --- a/test/microwaveprop/workers/pskr_calibration_worker_test.exs +++ b/test/microwaveprop/workers/pskr_calibration_worker_test.exs @@ -7,56 +7,95 @@ defmodule Microwaveprop.Workers.PskrCalibrationWorkerTest do alias Microwaveprop.Repo alias Microwaveprop.Workers.PskrCalibrationWorker - defp insert_spot!(hour_utc) do + defp insert_spot!(hour_utc, opts \\ []) do + sender_grid = Keyword.get(opts, :sender_grid, "EM12KL") + midpoint_lat = Keyword.get(opts, :midpoint_lat, 33.0) + midpoint_lon = Keyword.get(opts, :midpoint_lon, -97.0) + {:ok, _} = %SpotHourly{} |> SpotHourly.changeset(%{ hour_utc: hour_utc, band: "10000", - sender_grid: "EM12KL", + sender_grid: sender_grid, receiver_grid: "DM43ST", spot_count: 1, - midpoint_lat: 33.0, - midpoint_lon: -97.0, + midpoint_lat: midpoint_lat, + midpoint_lon: midpoint_lon, distance_km: 200.0, modes: ["FT8"] }) |> Repo.insert() end - test "perform/1 with explicit hour_utc samples that hour" do - hour = ~U[2026-05-04 18:00:00Z] - insert_spot!(hour) + defp truncate_to_hour(dt), do: %{dt | minute: 0, second: 0, microsecond: {0, 0}} - assert :ok = - perform_job(PskrCalibrationWorker, %{"hour_utc" => DateTime.to_iso8601(hour)}) + describe "perform/1 with explicit hour_utc" do + test "samples just that hour" do + hour = ~U[2026-05-04 18:00:00Z] + insert_spot!(hour) - [sample] = Repo.all(CalibrationSample) - assert sample.hour_utc == hour + assert :ok = + perform_job(PskrCalibrationWorker, %{"hour_utc" => DateTime.to_iso8601(hour)}) + + [sample] = Repo.all(CalibrationSample) + assert sample.hour_utc == hour + end + + test "succeeds when the hour has no spots" do + assert :ok = + perform_job(PskrCalibrationWorker, %{"hour_utc" => "2026-05-04T03:00:00Z"}) + + assert Repo.aggregate(CalibrationSample, :count) == 0 + end end - test "perform/1 with no args defaults to the previous full hour" do - # Pick the hour just before "now" — the same logic the worker - # uses internally — so the spot we insert lines up with what the - # default arg path will pull. - target = - DateTime.utc_now() - |> DateTime.add(-3600, :second) - |> Map.put(:minute, 0) - |> Map.put(:second, 0) - |> Map.put(:microsecond, {0, 0}) + describe "perform/1 default rolling window" do + test "samples both the current hour and the previous hour" do + # Frequent processing means a single fire must cover the hour + # boundary: spots in the just-ended hour and spots in the + # in-progress hour both get sample rows. Otherwise spots that + # arrive in the last few minutes of an hour wait up to a full + # cron interval before they're sampled. + now = truncate_to_hour(DateTime.utc_now()) + previous = DateTime.add(now, -3600, :second) - insert_spot!(target) - assert :ok = perform_job(PskrCalibrationWorker, %{}) - assert Repo.aggregate(CalibrationSample, :count) == 1 + insert_spot!(now) + insert_spot!(previous, sender_grid: "EM10AA") + + assert :ok = perform_job(PskrCalibrationWorker, %{}) + + hours = CalibrationSample |> Repo.all() |> Enum.map(& &1.hour_utc) |> Enum.sort() + assert hours == [previous, now] + end + + test "is idempotent — re-running merges into the same rows" do + now = truncate_to_hour(DateTime.utc_now()) + insert_spot!(now) + + assert :ok = perform_job(PskrCalibrationWorker, %{}) + first_count = Repo.aggregate(CalibrationSample, :count) + + assert :ok = perform_job(PskrCalibrationWorker, %{}) + assert Repo.aggregate(CalibrationSample, :count) == first_count + end end - test "perform/1 returns :ok even when no spots exist for the hour" do - assert :ok = - perform_job(PskrCalibrationWorker, %{ - "hour_utc" => "2026-05-04T03:00:00Z" - }) + describe "perform/1 with explicit lookback_hours" do + test "extends the rolling window to backfill recent gaps" do + # Operator sets lookback_hours=3 to recover from a short pod + # outage. The worker fires once and produces samples for the + # current hour plus the three before it. + now = truncate_to_hour(DateTime.utc_now()) - assert Repo.aggregate(CalibrationSample, :count) == 0 + Enum.each(0..3, fn offset -> + hour = DateTime.add(now, -offset * 3600, :second) + insert_spot!(hour, sender_grid: "EM1#{offset}AA") + end) + + assert :ok = perform_job(PskrCalibrationWorker, %{"lookback_hours" => 3}) + + assert Repo.aggregate(CalibrationSample, :count) == 4 + end end end