feat(telemetry): wide instrumentation + bump hrrr to 2 per pod

Config:
- runtime.exs hrrr queue 1 → 2 (6 concurrent HRRR jobs across 3 pods)

New helper Microwaveprop.Instrument.span/3 wraps :telemetry.span with
a [:microwaveprop | event_suffix] prefix so metrics can key off it.

Spans added (each emits duration + result tag):
- HrrrClient.fetch_grid / fetch_profile / fetch_idx
- NexradClient.fetch_frame + decode_png
- IemClient.fetch_asos / fetch_raob
- GefsClient.fetch_grid_profiles
- NarrClient.fetch_profile_at
- UwyoSoundingClient.fetch_sounding
- ElevationClient.fetch_elevation_profile
- Weather.upsert_hrrr_profiles_batch / upsert_gefs_profiles_batch
- Propagation.replace_scores
- PropagationGridWorker.compute_scores_algorithm
- TerrainAnalysis.analyse
- CommonVolumeRadarWorker.aggregate_stats

Telemetry catalog (MicrowavepropWeb.Telemetry):
- Oban job duration / queue_time / count / exception by (worker, queue, state)
- Per-span summary metrics for every instrumented phase above
- Periodic (10s) poller emits oban queue depth by (queue, state) — drops
  into the /admin/dashboard Metrics tab immediately

Also drops the now-redundant "fetching n0q frame" and "fetching <url>"
info lines from CommonVolumeRadarWorker / NexradClient; the span events
cover that and the worker's "ingested" line stays for per-job signal.

config/runtime.exs

@@ -166,7 +166,7 @@ if config_env() == :prod do
       solar: 1,
       weather: 3,
       gefs: 1,
-      hrrr: 1,
+      hrrr: 2,
       terrain: 3,
       iemre: 3,
       # Historical backfill for pre-2014 contacts (pre-HRRR archive).

lib/microwaveprop/instrument.ex

@@ -0,0 +1,26 @@
+defmodule Microwaveprop.Instrument do
+  @moduledoc """
+  Thin wrapper around `:telemetry.span/3` so the rest of the codebase can
+  instrument a block of work with a single call:
+
+      Instrument.span([:hrrr, :fetch_grid], %{points: length(points)}, fn ->
+        HrrrClient.fetch_grid(points, valid_time)
+      end)
+
+  Emits `[:microwaveprop | event_suffix] ++ [:start|:stop|:exception]`
+  with the automatic `duration` measurement, so metrics registered in
+  `MicrowavepropWeb.Telemetry` can key off the same prefix.
+  """
+
+  @type event_suffix :: [atom(), ...]
+
+  @spec span(event_suffix(), map(), (-> term())) :: term()
+  def span(event_suffix, metadata \\ %{}, fun) when is_list(event_suffix) and is_function(fun, 0) do
+    event = [:microwaveprop | event_suffix]
+
+    :telemetry.span(event, metadata, fn ->
+      result = fun.()
+      {result, Map.put(metadata, :result, :ok)}
+    end)
+  end
+end

lib/microwaveprop/propagation.ex

