diff --git a/config/config.exs b/config/config.exs index 71c4446d..89b7dac8 100644 --- a/config/config.exs +++ b/config/config.exs @@ -98,6 +98,7 @@ config :microwaveprop, Oban, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, + {"15 * * * *", Microwaveprop.Workers.GridCachePruneWorker}, # UWYO publishes the 00Z/12Z Canadian radiosondes ~90 minutes after launch {"30 1 * * *", CanadianSoundingFetchWorker}, {"30 13 * * *", CanadianSoundingFetchWorker}, diff --git a/config/dev.exs b/config/dev.exs index 74b0afa7..2c2bd08f 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -97,6 +97,7 @@ config :microwaveprop, Oban, crontab: [ {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, + {"15 * * * *", Microwaveprop.Workers.GridCachePruneWorker}, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} ]} ] diff --git a/config/runtime.exs b/config/runtime.exs index 58580e0e..509cf408 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -205,6 +205,7 @@ if config_env() == :prod do # (25 jobs per run, 100/day) into the :gefs queue. {"30 5,11,17,23 * * *", Microwaveprop.Workers.GefsFetchWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, + {"15 * * * *", Microwaveprop.Workers.GridCachePruneWorker}, # The :narr type is virtual: it targets contacts with hrrr_status = # :unavailable (pre-2014, missing from the HRRR archive) and dispatches # NarrFetchWorker against NCEI. See narr_jobs_for_contact/1. diff --git a/lib/microwaveprop/workers/grid_cache_prune_worker.ex b/lib/microwaveprop/workers/grid_cache_prune_worker.ex new file mode 100644 index 00000000..ac9d2d14 --- /dev/null +++ b/lib/microwaveprop/workers/grid_cache_prune_worker.ex @@ -0,0 +1,28 @@ +defmodule Microwaveprop.Workers.GridCachePruneWorker do + @moduledoc """ + Prunes stale rows from the node-local `Microwaveprop.Weather.GridCache` ETS + table. Without this cron the cache grew unbounded: every hourly + `PropagationGridWorker` run plus peer `broadcast_put` added an entry per + forecast hour (~92k points * 19 hours), so long-lived pods eventually OOM'd. + + Runs on the shared `:propagation` queue alongside `PropagationPruneWorker` + at a lower priority so the hourly grid chain never waits behind it. + """ + + use Oban.Worker, + queue: :propagation, + priority: 5, + max_attempts: 3, + unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]] + + alias Microwaveprop.Weather.GridCache + + @cutoff_hours 3 + + @impl Oban.Worker + def perform(%Oban.Job{}) do + cutoff = DateTime.add(DateTime.utc_now(), -@cutoff_hours, :hour) + _removed = GridCache.prune_older_than(cutoff) + :ok + end +end diff --git a/test/microwaveprop/workers/grid_cache_prune_worker_test.exs b/test/microwaveprop/workers/grid_cache_prune_worker_test.exs new file mode 100644 index 00000000..9bad223c --- /dev/null +++ b/test/microwaveprop/workers/grid_cache_prune_worker_test.exs @@ -0,0 +1,40 @@ +defmodule Microwaveprop.Workers.GridCachePruneWorkerTest do + use ExUnit.Case, async: false + + alias Microwaveprop.Weather.GridCache + alias Microwaveprop.Workers.GridCachePruneWorker + + setup do + GridCache.clear() + :ok + end + + describe "perform/1" do + test "removes cache entries with valid_time older than 3 hours" do + now = DateTime.utc_now() + old = now |> DateTime.add(-3, :hour) |> DateTime.add(-1, :second) |> DateTime.truncate(:second) + fresh = now |> DateTime.add(-30, :minute) |> DateTime.truncate(:second) + future = now |> DateTime.add(6, :hour) |> DateTime.truncate(:second) + + GridCache.put(old, [%{lat: 32.0, lon: -97.0, temperature: 10.0}]) + GridCache.put(fresh, [%{lat: 32.0, lon: -97.0, temperature: 20.0}]) + GridCache.put(future, [%{lat: 32.0, lon: -97.0, temperature: 30.0}]) + + assert :ok = GridCachePruneWorker.perform(%Oban.Job{}) + + assert GridCache.fetch(old) == :miss + assert {:ok, [%{temperature: 20.0}]} = GridCache.fetch(fresh) + assert {:ok, [%{temperature: 30.0}]} = GridCache.fetch(future) + end + + test "is a no-op when nothing is stale" do + fresh = + DateTime.utc_now() |> DateTime.add(-30, :minute) |> DateTime.truncate(:second) + + GridCache.put(fresh, [%{lat: 32.0, lon: -97.0, temperature: 42.0}]) + + assert :ok = GridCachePruneWorker.perform(%Oban.Job{}) + assert {:ok, [%{temperature: 42.0}]} = GridCache.fetch(fresh) + end + end +end