Pause backfill queues during propagation grid fetch

The hrrr/weather/iemre/terrain backfill queues (20+ concurrent jobs)
were competing for bandwidth with the propagation grid HRRR download.
Now pauses those queues before the grid fetch and resumes them after,
ensuring the hourly propagation update completes reliably.
This commit is contained in:
Graham McIntire 2026-03-30 17:42:05 -05:00
parent 9c504acb67
commit 7f6905645f
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59

View file

@ -15,6 +15,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
require Logger
# Pause these queues while the grid fetch runs so they don't compete for bandwidth
@pause_queues [:hrrr, :weather, :iemre, :terrain]
@impl Oban.Worker
def perform(%Oban.Job{}) do
# HRRR takes ~45min to publish after the hour. Use 2 hours ago to ensure availability.
@ -22,45 +25,55 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
valid_time = HrrrClient.nearest_hrrr_hour(two_hours_ago)
points = Grid.conus_points()
Logger.info("PropagationGrid: fetching HRRR for #{valid_time}, #{length(points)} points")
Logger.info("PropagationGrid: pausing backfill queues, fetching HRRR for #{valid_time}, #{length(points)} points")
Enum.each(@pause_queues, &Oban.pause_queue(queue: &1))
with {:ok, grid_data} <- HrrrClient.fetch_grid(points, valid_time) do
scores =
grid_data
|> Task.async_stream(
fn {{lat, lon}, profile} ->
band_scores = Propagation.score_grid_point(profile, valid_time)
result =
with {:ok, grid_data} <- HrrrClient.fetch_grid(points, valid_time) do
scores = compute_scores(grid_data, valid_time)
Enum.map(band_scores, fn result ->
%{
lat: lat,
lon: lon,
valid_time: valid_time,
band_mhz: result.band_mhz,
score: result.score,
factors: result.factors
}
end)
end,
max_concurrency: System.schedulers_online() * 2,
timeout: 30_000
)
|> Enum.flat_map(fn
{:ok, results} -> results
{:exit, _reason} -> []
end)
Logger.info("PropagationGrid: computed #{length(scores)} scores, upserting")
Logger.info("PropagationGrid: computed #{length(scores)} scores, upserting")
case Propagation.upsert_scores(scores) do
{:ok, count} ->
Logger.info("PropagationGrid: upserted #{count} scores for #{valid_time}")
:ok
case Propagation.upsert_scores(scores) do
{:ok, count} ->
Logger.info("PropagationGrid: upserted #{count} scores for #{valid_time}")
:ok
error ->
Logger.error("PropagationGrid: upsert failed: #{inspect(error)}")
error
error ->
Logger.error("PropagationGrid: upsert failed: #{inspect(error)}")
error
end
end
end
Logger.info("PropagationGrid: resuming backfill queues")
Enum.each(@pause_queues, &Oban.resume_queue(queue: &1))
result
end
defp compute_scores(grid_data, valid_time) do
grid_data
|> Task.async_stream(
fn {{lat, lon}, profile} ->
band_scores = Propagation.score_grid_point(profile, valid_time)
Enum.map(band_scores, fn r ->
%{
lat: lat,
lon: lon,
valid_time: valid_time,
band_mhz: r.band_mhz,
score: r.score,
factors: r.factors
}
end)
end,
max_concurrency: System.schedulers_online() * 2,
timeout: 30_000
)
|> Enum.flat_map(fn
{:ok, results} -> results
{:exit, _reason} -> []
end)
end
end