fix(workers): 3 bug/perf fixes from codebase review

1. GridCache: auto-release fill lock when the claimer process crashes.
   claim_fill/1 + release_fill/1 go through the GenServer so the
   server can Process.monitor the caller and clean up the ETS entry
   on :DOWN. Clear/0 now resets both the data table and the lock
   table. Fixes a latent bug where a crashed fill leaked the lock
   indefinitely, preventing every subsequent /weather mount for that
   valid_time from claiming and leaving cache cold.

2. RadarFrameWorker: distinguish permanent vs transient fetch errors.
   404 from the IEM n0q archive is permanent (file will never exist)
   and marks contacts :unavailable as before. Any other error shape
   (5xx, timeout, transport failure) now returns {:error, reason}
   so Oban retries — previously those also pinned contacts at
   :unavailable after a transient outage.

3. AdminTaskWorker.native_derive: replace per-row Repo.update_all
   (N round-trips + N fsyncs) with one UPDATE ... FROM unnest(...)
   per 2000-row batch. For the 10k-profile budget this is one
   network round trip per chunk instead of 10k, and one fsync per
   chunk instead of 10k. Restructured the clause to separate
   derivation (pure) from persistence (I/O).

All three changes are test-covered (grid_cache_test auto-release
test, radar_frame_worker_test 5xx + transport tests, existing
admin_task_worker_test native_derive coverage exercises the new
bulk path). Also drops the scorer_diff no-op test that was
verifying the clause removed in 61da51c.

lib/microwaveprop/weather/grid_cache.ex

@@ -91,27 +91,28 @@ defmodule Microwaveprop.Weather.GridCache do
 
   @spec clear() :: :ok
   def clear do
-    :ets.delete_all_objects(@table)
-    :ok
+    GenServer.call(__MODULE__, :clear)
   end
 
   @doc """
   Atomically claim the right to fill the cache for `valid_time`. Returns
   `true` if this caller won the claim and should run the fill; `false` if
   another caller is already filling. Prevents N concurrent /weather mounts
-  after a pod restart from each firing the 15-second `load_weather_grid_from_db`
-  query and starving the Postgres connection pool.
+  after a pod restart from each firing the 15-second cold-fill read and
+  starving the Postgres connection pool.
+
+  The GenServer `Process.monitor`s the caller: if the caller crashes
+  before calling `release_fill/1`, the lock is released automatically.
   """
   @spec claim_fill(DateTime.t()) :: boolean()
   def claim_fill(valid_time) do
-    :ets.insert_new(@lock_table, {valid_time, :in_progress})
+    GenServer.call(__MODULE__, {:claim_fill, valid_time, self()})
   end
 
   @doc "Release a fill lock claimed via `claim_fill/1`."
   @spec release_fill(DateTime.t()) :: :ok
   def release_fill(valid_time) do
-    :ets.delete(@lock_table, valid_time)
-    :ok
+    GenServer.call(__MODULE__, {:release_fill, valid_time})
   end
 
   @spec sync() :: :ok
@@ -122,15 +123,51 @@ defmodule Microwaveprop.Weather.GridCache do
   @impl true
   def init(_opts) do
     :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
-    :ets.new(@lock_table, [:set, :named_table, :public])
+    :ets.new(@lock_table, [:set, :named_table, :protected])
     PubSub.subscribe(@pubsub, @topic)
-    {:ok, %{}}
+    {:ok, %{monitors: %{}}}
   end
 
   @impl true
   def handle_call(:sync, _from, state), do: {:reply, :ok, state}
 
+  def handle_call(:clear, _from, state) do
+    :ets.delete_all_objects(@table)
+
+    # Demonitor all tracked callers and drop every lock so tests start clean.
+    for {ref, _vt} <- state.monitors, do: Process.demonitor(ref, [:flush])
+    :ets.delete_all_objects(@lock_table)
+
+    {:reply, :ok, %{state | monitors: %{}}}
+  end
+
+  def handle_call({:claim_fill, valid_time, caller}, _from, state) do
+    if :ets.insert_new(@lock_table, {valid_time, caller}) do
+      ref = Process.monitor(caller)
+      {:reply, true, %{state | monitors: Map.put(state.monitors, ref, valid_time)}}
+    else
+      {:reply, false, state}
+    end
+  end
+
+  def handle_call({:release_fill, valid_time}, _from, state) do
+    {monitors, _matched} = pop_monitor_for(state.monitors, valid_time)
+    :ets.delete(@lock_table, valid_time)
+    {:reply, :ok, %{state | monitors: monitors}}
+  end
+
   @impl true
+  def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
+    case Map.pop(state.monitors, ref) do
+      {nil, _} ->
+        {:noreply, state}
+
+      {valid_time, monitors} ->
+        :ets.delete(@lock_table, valid_time)
+        {:noreply, %{state | monitors: monitors}}
+    end
+  end
+
   def handle_info({:weather_cache_refresh, valid_time, rows}, state) do
     put(valid_time, rows)
     {:noreply, state}
