fix(propagation): hourly grid chain preempts MRMS/prune backlog
The :propagation queue (2 slots) is shared by PropagationGridWorker, MrmsFetchWorker (every 2 min), and PropagationPruneWorker (every 15 min). Oban dispatches priority-then-FIFO, so a post-deploy MRMS/prune backlog could delay the hourly chain step until the siblings drained. Set PropagationGridWorker priority: 0 explicitly and drop the two siblings to priority: 5 so new chain steps always jump ahead. Backfill workers live on separate queues (:hrrr, :weather, :terrain, :iemre, :narr, :radar, :mechanism) and already don't interact with the predictor.
This commit is contained in:
parent
fe0bdd32c2
commit
504f5147f6
4 changed files with 29 additions and 1 deletions
|
|
@ -8,7 +8,10 @@ defmodule Microwaveprop.Workers.MrmsFetchWorker do
|
||||||
Runs on the `propagation` queue so it shares the scoring node's crontab
|
Runs on the `propagation` queue so it shares the scoring node's crontab
|
||||||
and isn't competing with the heavier HRRR pipeline for slots.
|
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.MrmsCache
|
||||||
alias Microwaveprop.Weather.MrmsClient
|
alias Microwaveprop.Weather.MrmsClient
|
||||||
|
|
|
||||||
|
|
@ -20,6 +20,12 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
||||||
|
|
||||||
use Oban.Worker,
|
use Oban.Worker,
|
||||||
queue: :propagation,
|
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
|
# Higher than the default 3 so a few DynamicLifeline rescues
|
||||||
# (e.g., a rolling deploy that kills a mid-flight chain step)
|
# (e.g., a rolling deploy that kills a mid-flight chain step)
|
||||||
# don't exhaust the chain's retry budget and discard the whole
|
# don't exhaust the chain's retry budget and discard the whole
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,10 @@ defmodule Microwaveprop.Workers.PropagationPruneWorker do
|
||||||
|
|
||||||
use Oban.Worker,
|
use Oban.Worker,
|
||||||
queue: :propagation,
|
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,
|
max_attempts: 3,
|
||||||
unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]]
|
unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]]
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -12,7 +12,22 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
|
||||||
use Microwaveprop.DataCase, async: false
|
use Microwaveprop.DataCase, async: false
|
||||||
use Oban.Testing, repo: Microwaveprop.Repo
|
use Oban.Testing, repo: Microwaveprop.Repo
|
||||||
|
|
||||||
|
alias Microwaveprop.Workers.MrmsFetchWorker
|
||||||
alias Microwaveprop.Workers.PropagationGridWorker
|
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
|
describe "perform/1 — chain seeding (empty args)" do
|
||||||
test "enqueues a single fh=0 chain step with a normalized run_time" do
|
test "enqueues a single fh=0 chain step with a normalized run_time" do
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue