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.
78 lines
2.8 KiB
Elixir
78 lines
2.8 KiB
Elixir
defmodule Microwaveprop.Workers.IonosphereFetchWorker do
|
|
@moduledoc """
|
|
Polls GIRO DIDBase for real-time ionosonde measurements (foF2, foE,
|
|
foEs, hmF2, MUFD) from a handful of CONUS stations. Drives the 144
|
|
MHz sporadic-E and HF MUF scoring inputs.
|
|
|
|
Runs on a ~10-minute cron — GIRO publishes each station at 7.5-minute
|
|
cadence so a 10-minute poll guarantees we pick up every new sample
|
|
within one cycle without hammering the endpoint.
|
|
|
|
Pulls a 2-hour window each poll to catch any late-arriving samples
|
|
and to be resilient to the occasional missed run. Upserts are
|
|
idempotent on (station_code, valid_time).
|
|
"""
|
|
|
|
use Oban.Worker,
|
|
queue: :ionosphere,
|
|
max_attempts: 3,
|
|
unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]]
|
|
|
|
alias Microwaveprop.Ionosphere
|
|
alias Microwaveprop.Ionosphere.GiroClient
|
|
|
|
require Logger
|
|
|
|
# CONUS stations that have been publishing reliably. Others (Boulder,
|
|
# Wallops, Austin, Idaho, Point Arguello) are in the GIRO catalog but
|
|
# intermittent — add them here once they come back online.
|
|
@stations [
|
|
%{code: "MHJ45", name: "Millstone Hill MA", lat: 42.6, lon: -71.5},
|
|
%{code: "AL945", name: "Alpena MI", lat: 45.1, lon: -83.6}
|
|
]
|
|
|
|
@lookback_seconds 2 * 3600
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{}) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
from_dt = DateTime.add(now, -@lookback_seconds, :second)
|
|
|
|
Enum.each(@stations, fn station ->
|
|
fetch_and_upsert(station, from_dt, now)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc false
|
|
@spec stations() :: [map()]
|
|
def stations, do: @stations
|
|
|
|
defp fetch_and_upsert(%{code: code, name: name}, from_dt, to_dt) do
|
|
case GiroClient.fetch(code, from_dt, to_dt) do
|
|
{:ok, observations} ->
|
|
{:ok, count} = Ionosphere.upsert_observations(code, observations)
|
|
Logger.info("IonosphereFetch: #{code} (#{name}) upserted #{count} observations")
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("IonosphereFetch: #{code} (#{name}) failed: #{format_reason(reason)}")
|
|
end
|
|
end
|
|
|
|
# GIRO's TLS cert chains through a CA that isn't always in the pod's
|
|
# bundle, surfacing as a Req.TransportError with a multi-line erl_alert
|
|
# string. Inspecting the full struct dumps ~400 chars of stacktrace-y
|
|
# noise per station per 10-min cycle. Compress to something grep-able.
|
|
defp format_reason("GIRO request failed: " <> tail) do
|
|
cond do
|
|
String.contains?(tail, ":unknown_ca") -> "TLS unknown_ca (CA bundle missing)"
|
|
String.contains?(tail, ":timeout") -> "timeout"
|
|
String.contains?(tail, ":econnrefused") -> "connection refused"
|
|
true -> String.slice(tail, 0, 120)
|
|
end
|
|
end
|
|
|
|
defp format_reason(reason) when is_binary(reason), do: reason
|
|
defp format_reason(reason), do: inspect(reason, printable_limit: 200)
|
|
end
|