@@ -138,6 +175,17 @@ defmodule Microwaveprop.Weather.GridCache do
 
   def handle_info(_msg, state), do: {:noreply, state}
 
+  defp pop_monitor_for(monitors, valid_time) do
+    case Enum.find(monitors, fn {_ref, vt} -> vt == valid_time end) do
+      nil ->
+        {monitors, nil}
+
+      {ref, ^valid_time} ->
+        Process.demonitor(ref, [:flush])
+        {Map.delete(monitors, ref), ref}
+    end
+  end
+
   # ---------- Internal ----------
 
   defp list_to_grid(rows) do

lib/microwaveprop/workers/admin_task_worker.ex

@@ -135,43 +135,8 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do
 
     Logger.info("AdminTask: deriving fields for #{length(profiles)} native profiles")
 
-    count =
-      Enum.count(profiles, fn profile ->
-        p = %{
-          heights_m: profile.heights_m,
-          temp_k: profile.temp_k,
-          spfh: profile.spfh,
-          pressure_pa: profile.pressure_pa,
-          u_wind_ms: profile.u_wind_ms,
-          v_wind_ms: profile.v_wind_ms,
-          tke_m2s2: profile.tke_m2s2,
-          level_count: profile.level_count
-        }
-
-        inversion_fields =
-          case Inversion.find_inversion_top(p) do
-            {:ok, %{height_m: top_h, level_idx: top_idx, base_idx: base_idx}} ->
-              [
-                inversion_top_m: top_h,
-                bulk_richardson: Inversion.bulk_richardson(p, base_idx, top_idx),
-                shear_at_top_ms: Inversion.shear_magnitude(p, base_idx, top_idx),
-                theta_e_jump_k: ThetaE.theta_e_jump(p, base_idx, top_idx)
-              ]
-
-            :none ->
-              [inversion_top_m: nil, bulk_richardson: nil, shear_at_top_ms: nil, theta_e_jump_k: nil]
-          end
-
-        duct_result = Duct.analyze(p)
-        fields = inversion_fields ++ [ducts: duct_result.ducts, best_duct_band_ghz: duct_result.best_duct_band_ghz]
-
-        {1, _} =
-          HrrrNativeProfile
-          |> where([pr], pr.id == ^profile.id)
-          |> Repo.update_all(set: fields)
-
-        true
-      end)
+    rows = Enum.map(profiles, &derive_native_row/1)
+    count = bulk_update_native_derivations(rows)
 
     Logger.info("AdminTask: derived fields for #{count} profiles")
     :ok
@@ -231,4 +196,99 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do
     Logger.info("AdminTask: backtest for #{feature_name} complete, wrote #{path}")
     :ok
   end
