From e641fce2789126ca4db063d41597da67a5f4ad17 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 20 Apr 2026 16:46:32 -0500 Subject: [PATCH] fix(backfill): re-probe :unavailable contacts after a 24 h cooldown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Contact list's 'Partial' badge was sticky: once a worker gave up on a contact (upstream 404, no nearby station, IEM throttling) or the stale-queued reconciler flipped it to :unavailable after 3 days, the backfill cron never looked at it again. Upstream archives heal over time — IEM backfills gaps, new ASOS stations come online, Rust worker gains capability — so rows stayed 'Partial' that shouldn't have. Changes: * BackfillEnqueueWorker.type_filter now also matches _status = :unavailable when contacts.updated_at is older than a 24 h cooldown. enqueue_for_contact is idempotent: if there's still nothing to fetch it'll flip the status to :complete on the empty-jobs branch; if new data shows up it builds fresh jobs. * reconcile_stale_queued_to_unavailable now bumps updated_at when flipping to :unavailable so the cooldown starts fresh — otherwise the main loop would pick it right back up on the same cron tick (stale updated_at < any reasonable cooldown) and immediately transition :unavailable → :complete, defeating the :unavailable signal entirely. Also: * Drop ScoresFile.point_score/3, dead since read_point moved to the pread fast path in commit bd3b114. * Clear Microwaveprop.Cache in WeatherGridTest setup. The new ProfilesFile read cache is keyed by (base_dir, valid_time); tests in this file reuse DateTime.utc_now() truncated to seconds and share the default base_dir, so a prior test's cached read leaks into the next one's expectations. --- lib/microwaveprop/propagation/scores_file.ex | 13 ---- .../workers/backfill_enqueue_worker.ex | 35 ++++++++++- test/microwaveprop/weather_grid_test.exs | 5 ++ .../workers/backfill_enqueue_worker_test.exs | 61 +++++++++++++++++++ 4 files changed, 99 insertions(+), 15 deletions(-) diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index 903d746f..3e5f6fe2 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -409,19 +409,6 @@ defmodule Microwaveprop.Propagation.ScoresFile do defp clamp(n, _lo, hi) when n > hi, do: hi defp clamp(n, _lo, _hi), do: n - defp point_score(payload, lat, lon) do - %{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols, scores: body} = payload - row = round((lat - lat_min) / step) - col = round((lon - lon_min) / step) - - if row in 0..(n_rows - 1) and col in 0..(n_cols - 1) do - case :binary.at(body, row * n_cols + col) do - @no_data -> nil - score -> score - end - end - end - defp max_datetime([]), do: nil defp max_datetime(times), do: Enum.max(times, DateTime) diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index 582ebef4..4410b3da 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -15,6 +15,17 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do require Logger @enrichable [:pending, :queued, :failed] + + # How long a contact must sit at :unavailable before the backfill + # will re-probe it. :unavailable is reached either because a worker + # gave up (upstream 404/5xx, no nearby station, pre-2014 archive + # gap) or because the stale-queued reconciler flipped it after 3 + # days of :queued without data. Upstream conditions change over + # time (new ASOS stations, IEM backfills, Rust worker new + # capabilities), so "Partial" contacts must eventually get a second + # look — without pounding the upstream every 30 min. 24 h strikes + # a balance. + @unavailable_reprobe_cooldown_seconds 86_400 @valid_types ~w(hrrr weather terrain iemre narr radar mechanism) # Virtual types have no _status column on contacts; they're synthesized @@ -121,15 +132,23 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do |> DateTime.add(-@stale_queued_cutoff_seconds, :second) |> DateTime.truncate(:second) + now = DateTime.truncate(DateTime.utc_now(), :second) + types |> Enum.reject(&(&1 in @virtual_types)) |> Enum.reduce(0, fn type, acc -> field = status_column(type) + # Bump updated_at to now when flipping so the + # @unavailable_reprobe_cooldown window starts fresh — otherwise + # the main loop's :unavailable-reprobe filter would pick this + # contact back up on the same cron tick (stale timestamp < any + # reasonable cooldown) and immediately mark it :complete on the + # empty-jobs branch, defeating the point of flagging :unavailable. {n, _} = Contact |> where([c], field(c, ^field) == :queued and c.updated_at < ^cutoff) - |> Repo.update_all(set: [{field, :unavailable}]) + |> Repo.update_all(set: [{field, :unavailable}, {:updated_at, now}]) acc + n end) @@ -158,6 +177,11 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do end defp type_filter(types) do + reprobe_cutoff = + DateTime.utc_now() + |> DateTime.add(-@unavailable_reprobe_cooldown_seconds, :second) + |> DateTime.truncate(:second) + Enum.reduce(types, dynamic(false), fn :narr, acc -> # NARR targets pre-2014 contacts where HRRR is unavailable. The NCEI @@ -169,7 +193,14 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do type, acc -> field = status_column(type) - dynamic([c], ^acc or field(c, ^field) in ^@enrichable) + + # Include :unavailable rows older than the cooldown so "Partial" + # contacts get another chance when upstream archives heal. + dynamic( + [c], + ^acc or field(c, ^field) in ^@enrichable or + (field(c, ^field) == :unavailable and c.updated_at < ^reprobe_cutoff) + ) end) end end diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index c8ee2cb1..96d4fba5 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -11,6 +11,11 @@ defmodule Microwaveprop.WeatherGridTest do # test sandbox). Clear it so each test gets a fresh cache and the DB load # path actually runs. GridCache.clear() + # Microwaveprop.Cache now memoizes ProfilesFile.read/1 keyed by + # `(base_dir, valid_time)`. Tests within this file share the default + # base_dir and frequently reuse `DateTime.utc_now()` truncated to seconds, + # so a prior test's cached read leaks into the next one's assertion. + Microwaveprop.Cache.clear() :ok end diff --git a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs index edb9077f..cb994002 100644 --- a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs @@ -138,6 +138,67 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorkerTest do assert_receive {:enqueue_complete, 1} end + # :unavailable is a terminal state in the sense that workers don't + # retry it, but the stale-queued reconciler flips :queued → :unavailable + # after 3 days when no data has landed — which means a contact can end + # up "Partial" forever even if the upstream archive later starts + # serving data (new ASOS station in range, re-enabled IEM endpoint). + # The backfill must re-probe :unavailable contacts so they get another + # shot at enrichment; otherwise the contact list shows "Partial" for + # rows that haven't been touched in months. + test "re-selects :unavailable contacts older than the cooldown for another shot at enrichment" do + import Ecto.Query + + Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete") + + contact = create_contact() + + contact + |> Ecto.Changeset.change(%{ + hrrr_status: :complete, + weather_status: :unavailable, + terrain_status: :complete, + iemre_status: :complete + }) + |> Repo.update!() + + # Force updated_at past the 24 h cooldown so the type_filter's + # :unavailable branch picks this contact up. + old = DateTime.utc_now() |> DateTime.add(-2 * 86_400, :second) |> DateTime.truncate(:second) + Repo.update_all(from(c in Contact, where: c.id == ^contact.id), set: [updated_at: old]) + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => @non_era5_types} + }) + + assert_receive {:enqueue_complete, 1} + end + + test "leaves :unavailable contacts inside the cooldown window alone" do + Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete") + + contact = create_contact() + + contact + |> Ecto.Changeset.change(%{ + hrrr_status: :complete, + weather_status: :unavailable, + terrain_status: :complete, + iemre_status: :complete + }) + |> Repo.update!() + + # updated_at is fresh (right now) — inside the 24 h cooldown. + # Contact must not be re-probed yet. + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => @non_era5_types} + }) + + assert_receive {:enqueue_complete, 0} + end + # :narr is a virtual type (no narr_status column) — it targets contacts # whose hrrr_status is :unavailable. BackfillEnqueueWorker must not try # to reconcile a non-existent narr_status column or use it for priority