diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index 7d3e7c25..ff01843a 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -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 diff --git a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs index 158d7d24..e63cfe3c 100644 --- a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs @@ -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