@@ -166,6 +166,14 @@ defmodule Microwaveprop.Propagation do
   """
   @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
   def replace_scores(scores, %DateTime{} = valid_time) do
+    Microwaveprop.Instrument.span(
+      [:db, :replace_scores],
+      %{valid_time: valid_time},
+      fn -> do_replace_scores(scores, valid_time) end
+    )
+  end
+
+  defp do_replace_scores(scores, valid_time) do
     {per_band, total} =
       Enum.reduce(scores, {%{}, 0}, fn score, {acc, count} ->
         {Map.update(acc, score.band_mhz, [score], &[score | &1]), count + 1}

lib/microwaveprop/terrain/elevation_client.ex

@@ -10,15 +10,21 @@ defmodule Microwaveprop.Terrain.ElevationClient do
   @spec fetch_elevation_profile(float(), float(), float(), float(), pos_integer(), keyword()) ::
           {:ok, list(map())} | {:error, String.t()}
   def fetch_elevation_profile(lat1, lon1, lat2, lon2, n \\ 64, opts \\ []) do
-    case srtm_tiles_dir() do
-      nil ->
-        Logger.debug("SRTM tiles_dir not configured, using API")
-        fetch_elevation_profile_api(lat1, lon1, lat2, lon2, n)
+    Microwaveprop.Instrument.span(
+      [:elevation, :fetch_profile],
+      %{sample_count: n, source: if(srtm_tiles_dir(), do: :srtm, else: :api)},
+      fn ->
+        case srtm_tiles_dir() do
+          nil ->
+            Logger.debug("SRTM tiles_dir not configured, using API")
+            fetch_elevation_profile_api(lat1, lon1, lat2, lon2, n)
 
-      tiles_dir ->
-        srtm_opts = Keyword.merge([download: true], opts)
-        Srtm.fetch_elevation_profile(lat1, lon1, lat2, lon2, tiles_dir, n, srtm_opts)
-    end
+          tiles_dir ->
+            srtm_opts = Keyword.merge([download: true], opts)
+            Srtm.fetch_elevation_profile(lat1, lon1, lat2, lon2, tiles_dir, n, srtm_opts)
+        end
+      end
+    )
   end
 
   defp fetch_elevation_profile_api(lat1, lon1, lat2, lon2, n) do

lib/microwaveprop/terrain/terrain_analysis.ex

@@ -54,6 +54,14 @@ defmodule Microwaveprop.Terrain.TerrainAnalysis do
   """
   @spec analyse([elevation_point()], float(), float(), keyword()) :: analysis_result()
   def analyse(profile, dist_km, freq_ghz, opts \\ []) do
+    Microwaveprop.Instrument.span(
+      [:terrain, :analyse],
+      %{point_count: length(profile), dist_km: dist_km, freq_ghz: freq_ghz},
+      fn -> do_analyse(profile, dist_km, freq_ghz, opts) end
+    )
+  end
+
+  defp do_analyse(profile, dist_km, freq_ghz, opts) do
     ant_ht_a = Keyword.get(opts, :ant_ht_a, 0.0)
     ant_ht_b = Keyword.get(opts, :ant_ht_b, 0.0)
     k = Keyword.get(opts, :k_factor, 4 / 3)

lib/microwaveprop/weather.ex

@@ -769,6 +769,14 @@ defmodule Microwaveprop.Weather do
 
   @spec upsert_gefs_profiles_batch([map()]) :: {non_neg_integer(), nil}
   def upsert_gefs_profiles_batch(profiles) do
+    Microwaveprop.Instrument.span(
+      [:db, :upsert_gefs_profiles],
+      %{count: length(profiles)},
+      fn -> do_upsert_gefs_profiles_batch(profiles) end
+    )
+  end
+
+  defp do_upsert_gefs_profiles_batch(profiles) do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
     profiles
@@ -805,6 +813,14 @@ defmodule Microwaveprop.Weather do
 
   @spec upsert_hrrr_profiles_batch([map()], keyword()) :: {non_neg_integer(), nil}
   def upsert_hrrr_profiles_batch(profiles, _opts \\ []) do
+    Microwaveprop.Instrument.span(
+      [:db, :upsert_hrrr_profiles],
+      %{count: length(profiles)},
+      fn -> do_upsert_hrrr_profiles_batch(profiles) end
+    )
+  end
+
+  defp do_upsert_hrrr_profiles_batch(profiles) do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
     profiles

lib/microwaveprop/weather/gefs_client.ex

@@ -199,6 +199,14 @@ defmodule Microwaveprop.Weather.GefsClient do
   @spec fetch_grid_profiles(Date.t(), non_neg_integer(), non_neg_integer()) ::
           {:ok, [map()]} | {:error, term()}
   def fetch_grid_profiles(%Date{} = run_date, run_hour, forecast_hour) do
