perf(concurrency): partition Task.Supervisor to remove bottleneck on heavy async_stream paths

Wrap a Task.Supervisor in a PartitionSupervisor (one partition per
scheduler) and route the four sustained async_stream callers through
`{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}`.
Concurrent work — the hourly PropagationGridWorker pulling f00-f18
HRRR range downloads, the GEFS worker scoring its grid, and the
Recalibrator's 20-way positive/negative fan-out — no longer funnels
through a single supervisor PID.

Light/one-shot Task.async + Task.start sites (hrrr_client surface/
pressure fan-out, application boot warmup, weather fire-and-forget)
are left alone; partitioning only helps under sustained concurrency.

lib/microwaveprop/application.ex

@@ -25,6 +25,12 @@ defmodule Microwaveprop.Application do
       Microwaveprop.Repo,
       {Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]},
       {Phoenix.PubSub, name: Microwaveprop.PubSub},
+      # Partitioned Task.Supervisor — callers route heavy `async_stream`
+      # work through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor,
+      # self()}}` so concurrent workers (recalibrator, HRRR/GEFS range
+      # downloads, GEFS scoring) do not serialize on a single supervisor PID.
+      {PartitionSupervisor,
+       child_spec: Task.Supervisor, name: Microwaveprop.TaskSupervisor, partitions: System.schedulers_online()},
       Microwaveprop.Cache,
       Microwaveprop.Propagation.ScoreCache,
       Microwaveprop.Propagation.ScoreCacheReconciler,

lib/microwaveprop/propagation/recalibrator.ex

@@ -180,9 +180,11 @@ defmodule Microwaveprop.Propagation.Recalibrator do
     # query round-trips. Parallelize at 20 concurrent queries, well
     # under the :db pool size (30 in prod), so the recalibration
     # finishes in seconds instead of minutes.
+    supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
+
     factor_vectors =
-      contacts
-      |> Task.async_stream(&contact_to_factors(&1, factor_band),
+      supervisor
+      |> Task.Supervisor.async_stream_nolink(contacts, &contact_to_factors(&1, factor_band),
         max_concurrency: 20,
         timeout: 30_000,
         on_timeout: :kill_task
@@ -213,8 +215,11 @@ defmodule Microwaveprop.Propagation.Recalibrator do
 
     factor_band = band_mhz || 10_000
 
-    baselines
-    |> Task.async_stream(
+    supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
+
+    supervisor
+    |> Task.Supervisor.async_stream_nolink(
+      baselines,
       fn {lat, lon, time} ->
         case Weather.find_nearest_hrrr(lat, lon, time) do
           nil -> nil

lib/microwaveprop/weather/gefs_client.ex

@@ -256,9 +256,12 @@ defmodule Microwaveprop.Weather.GefsClient do
       end)
       |> Enum.reverse()
 
+    supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
+
     results =
-      merged
-      |> Task.async_stream(
+      supervisor
+      |> Task.Supervisor.async_stream_nolink(
+        merged,
         fn {start, stop} ->
           case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do
             {:ok, %{status: 206, body: body}} -> {:ok, start, body}

lib/microwaveprop/weather/hrrr_client.ex

@@ -496,12 +496,16 @@ defmodule Microwaveprop.Weather.HrrrClient do
     # when sequential. With 40+ ranges per native-duct fetch, serial
     # download dominated the per-forecast-hour wall time. 8-way
     # parallelism drops a 20s download to ~3s while still respecting
-    # the mirror's modest connection concurrency.
+    # the mirror's modest connection concurrency. Routed through the
+    # partitioned Task.Supervisor to avoid contention when the hourly
+    # PropagationGridWorker fetches f00-f18 alongside recalibrator/GEFS.
     merged = merge_ranges(ranges)
+    supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
 
     results =
-      merged
-      |> Task.async_stream(
+      supervisor
+      |> Task.Supervisor.async_stream_nolink(
+        merged,
         fn {start, stop} ->
           case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do
             {:ok, %{status: 206, body: body}} -> {:ok, start, body}
@@ -539,11 +543,15 @@ defmodule Microwaveprop.Weather.HrrrClient do
   defp download_grib_ranges_remote(url, ranges) do
     # Merge adjacent/overlapping ranges to reduce HTTP requests
     merged = merge_ranges(ranges)
+    supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
 
-    # Download ranges in parallel
+    # Download ranges in parallel, routed through the partitioned
+    # Task.Supervisor so concurrent f00-f18 fetches do not serialize on
+    # a single supervisor PID.
     results =
-      merged
-      |> Task.async_stream(
+      supervisor
+      |> Task.Supervisor.async_stream_nolink(
+        merged,
         fn {start, stop} ->
           case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do
             {:ok, %{status: 206, body: body}} -> {:ok, start, body}

lib/microwaveprop/workers/gefs_fetch_worker.ex

@@ -148,10 +148,12 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
   # `gefs_profiles` for future per-contact extended-horizon lookups.
   defp score_and_persist(profiles, run_time, fh, valid_time) do
     grid_data = profiles_to_grid(profiles)
+    supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
 
     scores =
-      grid_data
-      |> Task.async_stream(
+      supervisor
+      |> Task.Supervisor.async_stream_nolink(
+        grid_data,
         fn {{lat, lon}, profile} ->
           profile
           |> Propagation.score_grid_point(valid_time, lat, lon)

test/microwaveprop/application_test.exs

@@ -3,6 +3,28 @@ defmodule Microwaveprop.ApplicationTest do
 
   alias Microwaveprop.Application, as: App
 
+  describe "Microwaveprop.TaskSupervisor partition supervisor" do
+    test "is started and exposes schedulers_online partitions" do
+      # A partitioned Task.Supervisor removes contention on a single
+      # supervisor PID when many callers run Task.async_stream in parallel
+      # (recalibrator, hrrr_client range downloads, gefs worker, etc.).
+      %{active: active, specs: specs} = PartitionSupervisor.count_children(Microwaveprop.TaskSupervisor)
+
+      expected = System.schedulers_online()
+      assert active == expected
+      assert specs == expected
+    end
+
+    test "routes async_stream work through a partition keyed on the caller" do
+      results =
+        {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
+        |> Task.Supervisor.async_stream(1..5, fn n -> n * 2 end, max_concurrency: 2)
+        |> Enum.map(fn {:ok, v} -> v end)
+
+      assert results == [2, 4, 6, 8, 10]
+    end
+  end
+
   describe "build_timestamp/0" do
     setup do
       original_env = System.get_env("DEPLOY_TIMESTAMP")