fix(backfill): flip hrrr_status :queued -> :complete when data already present
hrrr_placeholder_jobs previously returned [contact.id] for any post-2014 contact with pos1, regardless of whether the HRRR profiles were already in the DB. That meant BackfillEnqueueWorker re-flagged the same ~22k contacts as :queued on every 30 min cron tick forever — HrrrPointEnqueuer saw 0 missing points and inserted nothing, but the status field never advanced. Now the placeholder also returns [] when every path point has an hrrr_profile row, so mark_hrrr_status!/3's empty-list branch flips the contact to :complete directly. Genuinely-missing contacts still flow through HrrrPointEnqueuer unchanged.
This commit is contained in:
parent
cb9d9fb30b
commit
cc9220b8d2
2 changed files with 43 additions and 1 deletions
|
|
@ -77,17 +77,36 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
|
||||||
# mark_hrrr_status!/3 flips hrrr_queued on the contact row. Actual
|
# mark_hrrr_status!/3 flips hrrr_queued on the contact row. Actual
|
||||||
# scheduling happens via HrrrPointEnqueuer.enqueue_for_contacts/1
|
# scheduling happens via HrrrPointEnqueuer.enqueue_for_contacts/1
|
||||||
# above. Returns [] for pre-coverage contacts (NARR's territory) so
|
# above. Returns [] for pre-coverage contacts (NARR's territory) so
|
||||||
# the status stays :unavailable.
|
# the status stays :unavailable, and [] for contacts whose HRRR
|
||||||
|
# profiles are already fully present in the DB so mark_hrrr_status!/3
|
||||||
|
# flips them directly to :complete — otherwise backfill re-flags the
|
||||||
|
# contact as :queued every cron cycle forever (HrrrPointEnqueuer
|
||||||
|
# sees 0 missing points, inserts nothing, but the status stays
|
||||||
|
# :queued and the contact is picked up again next tick).
|
||||||
defp hrrr_placeholder_jobs(contacts) do
|
defp hrrr_placeholder_jobs(contacts) do
|
||||||
Enum.flat_map(contacts, fn contact ->
|
Enum.flat_map(contacts, fn contact ->
|
||||||
cond do
|
cond do
|
||||||
is_nil(contact.pos1) -> []
|
is_nil(contact.pos1) -> []
|
||||||
NarrClient.in_coverage?(contact.qso_timestamp) -> []
|
NarrClient.in_coverage?(contact.qso_timestamp) -> []
|
||||||
|
hrrr_data_fully_present?(contact) -> []
|
||||||
true -> [contact.id]
|
true -> [contact.id]
|
||||||
end
|
end
|
||||||
end)
|
end)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defp hrrr_data_fully_present?(contact) do
|
||||||
|
alias Microwaveprop.Weather.HrrrClient
|
||||||
|
|
||||||
|
rounded = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp)
|
||||||
|
|
||||||
|
contact
|
||||||
|
|> Radio.contact_path_points()
|
||||||
|
|> Enum.all?(fn {lat, lon} ->
|
||||||
|
{rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon)
|
||||||
|
Weather.has_hrrr_profile?(rlat, rlon, rounded)
|
||||||
|
end)
|
||||||
|
end
|
||||||
|
|
||||||
defp mark_enrichment_statuses(contact, types, jobs_by_type) do
|
defp mark_enrichment_statuses(contact, types, jobs_by_type) do
|
||||||
ids = [contact.id]
|
ids = [contact.id]
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -282,6 +282,29 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do
|
||||||
assert [] = Repo.all(from t in "hrrr_fetch_tasks", select: t.id)
|
assert [] = Repo.all(from t in "hrrr_fetch_tasks", select: t.id)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "enqueue_for_contact marks hrrr_status :complete when all path points already have profiles" do
|
||||||
|
# Regression: ~22k contacts stuck at :queued forever because
|
||||||
|
# HrrrPointEnqueuer saw 0 missing points (data already landed)
|
||||||
|
# and inserted nothing, but hrrr_placeholder_jobs always returned
|
||||||
|
# [contact.id] so mark_hrrr_status!/3 re-flagged the row as
|
||||||
|
# :queued every backfill cron tick. Contacts with full data must
|
||||||
|
# flip to :complete.
|
||||||
|
# grid2 = nil prevents Radio.ensure_positions! from deriving pos2,
|
||||||
|
# so contact_path_points returns only the pos1 single-point form.
|
||||||
|
contact =
|
||||||
|
create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}, pos2: nil, grid2: nil})
|
||||||
|
|
||||||
|
rounded = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp)
|
||||||
|
|
||||||
|
# Seed an hrrr_profile at the exact HRRR grid cell for pos1
|
||||||
|
{rlat, rlon} = Weather.round_to_hrrr_grid(32.907, -97.038)
|
||||||
|
{:ok, _} = Weather.upsert_hrrr_profile(%{lat: rlat, lon: rlon, valid_time: rounded, profile: []})
|
||||||
|
|
||||||
|
:ok = ContactWeatherEnqueueWorker.enqueue_for_contact(contact, [:hrrr])
|
||||||
|
|
||||||
|
assert %Contact{hrrr_status: :complete} = Repo.get!(Contact, contact.id)
|
||||||
|
end
|
||||||
|
|
||||||
test "re-enqueuing the same valid_time UPSERT-unions the points" do
|
test "re-enqueuing the same valid_time UPSERT-unions the points" do
|
||||||
import Ecto.Query
|
import Ecto.Query
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue