diff --git a/lib/microwaveprop/workers/mrms_fetch_worker.ex b/lib/microwaveprop/workers/mrms_fetch_worker.ex index ce2166ff..ab42fa7e 100644 --- a/lib/microwaveprop/workers/mrms_fetch_worker.ex +++ b/lib/microwaveprop/workers/mrms_fetch_worker.ex @@ -8,7 +8,10 @@ defmodule Microwaveprop.Workers.MrmsFetchWorker do Runs on the `propagation` queue so it shares the scoring node's crontab and isn't competing with the heavier HRRR pipeline for slots. """ - use Oban.Worker, queue: :propagation, max_attempts: 2 + # Lower priority than PropagationGridWorker so an MRMS backlog + # (cron fires every 2 minutes) doesn't starve the hourly chain + # on the shared :propagation queue. + use Oban.Worker, queue: :propagation, priority: 5, max_attempts: 2 alias Microwaveprop.Weather.MrmsCache alias Microwaveprop.Weather.MrmsClient diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index bb4f3c0a..14eca6bd 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -20,6 +20,12 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do use Oban.Worker, queue: :propagation, + # Highest priority on the shared :propagation queue. MrmsFetchWorker + # and PropagationPruneWorker also live here; a backlog of those + # (e.g. after a deploy, or MRMS firing every 2 minutes) would + # otherwise starve the hourly chain because Oban dispatches + # priority-then-FIFO. Explicit so the invariant is visible. + priority: 0, # Higher than the default 3 so a few DynamicLifeline rescues # (e.g., a rolling deploy that kills a mid-flight chain step) # don't exhaust the chain's retry budget and discard the whole diff --git a/lib/microwaveprop/workers/propagation_prune_worker.ex b/lib/microwaveprop/workers/propagation_prune_worker.ex index a9c563af..1d24a18c 100644 --- a/lib/microwaveprop/workers/propagation_prune_worker.ex +++ b/lib/microwaveprop/workers/propagation_prune_worker.ex @@ -11,6 +11,10 @@ defmodule Microwaveprop.Workers.PropagationPruneWorker do use Oban.Worker, queue: :propagation, + # Lower priority than PropagationGridWorker so the hourly chain + # never waits behind a pending prune on the shared :propagation + # queue (2 slots). + priority: 5, max_attempts: 3, unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]] diff --git a/test/microwaveprop/workers/propagation_grid_worker_test.exs b/test/microwaveprop/workers/propagation_grid_worker_test.exs index 885be7d8..beabfb74 100644 --- a/test/microwaveprop/workers/propagation_grid_worker_test.exs +++ b/test/microwaveprop/workers/propagation_grid_worker_test.exs @@ -12,7 +12,22 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do use Microwaveprop.DataCase, async: false use Oban.Testing, repo: Microwaveprop.Repo + alias Microwaveprop.Workers.MrmsFetchWorker alias Microwaveprop.Workers.PropagationGridWorker + alias Microwaveprop.Workers.PropagationPruneWorker + + describe "queue priority" do + test "PropagationGridWorker runs at the highest priority on :propagation" do + assert PropagationGridWorker.__opts__()[:priority] == 0 + end + + test "MrmsFetchWorker and PropagationPruneWorker yield to the grid chain" do + # Same :propagation queue — must be lower priority so hourly chain + # steps jump ahead of a MRMS or pruner backlog. + assert MrmsFetchWorker.__opts__()[:priority] > 0 + assert PropagationPruneWorker.__opts__()[:priority] > 0 + end + end describe "perform/1 — chain seeding (empty args)" do test "enqueues a single fh=0 chain step with a normalized run_time" do