From 7f6905645f36aa5fbb14070206d1fe03a2a997d8 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Mon, 30 Mar 2026 17:42:05 -0500
Subject: [PATCH] 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.
---
 .../workers/propagation_grid_worker.ex        | 83 +++++++++++--------
 1 file changed, 48 insertions(+), 35 deletions(-)

diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex
index e7833c17..b917c944 100644
--- a/lib/microwaveprop/workers/propagation_grid_worker.ex
+++ b/lib/microwaveprop/workers/propagation_grid_worker.ex
@@ -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