From 7b78a2574ce5f22d15833a5092b71eff1f3488c2 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 09:53:05 -0500 Subject: [PATCH] fix(workers): 3 bug/perf fixes from codebase review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 66 +++++++-- .../workers/admin_task_worker.ex | 134 +++++++++++++----- .../workers/radar_frame_worker.ex | 36 ++++- .../microwaveprop/weather/grid_cache_test.exs | 51 +++++++ .../workers/admin_task_worker_test.exs | 12 +- .../workers/radar_frame_worker_test.exs | 38 +++++ 6 files changed, 274 insertions(+), 63 deletions(-) diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex index 87d1d5dd..e87ed74d 100644 --- a/lib/microwaveprop/weather/grid_cache.ex +++ b/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 diff --git a/lib/microwaveprop/workers/admin_task_worker.ex b/lib/microwaveprop/workers/admin_task_worker.ex index 40e1725e..e9cfcba2 100644 --- a/lib/microwaveprop/workers/admin_task_worker.ex +++ b/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 diff --git a/lib/microwaveprop/workers/radar_frame_worker.ex b/lib/microwaveprop/workers/radar_frame_worker.ex index c20d430c..f8c23a39 100644 --- a/lib/microwaveprop/workers/radar_frame_worker.ex +++ b/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. diff --git a/test/microwaveprop/weather/grid_cache_test.exs b/test/microwaveprop/weather/grid_cache_test.exs index 31e45fe4..d99d952e 100644 --- a/test/microwaveprop/weather/grid_cache_test.exs +++ b/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], []) diff --git a/test/microwaveprop/workers/admin_task_worker_test.exs b/test/microwaveprop/workers/admin_task_worker_test.exs index ed9fcb63..6b41c4bb 100644 --- a/test/microwaveprop/workers/admin_task_worker_test.exs +++ b/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 diff --git a/test/microwaveprop/workers/radar_frame_worker_test.exs b/test/microwaveprop/workers/radar_frame_worker_test.exs index 313bf3ad..e1719a98 100644 --- a/test/microwaveprop/workers/radar_frame_worker_test.exs +++ b/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()