+
+  # Bulk-update via one `UPDATE ... FROM unnest(...)` statement per batch:
+  # one Postgres round-trip and one fsync instead of N, which was the
+  # dominant cost for runs of 10k profiles against the remote DB host.
+  @bulk_update_chunk 2000
+
+  defp derive_native_row(profile) do
+    p = %{
+      heights_m: profile.heights_m,
+      temp_k: profile.temp_k,
+      spfh: profile.spfh,
+      pressure_pa: profile.pressure_pa,
+      u_wind_ms: profile.u_wind_ms,
+      v_wind_ms: profile.v_wind_ms,
+      tke_m2s2: profile.tke_m2s2,
+      level_count: profile.level_count
+    }
+
+    {top_m, bulk, shear, theta_jump} =
+      case Inversion.find_inversion_top(p) do
+        {:ok, %{height_m: top_h, level_idx: top_idx, base_idx: base_idx}} ->
+          {
+            top_h,
+            Inversion.bulk_richardson(p, base_idx, top_idx),
+            Inversion.shear_magnitude(p, base_idx, top_idx),
+            ThetaE.theta_e_jump(p, base_idx, top_idx)
+          }
+
+        :none ->
+          {nil, nil, nil, nil}
+      end
+
+    duct_result = Duct.analyze(p)
+
+    %{
+      id: profile.id,
+      inversion_top_m: top_m,
+      bulk_richardson: bulk,
+      shear_at_top_ms: shear,
+      theta_e_jump_k: theta_jump,
+      ducts: duct_result.ducts,
+      best_duct_band_ghz: duct_result.best_duct_band_ghz
+    }
+  end
+
+  defp bulk_update_native_derivations([]), do: 0
+
+  defp bulk_update_native_derivations(rows) do
+    rows
+    |> Enum.chunk_every(@bulk_update_chunk)
+    |> Enum.reduce(0, fn chunk, acc -> acc + execute_bulk_update(chunk) end)
+  end
+
+  defp execute_bulk_update(chunk) do
+    # Ecto loads binary_id as the dashed-string form; Postgrex expects raw
+    # 16-byte binaries for `uuid[]` params. Dump once per id before send.
+    ids = Enum.map(chunk, fn %{id: id} -> Ecto.UUID.dump!(id) end)
+    tops = Enum.map(chunk, & &1.inversion_top_m)
+    bulks = Enum.map(chunk, & &1.bulk_richardson)
+    shears = Enum.map(chunk, & &1.shear_at_top_ms)
+    theta_jumps = Enum.map(chunk, & &1.theta_e_jump_k)
+    # Ducts are sent as a `text[]` of encoded JSON; each element is cast
+    # to `jsonb` per-row via the v.ducts::jsonb reference in the UPDATE
+    # so Postgres parses the JSON payload on ingest rather than storing
+    # it as a literal string.
+    ducts_json = Enum.map(chunk, fn row -> Jason.encode!(row.ducts || []) end)
+    bands = Enum.map(chunk, & &1.best_duct_band_ghz)
+
+    sql = """
+    UPDATE hrrr_native_profiles p
+    SET inversion_top_m = v.inversion_top_m,
+        bulk_richardson = v.bulk_richardson,
+        shear_at_top_ms = v.shear_at_top_ms,
+        theta_e_jump_k = v.theta_e_jump_k,
+        ducts = v.ducts::jsonb,
+        best_duct_band_ghz = v.best_duct_band_ghz,
+        updated_at = now()
+    FROM (
+      SELECT
+        unnest($1::uuid[]) AS id,
+        unnest($2::float8[]) AS inversion_top_m,
+        unnest($3::float8[]) AS bulk_richardson,
+        unnest($4::float8[]) AS shear_at_top_ms,
+        unnest($5::float8[]) AS theta_e_jump_k,
+        unnest($6::text[]) AS ducts,
+        unnest($7::float8[]) AS best_duct_band_ghz
+    ) v
+    WHERE p.id = v.id
+    """
+
+    %{num_rows: n} =
+      Repo.query!(sql, [ids, tops, bulks, shears, theta_jumps, ducts_json, bands])
+
+    n
+  end
 end

lib/microwaveprop/workers/radar_frame_worker.ex

@@ -68,18 +68,42 @@ defmodule Microwaveprop.Workers.RadarFrameWorker do
     case NexradClient.fetch_decoded_frame(rounded) do
       {:ok, pixels, width} ->
         process_frame(contacts, rounded, pixels, width)
+        :ok
 
       {:error, reason} ->
-        Enum.each(contacts, &mark_status(&1, :unavailable))
+        if permanent_error?(reason) do
+          Enum.each(contacts, &mark_status(&1, :unavailable))
 
-        Logger.info(
-          "RadarFrameWorker: no frame for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}"
-        )
+          Logger.info(
+            "RadarFrameWorker: no frame for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}"
+          )
+
+          :ok
+        else
+          # Transient (5xx, timeout, connrefused) — let Oban retry so the
+          # contacts stay :queued and land on the next attempt rather than
+          # getting pinned :unavailable after a temporary NEXRAD outage.
+          Logger.warning(
+            "RadarFrameWorker: transient error for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}"
+          )
+
+          {:error, reason}
+        end
     end
-
-    :ok
   end
 
