Graham McIntire
b08fc5e449
Old design fired hourly at HH:25 and built one whole hour of cells
in a single batch. During PSKR peaks (50k+ spots/hour) that batch
held the DB pool long enough to look like a backlog, and a pod
restart between fires lost a full hour because the next cron only
caught the most-recent-prev hour.
- Default window is now current hour + 1 prior. Every 5-min fire
spreads work across the in-flight hour and always covers the
boundary without waiting for the next cron tick.
- lookback_hours arg widens the window for catch-up
(e.g. {"lookback_hours": 24} sweeps the last day after an
outage).
- Unique period 3600 → 240 so consecutive 5-min cron fires actually
slot in.
UPSERT is already idempotent so the ~12 redundant passes per hour
cost only the read+merge. Latency from spot ingestion to sample
creation drops from up-to-60min to ≤5min.
82 lines
2.9 KiB
Elixir
82 lines
2.9 KiB
Elixir
defmodule Microwaveprop.Workers.PskrCalibrationWorker do
|
||
@moduledoc """
|
||
Cron worker that builds `Pskr.CalibrationSample` rows by joining
|
||
PSKR spot density with HRRR atmospheric state and SWPC space
|
||
weather.
|
||
|
||
## Why frequent + rolling window
|
||
|
||
The naive design — one cron fire per hour, processing only the
|
||
previous hour — has two failure modes:
|
||
|
||
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.
|
||
|
||
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: [
|
||
# 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]
|
||
]
|
||
|
||
alias Microwaveprop.Pskr.CalibrationSampler
|
||
|
||
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: %{"hour_utc" => iso}}) when is_binary(iso) do
|
||
{:ok, hour, _} = DateTime.from_iso8601(iso)
|
||
process_hour(hour)
|
||
:ok
|
||
end
|
||
|
||
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 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
|