Give up on stuck :queued backfill contacts after 3 days

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.

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 (<field>, 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

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