Add FreshnessMonitor GenServer to detect and fill stale propagation data
Replaces the one-shot startup Task with a GenServer that checks every 5 minutes whether propagation scores are older than 2 hours. If stale, enqueues a PropagationGridWorker job (with dedup check to avoid double queuing). Disabled in test env to avoid SQL Sandbox conflicts.
This commit is contained in:
parent
45fb9736fa
commit
105298fc3f
3 changed files with 74 additions and 23 deletions
|
|
@ -28,6 +28,7 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
|
|||
|
||||
# Run Oban jobs inline during tests
|
||||
config :microwaveprop, Oban, testing: :inline
|
||||
config :microwaveprop, start_freshness_monitor: false
|
||||
config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Terrain.ElevationClient}, retry: false]
|
||||
config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false]
|
||||
config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false]
|
||||
|
|
|
|||
|
|
@ -5,8 +5,6 @@ defmodule Microwaveprop.Application do
|
|||
|
||||
use Application
|
||||
|
||||
require Logger
|
||||
|
||||
@impl true
|
||||
def start(_type, _args) do
|
||||
migrate()
|
||||
|
|
@ -19,7 +17,7 @@ defmodule Microwaveprop.Application do
|
|||
{Oban, Application.fetch_env!(:microwaveprop, Oban)},
|
||||
# Start to serve requests, typically the last entry
|
||||
MicrowavepropWeb.Endpoint,
|
||||
{Task, &ensure_fresh_scores/0}
|
||||
Microwaveprop.Propagation.FreshnessMonitor
|
||||
]
|
||||
|
||||
# See https://hexdocs.pm/elixir/Supervisor.html
|
||||
|
|
@ -34,26 +32,6 @@ defmodule Microwaveprop.Application do
|
|||
Microwaveprop.Release.migrate()
|
||||
end
|
||||
|
||||
# If propagation scores are older than 2 hours, enqueue a grid worker immediately.
|
||||
# Covers app restarts, deploys, and missed cron ticks.
|
||||
defp ensure_fresh_scores do
|
||||
case Microwaveprop.Propagation.latest_valid_time() do
|
||||
nil ->
|
||||
Logger.info("No propagation scores found, enqueuing grid worker")
|
||||
Oban.insert(Microwaveprop.Workers.PropagationGridWorker.new(%{}))
|
||||
|
||||
latest ->
|
||||
age_minutes = DateTime.diff(DateTime.utc_now(), latest, :minute)
|
||||
|
||||
if age_minutes > 120 do
|
||||
Logger.info("Propagation scores are #{age_minutes}m old, enqueuing grid worker")
|
||||
Oban.insert(Microwaveprop.Workers.PropagationGridWorker.new(%{}))
|
||||
else
|
||||
Logger.info("Propagation scores are #{age_minutes}m old, fresh enough")
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
def config_change(changed, _new, removed) do
|
||||
MicrowavepropWeb.Endpoint.config_change(changed, removed)
|
||||
|
|
|
|||
72
lib/microwaveprop/propagation/freshness_monitor.ex
Normal file
72
lib/microwaveprop/propagation/freshness_monitor.ex
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
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 :timer.minutes(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"]
|
||||
)
|
||||
)
|
||||
|
||||
unless pending do
|
||||
Oban.insert(PropagationGridWorker.new(%{}))
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue