Graham McIntire
504f5147f6
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.
71 lines
2.7 KiB
Elixir
71 lines
2.7 KiB
Elixir
defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
|
||
@moduledoc """
|
||
Tests the chain-orchestration behavior of PropagationGridWorker.
|
||
|
||
The worker processes forecast hours f00–f18 across the CONUS grid,
|
||
but a single full sweep takes ~2 hours of wall time — longer than a
|
||
typical pod restart window. To survive deploys, the worker processes
|
||
ONE forecast hour per `perform/1` call and enqueues the next hour as
|
||
a fresh Oban job. The tests here cover the dispatch + chain logic
|
||
without mocking the full HRRR / scoring stack.
|
||
"""
|
||
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
|
||
Oban.Testing.with_testing_mode(:manual, fn ->
|
||
assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}})
|
||
|
||
[child] = all_enqueued(worker: PropagationGridWorker)
|
||
assert child.args["forecast_hour"] == 0
|
||
assert is_binary(child.args["run_time"])
|
||
|
||
{:ok, run_time, _} = DateTime.from_iso8601(child.args["run_time"])
|
||
assert run_time.minute == 0
|
||
assert run_time.second == 0
|
||
assert DateTime.before?(run_time, DateTime.utc_now())
|
||
end)
|
||
end
|
||
end
|
||
|
||
describe "enqueue_next_step/2" do
|
||
test "enqueues fh+1 when fh < max" do
|
||
run_time = ~U[2026-04-14 16:00:00Z]
|
||
|
||
Oban.Testing.with_testing_mode(:manual, fn ->
|
||
assert {:ok, _} = PropagationGridWorker.enqueue_next_step(run_time, 0)
|
||
|
||
[child] = all_enqueued(worker: PropagationGridWorker)
|
||
assert child.args["forecast_hour"] == 1
|
||
assert child.args["run_time"] == DateTime.to_iso8601(run_time)
|
||
end)
|
||
end
|
||
|
||
test "does not enqueue anything after the final forecast hour" do
|
||
run_time = ~U[2026-04-14 16:00:00Z]
|
||
|
||
Oban.Testing.with_testing_mode(:manual, fn ->
|
||
assert :final = PropagationGridWorker.enqueue_next_step(run_time, 18)
|
||
assert [] = all_enqueued(worker: PropagationGridWorker)
|
||
end)
|
||
end
|
||
end
|
||
end
|