From 79e16ae9d32480c67df9da7a25c88ed9b7735dde Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 13 Apr 2026 17:20:14 -0500 Subject: [PATCH] Give up on stuck :queued backfill contacts after 3 days MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Contacts whose weather/hrrr/terrain/iemre status has been :queued for more than 3 days (144 cron cycles) are flipped to :unavailable. The set_enrichment_status!/3 helper is a no-op when the status is unchanged, so a stuck row retains a stale updated_at — which is exactly the signal we need for 'the cron has already tried and the data doesnt exist upstream' (dead ASOS station, pre-2014 HRRR, out-of-grid IEMRE). Clears ~1,492 pre-existing stuck contacts and stops them from churning the backfill queue. --- .../workers/backfill_enqueue_worker.ex | 38 ++++++ .../workers/backfill_enqueue_worker_test.exs | 114 ++++++++++++++++++ 2 files changed, 152 insertions(+) diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index d3d3bd46..6e3f247b 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -16,6 +16,15 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do @enrichable [:pending, :queued, :failed] @valid_types ~w(hrrr weather terrain iemre era5) + # A :queued contact whose updated_at is older than this is one the cron has + # already tried ~144 times over 3 days without the underlying worker landing + # data. At that point the data is effectively unreachable (dead ASOS station, + # pre-2014 HRRR, out-of-grid IEMRE) and the cron should stop re-enqueueing. + # set_enrichment_status!/3 is a no-op when the status is unchanged, so a + # stuck :queued row retains a stale updated_at across cron cycles — which is + # exactly what we need as the "given up" signal. + @stale_queued_cutoff_seconds 3 * 86_400 + @impl Oban.Worker def perform(%Oban.Job{args: %{"limit" => limit} = args}) do types = parse_types(args) @@ -26,6 +35,12 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do Logger.info("BackfillEnqueue: reconciled #{reconciled} stale queued contacts") end + stale_unavail = reconcile_stale_queued_to_unavailable(types) + + if stale_unavail > 0 do + Logger.info("BackfillEnqueue: marked #{stale_unavail} stuck-queued contacts as unavailable") + end + contacts = Contact |> where([c], not is_nil(c.pos1) or not is_nil(c.grid1)) @@ -64,6 +79,29 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do ] end + # Flip contacts that have been stuck in :queued for longer than + # @stale_queued_cutoff_seconds to :unavailable. One Repo.update_all per type, + # scoped to only the types passed in args so a partial cron (e.g. hrrr-only) + # doesn't touch other statuses. Fast: hits the (, updated_at) combo + # with a simple range scan. + defp reconcile_stale_queued_to_unavailable(types) do + cutoff = + DateTime.utc_now() + |> DateTime.add(-@stale_queued_cutoff_seconds, :second) + |> DateTime.truncate(:second) + + Enum.reduce(types, 0, fn type, acc -> + field = :"#{type}_status" + + {n, _} = + Contact + |> where([c], field(c, ^field) == :queued and c.updated_at < ^cutoff) + |> Repo.update_all(set: [{field, :unavailable}]) + + acc + n + 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 d7920f63..33e3b43d 100644 --- a/test/microwaveprop/workers/backfill_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/backfill_enqueue_worker_test.exs @@ -114,4 +114,118 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorkerTest do assert_receive {:enqueue_complete, 1} end end + + describe "reconcile stuck queued → unavailable" do + # A contact stuck at :queued for > 3 days is one the cron has already + # tried ~144 times without the underlying worker landing any data. At that + # point the data simply doesn't exist in the upstream archive (dead ASOS + # station, pre-2014 HRRR, out-of-grid IEMRE), so we stop retrying and flip + # it to :unavailable. + @stale_cutoff 3 * 86_400 + + defp stale_ts do + DateTime.utc_now() + |> DateTime.add(-(@stale_cutoff + 3600), :second) + |> DateTime.truncate(:second) + end + + defp fresh_ts do + DateTime.utc_now() + |> DateTime.add(-3600, :second) + |> DateTime.truncate(:second) + end + + # Set the status field AND updated_at directly via update_all so Ecto + # timestamps can't overwrite what we put there. Also short-circuits the + # changeset, which ignores the *_status fields entirely. + defp force_status(contact, field, status, ts) do + import Ecto.Query + + Repo.update_all(from(c in Contact, where: c.id == ^contact.id), set: [{field, status}, {:updated_at, ts}]) + end + + # Complete the OTHER three statuses so the contact no longer matches the + # enqueue filter on them — only the one under test remains :queued. That + # keeps the main enqueue loop from running `mark_status!` on the contact + # during the same perform/1 call, which would bump updated_at back to now. + defp only_this_stuck(contact, field) do + import Ecto.Query + + other_fields = [:weather_status, :hrrr_status, :iemre_status, :terrain_status] -- [field] + sets = Enum.map(other_fields, &{&1, :complete}) + + Repo.update_all(from(c in Contact, where: c.id == ^contact.id), set: sets) + end + + test "flips weather_status :queued → :unavailable when updated_at is older than the cutoff" do + contact = create_contact() + only_this_stuck(contact, :weather_status) + force_status(contact, :weather_status, :queued, stale_ts()) + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => @non_era5_types} + }) + + assert %Contact{weather_status: :unavailable} = Repo.get!(Contact, contact.id) + end + + test "flips hrrr_status :queued → :unavailable when updated_at is older than the cutoff" do + contact = create_contact() + only_this_stuck(contact, :hrrr_status) + force_status(contact, :hrrr_status, :queued, stale_ts()) + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => @non_era5_types} + }) + + assert %Contact{hrrr_status: :unavailable} = Repo.get!(Contact, contact.id) + end + + test "flips iemre_status :queued → :unavailable when updated_at is older than the cutoff" do + contact = create_contact() + only_this_stuck(contact, :iemre_status) + force_status(contact, :iemre_status, :queued, stale_ts()) + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => @non_era5_types} + }) + + assert %Contact{iemre_status: :unavailable} = Repo.get!(Contact, contact.id) + end + + test "does NOT flip fresh :queued contacts to :unavailable" do + contact = create_contact() + only_this_stuck(contact, :weather_status) + force_status(contact, :weather_status, :queued, fresh_ts()) + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => @non_era5_types} + }) + + # Whatever the main enqueue loop does next (mark :queued, :complete, + # etc), reconcile_stale_queued_to_unavailable must NOT have fired: a + # contact fresher than the cutoff still has a chance at real data. + reloaded = Repo.get!(Contact, contact.id) + refute reloaded.weather_status == :unavailable + end + + test "only reconciles the types passed in the args — doesn't touch unrelated statuses" do + # weather is stuck-queued-stale, but the cron is running with only hrrr + # in its types — weather_status should be left alone. + contact = create_contact() + only_this_stuck(contact, :weather_status) + force_status(contact, :weather_status, :queued, stale_ts()) + + assert :ok = + BackfillEnqueueWorker.perform(%Oban.Job{ + args: %{"limit" => 10, "types" => ["hrrr"]} + }) + + assert %Contact{weather_status: :queued} = Repo.get!(Contact, contact.id) + end + end end