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.
This commit is contained in:
parent
5b41ad00e7
commit
bfddb898bf
6 changed files with 139 additions and 2 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue