Graham McIntire
40d1fa03aa
- Add enqueue_for_qso/1 to directly enqueue weather/HRRR/terrain/IEMRE jobs for a single user-submitted QSO (no cron, no bulk processing) - Submit flow calls enqueue_for_qso instead of generic enqueue worker - Add enrichment queues to prod config for on-demand processing - Guard against HRRR fill values in store_hrrr_profiles (fixes badarith) - Filter QSOs without pos2 in build_terrain_jobs
72 lines
1.8 KiB
Elixir
72 lines
1.8 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
|
|
|
|
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 ->
|
|
Logger.info("FreshnessMonitor: no scores found, enqueuing grid worker")
|
|
enqueue_if_not_queued()
|
|
|
|
latest ->
|
|
age = DateTime.diff(DateTime.utc_now(), latest, :minute)
|
|
|
|
if age > @stale_threshold_minutes do
|
|
Logger.info("FreshnessMonitor: scores are #{age}m old, enqueuing grid worker")
|
|
enqueue_if_not_queued()
|
|
end
|
|
end
|
|
end
|
|
|
|
defp enqueue_if_not_queued do
|
|
# Only enqueue if there isn't already a pending/running grid worker job
|
|
import Ecto.Query
|
|
|
|
pending =
|
|
Microwaveprop.Repo.exists?(
|
|
from(j in "oban_jobs",
|
|
where:
|
|
j.worker == "Microwaveprop.Workers.PropagationGridWorker" and
|
|
j.state in ["available", "scheduled", "executing"]
|
|
)
|
|
)
|
|
|
|
if !pending do
|
|
Oban.insert(PropagationGridWorker.new(%{}))
|
|
end
|
|
end
|
|
end
|