diff --git a/config/test.exs b/config/test.exs index 4d13c776..cdc7a1ac 100644 --- a/config/test.exs +++ b/config/test.exs @@ -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] diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 310ebb1c..f62075f9 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -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) diff --git a/lib/microwaveprop/propagation/freshness_monitor.ex b/lib/microwaveprop/propagation/freshness_monitor.ex new file mode 100644 index 00000000..1b2647e6 --- /dev/null +++ b/lib/microwaveprop/propagation/freshness_monitor.ex @@ -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