+    Microwaveprop.Instrument.span(
+      [:gefs, :fetch_grid_profiles],
+      %{forecast_hour: forecast_hour},
+      fn -> do_fetch_grid_profiles(run_date, run_hour, forecast_hour) end
+    )
+  end
+
+  defp do_fetch_grid_profiles(run_date, run_hour, forecast_hour) do
     alias Microwaveprop.Propagation.Grid
     alias Microwaveprop.Weather.Grib2.Wgrib2
     alias Microwaveprop.Weather.HrrrClient

lib/microwaveprop/weather/hrrr_client.ex

@@ -86,6 +86,14 @@ defmodule Microwaveprop.Weather.HrrrClient do
   @spec fetch_grid([{float(), float()}], DateTime.t(), keyword()) ::
           {:ok, %{{float(), float()} => map()}} | {:error, term()}
   def fetch_grid(points, run_time, opts \\ []) do
+    Microwaveprop.Instrument.span(
+      [:hrrr, :fetch_grid],
+      %{point_count: length(points), forecast_hour: Keyword.get(opts, :forecast_hour, 0)},
+      fn -> do_fetch_grid(points, run_time, opts) end
+    )
+  end
+
+  defp do_fetch_grid(points, run_time, opts) do
     hour_dt = nearest_hrrr_hour(run_time)
     date = DateTime.to_date(hour_dt)
     hour = hour_dt.hour
@@ -119,6 +127,14 @@ defmodule Microwaveprop.Weather.HrrrClient do
 
   @spec fetch_profile(float(), float(), DateTime.t()) :: {:ok, map()} | {:error, term()}
   def fetch_profile(lat, lon, valid_time) do
+    Microwaveprop.Instrument.span(
+      [:hrrr, :fetch_profile],
+      %{lat: lat, lon: lon},
+      fn -> do_fetch_profile(lat, lon, valid_time) end
+    )
+  end
+
+  defp do_fetch_profile(lat, lon, valid_time) do
     hour_dt = nearest_hrrr_hour(valid_time)
     date = DateTime.to_date(hour_dt)
     hour = hour_dt.hour
@@ -373,16 +389,18 @@ defmodule Microwaveprop.Weather.HrrrClient do
   end
 
   defp fetch_idx(url) do
