fix(hrrr): mark contacts :unavailable on failed fetch tasks
When the Rust hrrr-point-worker marks a task as `failed` (upstream `idx HTTP 404` or wgrib2 decode error on old archive formats), the corresponding contacts sit at `hrrr_status = :queued` until the generic 3-day stale-queued reconcile flips them. Speed that up: on every BackfillEnqueue tick, join contacts to `hrrr_fetch_tasks` by rounded HRRR hour and flip any `:queued` contact whose cycle is permanently `failed` to `:unavailable` immediately. Motivation: a handful of 2018-08-04/05 cycles (t20z/t21z/t15z/t17z) are genuinely missing `wrfsfcf`/`wrfprsf` files on the NOAA S3 archive — other hours on the same dates are fine, so this is an upstream data gap rather than a structural pre-2019 issue. Users see an accurate "Partial" marker on those contacts instead of a three-day `:queued` stall. Tested: 4 new cases covering rounding, failed-vs-done, and the `hrrr not in types` skip branch; full suite 2283/0.
This commit is contained in:
parent
b8e284bde1
commit
004f302374
2 changed files with 156 additions and 0 deletions
|
|
@ -53,6 +53,14 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do
|
|||
Logger.info("BackfillEnqueue: reconciled #{reconciled} stale queued contacts")
|
||||
end
|
||||
|
||||
failed_task_unavail = reconcile_failed_hrrr_task_contacts(types)
|
||||
|
||||
if failed_task_unavail > 0 do
|
||||
Logger.info(
|
||||
"BackfillEnqueue: marked #{failed_task_unavail} hrrr-queued contacts as unavailable (upstream archive gap)"
|
||||
)
|
||||
end
|
||||
|
||||
stale_unavail = reconcile_stale_queued_to_unavailable(types)
|
||||
|
||||
if stale_unavail > 0 do
|
||||
|
|
@ -149,6 +157,43 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do
|
|||
end)
|
||||
end
|
||||
|
||||
# A `hrrr_fetch_tasks` row in `failed` state is one the Rust worker
|
||||
# gave up on after retrying — either wgrib2 decode errors (very old
|
||||
# HRRR format) or `idx HTTP 404` from the upstream archive. There's
|
||||
# no path to profiles for any contact whose rounded HRRR hour lands
|
||||
# on that valid_time, so flip them to `:unavailable` immediately
|
||||
# rather than letting them wait 3 days for the generic stale-queue
|
||||
# reconcile. Matches `HrrrClient.nearest_hrrr_hour/1` rounding in SQL:
|
||||
# drop minutes, bump the hour up when the original minute was >= 30.
|
||||
defp reconcile_failed_hrrr_task_contacts(types) do
|
||||
if :hrrr in types do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
failed_vts_sub =
|
||||
from(t in "hrrr_fetch_tasks",
|
||||
where: t.status == "failed",
|
||||
select: t.valid_time
|
||||
)
|
||||
|
||||
{n, _} =
|
||||
Contact
|
||||
|> where([c], c.hrrr_status == :queued)
|
||||
|> where(
|
||||
[c],
|
||||
fragment(
|
||||
"date_trunc('hour', ?) + (CASE WHEN date_part('minute', ?) >= 30 THEN interval '1 hour' ELSE interval '0' END)",
|
||||
c.qso_timestamp,
|
||||
c.qso_timestamp
|
||||
) in subquery(failed_vts_sub)
|
||||
)
|
||||
|> Repo.update_all(set: [hrrr_status: :unavailable, updated_at: now])
|
||||
|
||||
n
|
||||
else
|
||||
0
|
||||
end
|
||||
end
|
||||
|
||||
# Fast SQL reconciliation for contacts stuck in "queued" where data already exists.
|
||||
# Terrain profiles have a direct contact_id FK so we can reconcile in bulk.
|
||||
defp reconcile_stale_queued(types) do
|
||||
|
|
|
|||
|
|
@ -359,4 +359,115 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorkerTest do
|
|||
assert %Contact{weather_status: :queued} = Repo.get!(Contact, contact.id)
|
||||
end
|
||||
end
|
||||
|
||||
describe "reconcile hrrr_status against failed hrrr_fetch_tasks" do
|
||||
# When the Rust hrrr-point-worker marks a task as `failed` (upstream
|
||||
# archive hole, wgrib2 decode error, etc), the matching contacts
|
||||
# stay at :queued until the 3-day stale reconcile flips them. For
|
||||
# unambiguous upstream data gaps we can do better: if a task is in
|
||||
# :failed state, every contact whose rounded HRRR hour matches that
|
||||
# task's valid_time has no path to data — flip it to :unavailable
|
||||
# immediately so users see an accurate "Partial" marker instead of
|
||||
# waiting 3 days.
|
||||
import Ecto.Query
|
||||
|
||||
defp insert_failed_task(valid_time) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :microsecond)
|
||||
|
||||
Repo.insert_all(
|
||||
"hrrr_fetch_tasks",
|
||||
[
|
||||
%{
|
||||
id: Ecto.UUID.bingenerate(),
|
||||
valid_time: DateTime.to_naive(valid_time),
|
||||
points: [%{"lat" => 32.9, "lon" => -97.0}],
|
||||
status: "failed",
|
||||
attempt: 3,
|
||||
error: "fetch: idx HTTP 404",
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
test "flips hrrr_status :queued → :unavailable for contacts whose rounded hour matches a failed task" do
|
||||
# QSO at 17:02 — nearest_hrrr_hour rounds to 17:00 (minute < 30).
|
||||
contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:02:00Z]})
|
||||
only_this_stuck(contact, :hrrr_status)
|
||||
force_status(contact, :hrrr_status, :queued, fresh_ts())
|
||||
|
||||
insert_failed_task(~U[2018-08-05 17:00:00Z])
|
||||
|
||||
assert :ok =
|
||||
BackfillEnqueueWorker.perform(%Oban.Job{
|
||||
args: %{"limit" => 10, "types" => ["hrrr"]}
|
||||
})
|
||||
|
||||
assert %Contact{hrrr_status: :unavailable} = Repo.get!(Contact, contact.id)
|
||||
end
|
||||
|
||||
test "rounds UP when qso minute >= 30 before matching failed tasks" do
|
||||
# 17:45 rounds to 18:00 (minute >= 30), NOT 17:00.
|
||||
contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:45:00Z]})
|
||||
only_this_stuck(contact, :hrrr_status)
|
||||
force_status(contact, :hrrr_status, :queued, fresh_ts())
|
||||
|
||||
insert_failed_task(~U[2018-08-05 18:00:00Z])
|
||||
|
||||
assert :ok =
|
||||
BackfillEnqueueWorker.perform(%Oban.Job{
|
||||
args: %{"limit" => 10, "types" => ["hrrr"]}
|
||||
})
|
||||
|
||||
assert %Contact{hrrr_status: :unavailable} = Repo.get!(Contact, contact.id)
|
||||
end
|
||||
|
||||
test "leaves contacts alone when the matching task is still :queued/:done" do
|
||||
contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:02:00Z]})
|
||||
only_this_stuck(contact, :hrrr_status)
|
||||
force_status(contact, :hrrr_status, :queued, fresh_ts())
|
||||
|
||||
# No failed task for this hour.
|
||||
now = DateTime.truncate(DateTime.utc_now(), :microsecond)
|
||||
|
||||
Repo.insert_all("hrrr_fetch_tasks", [
|
||||
%{
|
||||
id: Ecto.UUID.bingenerate(),
|
||||
valid_time: ~N[2018-08-05 17:00:00],
|
||||
points: [%{"lat" => 32.9, "lon" => -97.0}],
|
||||
status: "done",
|
||||
attempt: 1,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
])
|
||||
|
||||
assert :ok =
|
||||
BackfillEnqueueWorker.perform(%Oban.Job{
|
||||
args: %{"limit" => 10, "types" => ["hrrr"]}
|
||||
})
|
||||
|
||||
refute Repo.get!(Contact, contact.id).hrrr_status == :unavailable
|
||||
end
|
||||
|
||||
test "skips the reconcile when :hrrr is not in the types arg" do
|
||||
contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:02:00Z]})
|
||||
only_this_stuck(contact, :hrrr_status)
|
||||
force_status(contact, :hrrr_status, :queued, fresh_ts())
|
||||
|
||||
insert_failed_task(~U[2018-08-05 17:00:00Z])
|
||||
|
||||
assert :ok =
|
||||
BackfillEnqueueWorker.perform(%Oban.Job{
|
||||
args: %{"limit" => 10, "types" => ["weather"]}
|
||||
})
|
||||
|
||||
# hrrr_status untouched because the reconcile only runs when :hrrr
|
||||
# is in the active type list.
|
||||
assert Repo.get!(Contact, contact.id).hrrr_status == :queued
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue