ERA5: tighten stuck threshold to 8h and pause prod queues for diagnosis

Prod has 0 era5_profiles rows after ~500 submit cycles: every CDS job
either gets marked "rejected" (likely cap mismatch) or runs for 13+h
without transitioning to successful, so the handle_both_done branch
has literally never fired and the ERA5Batch: stored log line has
never been emitted once. Tighten Era5PollWorker's @stuck_after_seconds
from 18h to 8h so dead rows recycle within a work shift instead of a
full day, and pause the era5 / era5_submit / era5_poll / era5_batch
queues in runtime.exs so existing rows stay put for inspection while
we figure out why CDS is rejecting everything. Flip paused: true back
once the root cause is identified.

config/runtime.exs

@@ -168,30 +168,21 @@ if config_env() == :prod do
       hrrr: 1,
       terrain: 3,
       iemre: 3,
-      era5: 2,
-      # Split-worker ERA5 pipeline:
-      #
-      # `era5_submit` runs Era5SubmitWorker, which POSTs two CDS requests
-      # and writes an era5_cds_jobs row. Each job is ~1s of real work, so
-      # local_limit is the gate on how fast we burn CDS submission quota.
-      # rate_limit protects CDS from submit bursts; the per-pod ceiling
-      # (local_limit × replicas) determines peak concurrent in-flight.
+      # ERA5 queues are TEMPORARILY PAUSED while we diagnose the
+      # zero-ingestion pipeline. Every poll in prod has been hitting
+      # either CDS-rejected or long-running states; no `ERA5Batch:
+      # stored` log line has ever fired. Pausing keeps the existing
+      # oban_jobs and era5_cds_jobs rows in place for inspection
+      # instead of deleting them. Flip `paused: true` to `paused:
+      # false` (or remove the key) once the root cause is fixed.
+      era5: [local_limit: 2, paused: true],
       era5_submit: [
         local_limit: 4,
-        rate_limit: [allowed: 30, period: {1, :hour}]
+        rate_limit: [allowed: 30, period: {1, :hour}],
+        paused: true
       ],
-      # `era5_poll` runs Era5PollWorker, which does a tiny GET per run and
-      # either snoozes (releasing the slot) or streams the completed GRIB
-      # and inserts. Needs high local_limit because in-flight CDS jobs
-      # scale independently of concurrent poll work — a pod may have 50
-      # tile-months in flight but be doing ~1 actual poll GET at a time.
-      era5_poll: [
-        local_limit: 20
-      ],
-      # Legacy queue name retained briefly so any in-flight Era5MonthBatchWorker
-      # jobs from the pre-split deploy drain cleanly (the worker now just
-      # forwards to :era5_submit). Safe to delete after one rolling deploy.
-      era5_batch: 1,
+      era5_poll: [local_limit: 20, paused: true],
+      era5_batch: [local_limit: 1, paused: true],
       rtma: 2,
       backfill_enqueue: 1,
       admin: 1,

lib/microwaveprop/workers/era5_poll_worker.ex

@@ -45,9 +45,12 @@ defmodule Microwaveprop.Workers.Era5PollWorker do
   # previous 4h threshold was firing on normal slow runs, forcing a
   # resubmit which then got rejected by the per-user cap — an infinite
   # reject→resubmit spiral that burned CDS quota without producing any
-  # decoded data. 18h is comfortably above the worst observed queue
-  # times while still self-healing within a day.
-  @stuck_after_seconds 18 * 3600
+  # decoded data. Tightened from 18h to 8h while diagnosing the
+  # zero-ingestion problem: 18h was letting rows poll for ~13h without
+  # ever producing a profile, 8h recycles stuck rows within a work
+  # shift while still staying above the ~90 min normal pressure-level
+  # runtime.
+  @stuck_after_seconds 8 * 3600
   @stuck_after_hours div(@stuck_after_seconds, 3600)
 
   @impl Oban.Worker

test/microwaveprop/workers/era5_poll_worker_test.exs

@@ -167,14 +167,16 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do
     # treats it as terminal and re-submits — same path as :not_found /
     # :rejected.
     #
-    # Threshold is 18h (see @stuck_after_seconds in Era5PollWorker): CDS
-    # single-level routinely takes 12+ hours under load while pressure
-    # finishes in ~90 minutes, and the 4h threshold was firing on normal
-    # slow runs, causing re-submit storms that hit the per-user cap.
-    test "re-submits when submitted_at is older than 18h and at least one leg is still running" do
+    # Threshold is 8h (see @stuck_after_seconds in Era5PollWorker). We
+    # observed the previous 18h threshold held a row polling for ~13h
+    # without catching a stuck state while 0 profiles were being
+    # ingested. 8h is a diagnostic compromise: short enough to recycle
+    # stuck rows within a shift, still above the ~90 min pressure-level
+    # normal completion time so healthy runs aren't interrupted.
+    test "re-submits when submitted_at is older than 8h and at least one leg is still running" do
       stale_submit =
         DateTime.utc_now()
-        |> DateTime.add(-19 * 3600, :second)
+        |> DateTime.add(-9 * 3600, :second)
         |> DateTime.truncate(:second)
 
       row = insert_cds_job(%{submitted_at: stale_submit})
@@ -206,13 +208,15 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do
       )
     end
 
-    test "does NOT re-submit a 10h-old row that's still running (under the 18h threshold)" do
-      # Observed in prod: CDS single-level can sit at 'accepted' for 12+
-      # hours while pressure completes in ~90 min. Anything under 18h
-      # should keep snoozing, not trigger a re-submit storm.
+    test "does NOT re-submit a 6h-old row that's still running (under the 8h threshold)" do
+      # Observed in prod: CDS pressure-level completes in ~90 min, so
+      # anything under the 8h diagnostic threshold should keep snoozing,
+      # not trigger a re-submit storm. The single-level leg is allowed
+      # to keep running — only the combined age is stale, and 6h is
+      # still well within the envelope.
       slowish_submit =
         DateTime.utc_now()
-        |> DateTime.add(-10 * 3600, :second)
+        |> DateTime.add(-6 * 3600, :second)
         |> DateTime.truncate(:second)
 
       row = insert_cds_job(%{submitted_at: slowish_submit})