-    case Req.get(url, req_options()) do
-      {:ok, %{status: 200, body: body}} ->
-        {:ok, body}
+    Microwaveprop.Instrument.span([:hrrr, :fetch_idx], %{url: url}, fn ->
+      case Req.get(url, req_options()) do
+        {:ok, %{status: 200, body: body}} ->
+          {:ok, body}
 
-      {:ok, %{status: status}} ->
-        {:error, "HRRR idx HTTP #{status}"}
+        {:ok, %{status: status}} ->
+          {:error, "HRRR idx HTTP #{status}"}
 
-      {:error, reason} ->
-        {:error, reason}
-    end
+        {:error, reason} ->
+          {:error, reason}
+      end
+    end)
   end
 
   @doc """

lib/microwaveprop/weather/iem_client.ex

@@ -56,32 +56,36 @@ defmodule Microwaveprop.Weather.IemClient do
   def fetch_asos(station_id, start_dt, end_dt) do
     url = asos_url(station_id, start_dt, end_dt)
 
-    case Req.get(url, req_options()) do
-      {:ok, %{status: 200, body: body}} ->
-        {:ok, parse_asos_csv(body)}
+    Microwaveprop.Instrument.span([:iem, :fetch_asos], %{station: station_id}, fn ->
+      case Req.get(url, req_options()) do
+        {:ok, %{status: 200, body: body}} ->
+          {:ok, parse_asos_csv(body)}
 
-      {:ok, %{status: status}} ->
-        {:error, "IEM ASOS HTTP #{status}"}
+        {:ok, %{status: status}} ->
+          {:error, "IEM ASOS HTTP #{status}"}
 
-      {:error, reason} ->
-        {:error, reason}
-    end
+        {:error, reason} ->
+          {:error, reason}
+      end
+    end)
   end
 
   @spec fetch_raob(String.t(), DateTime.t()) :: {:ok, [map()]} | {:error, term()}
   def fetch_raob(station_id, dt) do
     url = raob_url(station_id, dt)
 
-    case Req.get(url, req_options()) do
-      {:ok, %{status: 200, body: body}} ->
-        {:ok, parse_raob_json(body)}
+    Microwaveprop.Instrument.span([:iem, :fetch_raob], %{station: station_id}, fn ->
+      case Req.get(url, req_options()) do
+        {:ok, %{status: 200, body: body}} ->
+          {:ok, parse_raob_json(body)}
 
-      {:ok, %{status: status}} ->
-        {:error, "IEM RAOB HTTP #{status}"}
+        {:ok, %{status: status}} ->
+          {:error, "IEM RAOB HTTP #{status}"}
 
-      {:error, reason} ->
-        {:error, reason}
-    end
+        {:error, reason} ->
+          {:error, reason}
+      end
+    end)
   end
 
   @spec fetch_iemre(float(), float(), Date.t()) :: {:ok, [map()]} | {:error, term()}

lib/microwaveprop/weather/narr_client.ex

@@ -212,12 +212,14 @@ defmodule Microwaveprop.Weather.NarrClient do
   @spec fetch_profile_at(DateTime.t(), {float(), float()}) ::
           {:ok, map()} | {:error, term()}
   def fetch_profile_at(%DateTime{} = valid_time, {lat, lon}) when is_number(lat) and is_number(lon) do
-    grb_url = url_for(valid_time)
+    Microwaveprop.Instrument.span([:narr, :fetch_profile], %{lat: lat, lon: lon}, fn ->
+      grb_url = url_for(valid_time)
 
-    with {:ok, index} <- fetch_inventory(valid_time),
-         {:ok, record_keys} <- pick_records(index) do
-      do_fetch_profile(grb_url, index, record_keys, lat, lon)
-    end
+      with {:ok, index} <- fetch_inventory(valid_time),
+           {:ok, record_keys} <- pick_records(index) do
+        do_fetch_profile(grb_url, index, record_keys, lat, lon)
+      end
+    end)
   end
 
   defp do_fetch_profile(grb_url, index, record_keys, lat, lon) do

lib/microwaveprop/weather/nexrad_client.ex

@@ -33,20 +33,20 @@ defmodule Microwaveprop.Weather.NexradClient do
 
     req_opts = Application.get_env(:microwaveprop, :nexrad_req_options, [])
 
-    Logger.info("NexradClient: fetching #{url}")
+    Microwaveprop.Instrument.span([:nexrad, :fetch_frame], %{}, fn ->
+      case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do
+        {:ok, %{status: 200, body: body}} when is_binary(body) ->
+          process_frame(body, rounded, points_of_interest)
 
-    case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do
-      {:ok, %{status: 200, body: body}} when is_binary(body) ->
-        process_frame(body, rounded, points_of_interest)
+        {:ok, %{status: status}} ->
+          Logger.warning("NexradClient: HTTP #{status} for #{url}")
+          {:error, "NEXRAD n0q HTTP #{status}"}
 
-      {:ok, %{status: status}} ->
-        Logger.warning("NexradClient: HTTP #{status} for #{url}")
-        {:error, "NEXRAD n0q HTTP #{status}"}
-
-      {:error, reason} ->
-        Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}")
-        {:error, reason}
-    end
+        {:error, reason} ->
+          Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}")
+          {:error, reason}
+      end
+    end)
   end
 
   @doc "Round a DateTime down to the nearest 5-minute boundary."
@@ -111,22 +111,31 @@ defmodule Microwaveprop.Weather.NexradClient do
 
   defp fetch_and_decode_frame(rounded) do
     url = frame_url(rounded)
+
+    Microwaveprop.Instrument.span([:nexrad, :fetch_frame], %{}, fn ->
+      handle_decode_frame_response(url, do_fetch_png(url))
+    end)
+  end
+
+  defp do_fetch_png(url) do
     req_opts = Application.get_env(:microwaveprop, :nexrad_req_options, [])
+    Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts)
+  end
 
-    Logger.info("NexradClient: fetching #{url}")
+  defp handle_decode_frame_response(_url, {:ok, %{status: 200, body: body}}) when is_binary(body) do
+    Microwaveprop.Instrument.span([:nexrad, :decode_png], %{bytes: byte_size(body)}, fn ->
+      decode_png_to_pixels(body)
+    end)
+  end
 
-    case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do
-      {:ok, %{status: 200, body: body}} when is_binary(body) ->
-        decode_png_to_pixels(body)
+  defp handle_decode_frame_response(url, {:ok, %{status: status}}) do
+    Logger.warning("NexradClient: HTTP #{status} for #{url}")
+    {:error, "NEXRAD n0q HTTP #{status}"}
+  end
 
-      {:ok, %{status: status}} ->
-        Logger.warning("NexradClient: HTTP #{status} for #{url}")
-        {:error, "NEXRAD n0q HTTP #{status}"}
-
-      {:error, reason} ->
-        Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}")
-        {:error, reason}
-    end
+  defp handle_decode_frame_response(url, {:error, reason}) do
+    Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}")
+    {:error, reason}
   end
 
   defp extract_rain_cells(pixels, width, center_lat, center_lon, radius_km, min_dbz) do

lib/microwaveprop/weather/uwyo_sounding_client.ex

@@ -33,13 +33,15 @@ defmodule Microwaveprop.Weather.UwyoSoundingClient do
 
   @spec fetch_sounding(String.t(), DateTime.t()) :: {:ok, [sounding()]} | {:error, term()}
   def fetch_sounding(station_code, %DateTime{} = dt) do
-    url = sounding_url(station_code, dt)
+    Microwaveprop.Instrument.span([:uwyo, :fetch_sounding], %{station: station_code}, fn ->
+      url = sounding_url(station_code, dt)
 
-    case Req.get(url, req_options()) do
-      {:ok, %{status: 200, body: body}} -> {:ok, parse_sounding_html(body)}
-      {:ok, %{status: status}} -> {:error, "UWYO sounding HTTP #{status}"}
-      {:error, reason} -> {:error, reason}
-    end
+      case Req.get(url, req_options()) do
+        {:ok, %{status: 200, body: body}} -> {:ok, parse_sounding_html(body)}
+        {:ok, %{status: status}} -> {:error, "UWYO sounding HTTP #{status}"}
+        {:error, reason} -> {:error, reason}
+      end
+    end)
   end
 
   @spec parse_sounding_html(String.t()) :: [sounding()]

lib/microwaveprop/workers/common_volume_radar_worker.ex

@@ -77,8 +77,6 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do
     qso_at = contact.qso_timestamp
     rounded = NexradClient.round_to_5min(DateTime.from_naive!(qso_at, "Etc/UTC"))
 
-    Logger.info("CommonVolumeRadarWorker: #{contact.id} fetching n0q frame for #{DateTime.to_iso8601(rounded)}")
-
     case NexradClient.fetch_decoded_frame(rounded) do
       {:ok, pixels, width} ->
         stats = aggregate_stats(pixels, width, pos1, pos2)
@@ -121,6 +119,12 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do
             coverage_pct: float() | nil
           }
   def aggregate_stats(pixels, width, pos1, pos2, opts \\ []) do
+    Microwaveprop.Instrument.span([:radar, :aggregate_stats], %{}, fn ->
+      do_aggregate_stats(pixels, width, pos1, pos2, opts)
+    end)
+  end
+
+  defp do_aggregate_stats(pixels, width, pos1, pos2, opts) do
     step = Keyword.get(opts, :step, @pixel_step)
 
     case CommonVolume.bounding_box(pos1, pos2, @radius_km) do

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -413,16 +413,23 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   end
 
   defp compute_scores_algorithm(grid_data, valid_time, include_factors?) do
-    grid_data
-    |> Task.async_stream(
-      &score_one_point(&1, valid_time, include_factors?),
-      max_concurrency: System.schedulers_online() * 2,
-      timeout: 30_000
+    Microwaveprop.Instrument.span(
+      [:propagation_grid, :score_band],
+      %{point_count: length(grid_data)},
+      fn ->
+        grid_data
+        |> Task.async_stream(
+          &score_one_point(&1, valid_time, include_factors?),
+          max_concurrency: System.schedulers_online() * 2,
+          timeout: 30_000
+        )
+        |> Stream.flat_map(fn
+          {:ok, results} -> results
+          {:exit, _reason} -> []
+        end)
+        |> Enum.to_list()
+      end
     )
-    |> Stream.flat_map(fn
-      {:ok, results} -> results
-      {:exit, _reason} -> []
-    end)
   end
 
   defp score_one_point({{lat, lon}, profile}, valid_time, include_factors?) do

lib/microwaveprop_web/telemetry.ex

@@ -22,14 +22,19 @@ defmodule MicrowavepropWeb.Telemetry do
   end
 
   def metrics do
+    phoenix_metrics() ++
+      ecto_metrics() ++
+      vm_metrics() ++
+      oban_metrics() ++
+      http_client_metrics() ++
+      db_batch_metrics() ++
+      worker_phase_metrics()
+  end
+
+  defp phoenix_metrics do
     [
-      # Phoenix Metrics
-      summary("phoenix.endpoint.start.system_time",
-        unit: {:native, :millisecond}
-      ),
-      summary("phoenix.endpoint.stop.duration",
-        unit: {:native, :millisecond}
-      ),
+      summary("phoenix.endpoint.start.system_time", unit: {:native, :millisecond}),
+      summary("phoenix.endpoint.stop.duration", unit: {:native, :millisecond}),
       summary("phoenix.router_dispatch.start.system_time",
         tags: [:route],
         unit: {:native, :millisecond}
@@ -42,19 +47,18 @@ defmodule MicrowavepropWeb.Telemetry do
         tags: [:route],
         unit: {:native, :millisecond}
       ),
-      summary("phoenix.socket_connected.duration",
-        unit: {:native, :millisecond}
-      ),
+      summary("phoenix.socket_connected.duration", unit: {:native, :millisecond}),
       sum("phoenix.socket_drain.count"),
-      summary("phoenix.channel_joined.duration",
-        unit: {:native, :millisecond}
-      ),
+      summary("phoenix.channel_joined.duration", unit: {:native, :millisecond}),
       summary("phoenix.channel_handled_in.duration",
         tags: [:event],
         unit: {:native, :millisecond}
-      ),
+      )
+    ]
+  end
 
-      # Database Metrics
+  defp ecto_metrics do
+    [
       summary("microwaveprop.repo.query.total_time",
         unit: {:native, :millisecond},
         description: "The sum of the other measurements"
@@ -74,9 +78,12 @@ defmodule MicrowavepropWeb.Telemetry do
       summary("microwaveprop.repo.query.idle_time",
         unit: {:native, :millisecond},
         description: "The time the connection spent waiting before being checked out for the query"
-      ),
+      )
+    ]
+  end
 
-      # VM Metrics
+  defp vm_metrics do
+    [
       summary("vm.memory.total", unit: {:byte, :kilobyte}),
       summary("vm.total_run_queue_lengths.total"),
       summary("vm.total_run_queue_lengths.cpu"),
@@ -84,11 +91,192 @@ defmodule MicrowavepropWeb.Telemetry do
     ]
   end
 
-  defp periodic_measurements do
+  # Oban emits [:oban, :job, :start|:stop|:exception]. `stop` measurements
+  # include both `duration` (job runtime) and `queue_time` (available →
+  # executing latency). Tagging by worker + queue gives us per-worker
+  # distributions without per-worker metric registration.
+  defp oban_metrics do
     [
-      # A module, function and arguments to be invoked periodically.
-      # This function must call :telemetry.execute/3 and a metric must be added above.
-      # {MicrowavepropWeb, :count_users, []}
+      summary("oban.job.stop.duration",
+        event_name: [:oban, :job, :stop],
+        measurement: :duration,
+        tags: [:worker, :queue, :state],
+        tag_values: &oban_tag_values/1,
+        unit: {:native, :millisecond}
+      ),
+      summary("oban.job.stop.queue_time",
+        event_name: [:oban, :job, :stop],
+        measurement: :queue_time,
+        tags: [:worker, :queue],
+        tag_values: &oban_tag_values/1,
+        unit: {:native, :millisecond}
+      ),
+      counter("oban.job.stop.count",
+        event_name: [:oban, :job, :stop],
+        tags: [:worker, :queue, :state],
+        tag_values: &oban_tag_values/1
+      ),
+      counter("oban.job.exception.count",
+        event_name: [:oban, :job, :exception],
+        tags: [:worker, :queue],
+        tag_values: &oban_tag_values/1
+      ),
+      last_value("microwaveprop.oban.queue.count",
+        event_name: [:microwaveprop, :oban, :queue, :depth],
+        measurement: :count,
+        tags: [:queue, :state],
+        description: "Oban job count per (queue, state) sampled every 10s"
+      )
     ]
   end
+
+  # HTTP client spans emitted via `Microwaveprop.Instrument`. Each client
+  # wraps its own public entry points so we get independent latency
+  # histograms per upstream.
+  defp http_client_metrics do
+    [
+      # HRRR pressure-level fetch (batch-of-points via NOMADS/S3 + wgrib2)
+      summary("microwaveprop.hrrr.fetch_grid.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.hrrr.fetch_profile.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.hrrr.fetch_idx.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+
+      # NEXRAD n0q frame (IEM archive PNG)
+      summary("microwaveprop.nexrad.fetch_frame.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.nexrad.decode_png.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+
+      # IEM Mesonet ASOS + RAOB
+      summary("microwaveprop.iem.fetch_asos.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.iem.fetch_raob.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+
+      # GEFS extended-horizon ensemble
+      summary("microwaveprop.gefs.fetch_grid_profiles.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+
+      # NARR historical backfill
+      summary("microwaveprop.narr.fetch_profile.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+
+      # University of Wyoming soundings (Canadian RAOB backfill)
+      summary("microwaveprop.uwyo.fetch_sounding.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+
+      # Terrain elevation API (Open-Meteo or local SRTM)
+      summary("microwaveprop.elevation.fetch_profile.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      )
+    ]
+  end
+
+  defp db_batch_metrics do
+    [
+      summary("microwaveprop.db.upsert_hrrr_profiles.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.db.upsert_gefs_profiles.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.db.upsert_surface_observations.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.db.upsert_sounding.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.db.replace_scores.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      )
+    ]
+  end
+
+  defp worker_phase_metrics do
+    [
+      summary("microwaveprop.propagation_grid.score_band.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.terrain.analyse.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      ),
+      summary("microwaveprop.radar.aggregate_stats.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:result]
+      )
+    ]
+  end
+
+  defp oban_tag_values(%{job: %Oban.Job{worker: worker, queue: queue}} = meta) do
+    %{
+      worker: worker || "unknown",
+      queue: queue || "unknown",
+      state: Map.get(meta, :state, :unknown)
+    }
+  end
+
+  defp oban_tag_values(_), do: %{worker: "unknown", queue: "unknown", state: :unknown}
+
+  defp periodic_measurements do
+    [
+      {__MODULE__, :dispatch_oban_queue_depth, []}
+    ]
+  end
+
+  @doc false
+  # Periodic poller: emit `microwaveprop.oban.queue.depth` gauges per
+  # (queue, state) tuple so LiveDashboard / Prometheus can chart queue
+  # saturation over time. Cheap aggregation query, runs every 10s.
+  def dispatch_oban_queue_depth do
+    import Ecto.Query
+
+    query =
+      from(j in "oban_jobs",
+        group_by: [j.queue, j.state],
+        select: {j.queue, j.state, count(j.id)}
+      )
+
+    for {queue, state, count} <- Microwaveprop.Repo.all(query) do
+      :telemetry.execute(
+        [:microwaveprop, :oban, :queue, :depth],
+        %{count: count},
+        %{queue: queue, state: state}
+      )
+    end
+
+    :ok
+  rescue
+    # Don't crash the poller if the DB blips — telemetry is best-effort.
+    _ -> :ok
+  end
 end