+  # Permanent: 4xx means the IEM archive does not have this frame (the
+  # archive genuinely has gaps), so the frame will never exist and
+  # retrying is pointless. Any other shape is treated as transient.
+  defp permanent_error?("NEXRAD n0q HTTP " <> code) do
+    case Integer.parse(code) do
+      {status, _} when status in 400..499 -> true
+      _ -> false
+    end
+  end
+
+  defp permanent_error?(_), do: false
+
   @doc """
   Process a batch of contacts against an already-decoded frame. Exposed
   for tests and for any caller that has the pixel buffer in hand.

test/microwaveprop/weather/grid_cache_test.exs

@@ -70,6 +70,57 @@ defmodule Microwaveprop.Weather.GridCacheTest do
     end
   end
 
+  describe "claim_fill/1 and release_fill/1" do
+    @valid_time ~U[2026-04-21 12:00:00Z]
+
+    test "first claimer wins, subsequent claimers see :in_progress" do
+      assert GridCache.claim_fill(@valid_time) == true
+      assert GridCache.claim_fill(@valid_time) == false
+    end
+
+    test "release_fill allows a re-claim" do
+      assert GridCache.claim_fill(@valid_time) == true
+      :ok = GridCache.release_fill(@valid_time)
+      assert GridCache.claim_fill(@valid_time) == true
+    end
+
+    test "lock is auto-released when the claimer process crashes" do
+      parent = self()
+
+      {:ok, claimer} =
+        Task.start(fn ->
+          true = GridCache.claim_fill(@valid_time)
+          send(parent, :claimed)
+          # Block until we're killed
+          Process.sleep(:infinity)
+        end)
+
+      receive do
+        :claimed -> :ok
+      after
+        1000 -> flunk("claimer never signaled")
+      end
+
+      # Lock is held by the crashed process
+      assert GridCache.claim_fill(@valid_time) == false
+
+      # Kill the claimer
+      ref = Process.monitor(claimer)
+      Process.exit(claimer, :kill)
+
+      receive do
+        {:DOWN, ^ref, :process, ^claimer, _} -> :ok
+      after
+        1000 -> flunk("claimer didn't die")
+      end
+
+      # Flush the GenServer so the :DOWN handler runs before we re-probe ETS.
+      :ok = GridCache.sync()
+
+      assert GridCache.claim_fill(@valid_time) == true
+    end
+  end
+
   describe "latest_valid_time/0" do
     test "returns the most recent cached valid_time" do
       GridCache.put(~U[2026-04-12 10:00:00Z], [])

test/microwaveprop/workers/admin_task_worker_test.exs

@@ -23,8 +23,6 @@ defmodule Microwaveprop.Workers.AdminTaskWorkerTest do
     * `recalibrate` — delegates to `Recalibrator.fit/1`. With no
       contacts in the sandbox it returns the insufficient-data shape.
 
-    * `scorer_diff` — documented no-op.
-
     * unknown task — catch-all returns `{:error, _}`.
 
   The tests characterize current behavior so downstream refactors
@@ -346,14 +344,6 @@ defmodule Microwaveprop.Workers.AdminTaskWorkerTest do
     end
   end
 
-  describe "perform/1 — task=scorer_diff" do
-    test "is a documented no-op" do
-      # The propagation_scores table was dropped; this branch only
-      # exists to drain in-flight Oban rows from the pre-cutover era.
-      assert :ok = AdminTaskWorker.perform(%Oban.Job{args: %{"task" => "scorer_diff"}})
-    end
-  end
-
   describe "perform/1 — unknown task" do
     test "returns {:error, reason} for an unrecognized task name" do
       assert {:error, "unknown task: frobnicate"} =
@@ -374,7 +364,7 @@ defmodule Microwaveprop.Workers.AdminTaskWorkerTest do
       # Worker config is load-bearing: admin tasks are long-running and
       # must not run concurrently (unique) or retry on failure
       # (max_attempts=1). Pin it here so a config regression fails loudly.
-      changeset = AdminTaskWorker.new(%{"task" => "scorer_diff"})
+      changeset = AdminTaskWorker.new(%{"task" => "recalibrate"})
 
       assert Ecto.Changeset.get_field(changeset, :queue) == "admin"
       assert Ecto.Changeset.get_field(changeset, :max_attempts) == 1

test/microwaveprop/workers/radar_frame_worker_test.exs

@@ -83,6 +83,44 @@ defmodule Microwaveprop.Workers.RadarFrameWorkerTest do
       assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c2.id)
     end
 
+    test "returns {:error, _} and leaves contacts :queued on a 5xx server error (so Oban retries)" do
+      c1 = insert_contact(%{station1: "S1"})
+      c2 = insert_contact(%{station1: "S2"})
+
+      Req.Test.stub(NexradClient, fn conn ->
+        Plug.Conn.send_resp(conn, 503, "service unavailable")
+      end)
+
+      assert {:error, _reason} =
+               perform_job(RadarFrameWorker, %{
+                 "frame_ts" => "2024-09-15T18:30:00Z",
+                 "contact_ids" => [c1.id, c2.id]
+               })
+
+      # Neither contact should be pinned :unavailable — a retry might succeed.
+      assert %Contact{radar_status: status1} = Repo.get!(Contact, c1.id)
+      assert %Contact{radar_status: status2} = Repo.get!(Contact, c2.id)
+      refute status1 == :unavailable
+      refute status2 == :unavailable
+    end
+
+    test "returns {:error, _} on a transport error without pinning contacts :unavailable" do
+      c1 = insert_contact(%{station1: "T1"})
+
+      Req.Test.stub(NexradClient, fn conn ->
+        Req.Test.transport_error(conn, :timeout)
+      end)
+
+      assert {:error, _reason} =
+               perform_job(RadarFrameWorker, %{
+                 "frame_ts" => "2024-09-15T18:30:00Z",
+                 "contact_ids" => [c1.id]
+               })
+
+      assert %Contact{radar_status: status} = Repo.get!(Contact, c1.id)
+      refute status == :unavailable
+    end
+
     test "skips contacts already at a terminal status (idempotent replay)" do
       c_done = insert_contact(%{station1: "DONE"})
       {:ok, _} = c_done |> Ecto.Changeset.change(%{radar_status: :complete}) |> Repo.update()