Graham McIntire
e9a38623d8
Batched from a system-wide review pass.
**Rate-limit + transient-error log hygiene**
- WeatherFetchWorker: classify IEM HTTP 429 as {:snooze, 300} instead
of {:error}. Stops Oban.PerformError stack-trace spam on every
routine rate-limit during backfill, keeps the retry semantics.
- IonosphereFetchWorker: compress GIRO TLS :unknown_ca error from a
~400-char inspect blob to 'TLS unknown_ca (CA bundle missing)'.
- FreshnessMonitor: log only when an enqueue actually lands (the
Oban unique conflict case was silently dropping jobs but still
producing 'enqueuing grid worker' info lines every 5 minutes).
**Perf**
- Rust grid_level_keys(): pre-format 'TMP:{p} mb' / DPT / HGT keys
once via OnceLock instead of format!()'ing per-cell. Removes ~3.6M
String allocations per f01..f18 chain step across pipeline.rs,
hrrr_points.rs. Same for the wgrib2 :(TMP|DPT|HGT): pattern in
hrrr_points.process_batch (sfc_pattern / prs_pattern OnceLock).
- MapLive preload_forecast: Task.async_stream with max_concurrency=4
replaces the 18-wide Enum.map serial walk. Forecast cache warms
~4× faster after a band change, with ordering preserved.
- grid_tasks: partial composite index on (run_time DESC,
forecast_hour ASC) WHERE status='queued' AND kind='forecast',
matching the Rust claim query's ORDER BY. Drops the old
status-only partial that forced a sort per claim.
**Correctness**
- ContactImportWorker: add Oban unique:[keys: [:import_run_id,
:offset]]. Was missing on a worker whose perform() does a
non-idempotent atomic counter increment — a retry would double-
count imported rows.
- CommonVolumeRadarWorker: x_min..x_max default step is -1 when
x_min > x_max, triggering a runtime deprecation warning. Force
Range.new(_, _, 1) explicitly.
**Cleanup**
- Drop unused tmp_dir parameter from hrrr_point_worker + its
process_batch signature.
72 lines
2.1 KiB
Elixir
72 lines
2.1 KiB
Elixir
defmodule Microwaveprop.Propagation.FreshnessMonitor do
|
|
@moduledoc """
|
|
Monitors propagation score freshness and enqueues grid worker jobs
|
|
when data is stale. Checks every 5 minutes. Covers missed cron ticks,
|
|
slow deploys, worker crashes, and any other gap in hourly scoring.
|
|
"""
|
|
|
|
use GenServer
|
|
|
|
alias Microwaveprop.Propagation
|
|
alias Microwaveprop.Workers.PropagationGridWorker
|
|
|
|
require Logger
|
|
|
|
@check_interval to_timeout(minute: 5)
|
|
@stale_threshold_minutes 120
|
|
|
|
@spec start_link(keyword()) :: GenServer.on_start() | :ignore
|
|
def start_link(_opts) do
|
|
if Application.get_env(:microwaveprop, :start_freshness_monitor, true) do
|
|
GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
|
|
else
|
|
:ignore
|
|
end
|
|
end
|
|
|
|
@impl true
|
|
def init(:ok) do
|
|
send(self(), :check)
|
|
{:ok, %{}}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(:check, state) do
|
|
_ = check_freshness()
|
|
Process.send_after(self(), :check, @check_interval)
|
|
{:noreply, state}
|
|
end
|
|
|
|
defp check_freshness do
|
|
case Propagation.latest_valid_time() do
|
|
nil -> maybe_enqueue(reason: "no scores found", age: nil)
|
|
latest -> check_stale(latest)
|
|
end
|
|
end
|
|
|
|
defp check_stale(latest) do
|
|
age = DateTime.diff(DateTime.utc_now(), latest, :minute)
|
|
|
|
if age > @stale_threshold_minutes do
|
|
maybe_enqueue(reason: "scores #{age}m old", age: age)
|
|
end
|
|
end
|
|
|
|
# PropagationGridWorker declares `unique:` over a 1-hour window on
|
|
# the seed args (`%{}`), so repeated 5-minute ticks during a long
|
|
# outage collapse into a single seed job — not one stacked chain
|
|
# per tick. Only log when a new job actually lands; the steady-state
|
|
# "still stale, still queued" case is uninteresting.
|
|
defp maybe_enqueue(reason: reason, age: _age) do
|
|
case Oban.insert(PropagationGridWorker.new(%{})) do
|
|
{:ok, %Oban.Job{conflict?: false}} ->
|
|
Logger.info("FreshnessMonitor: #{reason}, enqueued grid worker")
|
|
|
|
{:ok, %Oban.Job{conflict?: true}} ->
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("FreshnessMonitor: enqueue failed: #{inspect(reason, printable_limit: 200)}")
|
|
end
|
|
end
|
|
end
|