From bfddb898bfaf82c329efcc4a8e5eb59145f66722 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Wed, 22 Apr 2026 13:53:57 -0500 Subject: [PATCH] fix(backfill): stop clobbering :complete enrichment statuses Three bugs were letting mechanism/terrain/radar contacts ping-pong between :complete and :queued on every backfill cron: 1. enqueue_for_contact/2 now filters out types that are already :complete on the contact before building jobs. Previously mark_status!/3 demoted every non-empty jobs_by_type entry back to :queued, overwriting the :complete that workers had just set. 2. reconcile_stale_queued/1 now also flips mechanism_status and radar_status back to :complete when the companion data (propagation_mechanism / contact_common_volume_radar) already exists. Drains the existing ~14k row backlog on the first post- deploy tick. 3. IemClient.parse_iemre_json/1 accepts "" and nil so a 200-with- empty-body IEMRE response doesn't FunctionClauseError the worker through four retry attempts. --- lib/microwaveprop/weather/iem_client.ex | 8 +++- .../workers/backfill_enqueue_worker.ex | 42 ++++++++++++++++++- .../workers/contact_weather_enqueue_worker.ex | 16 +++++++ .../microwaveprop/weather/iem_client_test.exs | 13 ++++++ .../workers/backfill_enqueue_worker_test.exs | 28 +++++++++++++ .../contact_weather_enqueue_worker_test.exs | 34 +++++++++++++++ 6 files changed, 139 insertions(+), 2 deletions(-) diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index e12de686..560e9401 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/lib/microwaveprop/weather/iem_client.ex @@ -199,11 +199,17 @@ defmodule Microwaveprop.Weather.IemClient do # --- Parsers --- - @spec parse_iemre_json(map()) :: [map()] + @spec parse_iemre_json(map() | binary() | nil) :: [map()] def parse_iemre_json(json) when is_map(json) do Map.get(json, "data", []) end + # IEM ERDDAP sometimes serves a 200 with an empty body for valid + # station/date combinations during backfill windows. Treat as "no + # data" rather than letting the worker stack-trace on a function + # clause mismatch. + def parse_iemre_json(body) when body in [nil, ""], do: [] + @spec parse_network_json(map()) :: [map()] def parse_network_json(json) when is_map(json) do json diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index 9255fb18..7d3e7c25 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -168,7 +168,47 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do 0 end - terrain_count + # MechanismClassifyWorker sets propagation_mechanism + mechanism_status + # atomically, so a row with a populated propagation_mechanism was + # successfully classified. A prior re-enqueue bug clobbered status + # back to :queued on subsequent cron cycles; the next worker run + # will just re-classify and set :complete again. Reconcile here so + # the backlog drains in one tick instead of waiting for each + # contact's worker to run a second time. + mechanism_count = + if :mechanism in types do + {count, _} = + Repo.update_all( + from(c in Contact, + where: c.mechanism_status == :queued and not is_nil(c.propagation_mechanism) + ), + set: [mechanism_status: :complete] + ) + + count + else + 0 + end + + # Radar reconciliation: contact_common_volume_radar has a direct + # contact_id FK (same shape as terrain_profiles). + radar_count = + if :radar in types do + {count, _} = + Repo.update_all( + from(c in Contact, + where: c.radar_status == :queued, + where: c.id in subquery(from(r in "contact_common_volume_radar", select: r.contact_id)) + ), + set: [radar_status: :complete] + ) + + count + else + 0 + end + + terrain_count + mechanism_count + radar_count end defp type_filter(types) do diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 03069063..fc182b63 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -28,6 +28,13 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do @spec enqueue_for_contact(Contact.t(), [atom()]) :: :ok def enqueue_for_contact(contact, types \\ [:hrrr, :weather, :terrain, :iemre, :radar, :mechanism]) do contact = Radio.ensure_positions!(contact) + # BackfillEnqueueWorker selects any contact with at least one + # *_status in [:pending, :queued, :failed] and calls this with all + # types. Filter out types that are already :complete for this + # contact so mark_status!/3 doesn't demote them back to :queued on + # every cron tick — that caused tens of thousands of + # mechanism/terrain rows to ping-pong between :complete and :queued. + types = Enum.reject(types, &already_complete?(contact, &1)) jobs_by_type = build_jobs_by_type(contact, types) bulk_jobs = @@ -492,4 +499,13 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do defp mark_status!(ids, field, [_ | _]), do: Radio.set_enrichment_status!(ids, field, :queued) defp mark_status!(ids, field, []), do: Radio.set_enrichment_status!(ids, field, :complete) + + # :narr is a virtual type synthesized from hrrr_status = :unavailable; + # it has no column of its own, so there's nothing to "already be + # complete." + defp already_complete?(_contact, :narr), do: false + + defp already_complete?(contact, type) do + Map.get(contact, :"#{type}_status") == :complete + end end diff --git a/test/microwaveprop/weather/iem_client_test.exs b/test/microwaveprop/weather/iem_client_test.exs index eddac93c..f8227883 100644 --- a/test/microwaveprop/weather/iem_client_test.exs +++ b/test/microwaveprop/weather/iem_client_test.exs @@ -355,6 +355,19 @@ defmodule Microwaveprop.Weather.IemClientTest do test "returns empty list for empty data array" do assert IemClient.parse_iemre_json(%{"data" => []}) == [] end + + # IEM occasionally returns a 200 with an empty body for valid-looking + # requests (upstream ERDDAP glitch during backfill windows). + # parse_iemre_json/1 used to only match is_map, so the empty binary + # blew up with FunctionClauseError and the job exhausted retries. + # Treat "" and nil as "no data" so the worker can mark the slot and move on. + test "returns empty list for empty string body" do + assert IemClient.parse_iemre_json("") == [] + end + + test "returns empty list for nil body" do + assert IemClient.parse_iemre_json(nil) == [] + end end describe "fetch_iemre/3" do diff --git a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs index cb994002..158d7d24 100644 --- a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs @@ -216,6 +216,34 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorkerTest do args: %{"limit" => 10, "types" => ["narr"]} }) end + + # MechanismClassifyWorker sets both propagation_mechanism and + # mechanism_status = :complete atomically. A prior bug in the + # re-enqueue path clobbered mechanism_status back to :queued while + # leaving propagation_mechanism populated. The reconciler flips + # those rows back to :complete so the loop stops. + test "reconciles mechanism_queued → complete when propagation_mechanism is already set" do + contact = create_contact() + + contact + |> Ecto.Changeset.change(%{ + mechanism_status: :queued, + propagation_mechanism: "troposcatter", + hrrr_status: :complete, + weather_status: :complete, + terrain_status: :complete, + iemre_status: :complete, + radar_status: :complete + }) + |> Repo.update!() + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => ["mechanism"]} + }) + + assert %Contact{mechanism_status: :complete} = Repo.get!(Contact, contact.id) + end end describe "reconcile stuck queued → unavailable" do diff --git a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs index 6a89ef20..092712c6 100644 --- a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs @@ -755,5 +755,39 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do updated = Repo.get!(Contact, contact.id) assert updated.hrrr_status == :unavailable end + + # BackfillEnqueueWorker picks up any contact with at least one *_status + # in [:pending, :queued, :failed] and calls enqueue_for_contact/2 with + # all types. mark_status! used to blindly demote every non-empty + # jobs_by_type entry back to :queued even when the worker had already + # set :complete — which caused tens of thousands of mechanism/terrain + # rows to ping-pong between :complete and :queued forever. + test "preserves :complete statuses when a sibling enrichment is re-enqueued" do + _station = create_asos_station("KDFW", 32.90, -97.04) + + contact = + create_contact() + |> Ecto.Changeset.change(%{ + weather_status: :queued, + hrrr_status: :complete, + terrain_status: :complete, + iemre_status: :complete, + radar_status: :complete, + mechanism_status: :complete + }) + |> Repo.update!() + + assert :ok = ContactWeatherEnqueueWorker.enqueue_for_contact(contact) + + updated = Repo.get!(Contact, contact.id) + # Weather still needed, should be re-marked queued. + assert updated.weather_status == :queued + # Everything already complete must stay complete — do NOT clobber. + assert updated.hrrr_status == :complete + assert updated.terrain_status == :complete + assert updated.iemre_status == :complete + assert updated.radar_status == :complete + assert updated.mechanism_status == :complete + end end end