From 5cfb9e6c8e455670870bca42daa40966652d2014 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 09:26:09 -0500 Subject: [PATCH] fix(propagation): chain survives permanent step failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telemetry showed ~66 PropagationGridWorker exceptions per 6h with 55 ArgumentErrors and 11 TimeoutErrors, producing ~13 discarded chain steps. Each discard broke the chain: subsequent forecast hours were never enqueued, leaving the score store with huge gaps (e.g. at 14:11 UTC the earliest available forecast was 18:00, because f00-f05 all failed somewhere upstream and nothing ran after them). Three changes: 1. PropagationGridWorker: on the final attempt, still enqueue fh+1 even when this step failed. Oban discards the current job normally — but the rest of the chain keeps running, so one bad hour doesn't take out the remaining 12-18. The rescue is factored into a tested public helper. 2. HrrrClient.parse_idx: skip malformed idx lines instead of raising. NOAA S3 occasionally serves an HTML error page as the idx body, and the old strict String.to_integer path raised ArgumentError on the first non-numeric line and took down the chain step. This is the root cause of the 55 ArgumentErrors. 3. JS renderTimeline: when no forecast hour is at-or-before wall-clock (all times are future — the gap scenario the fixes above are designed to prevent), stop labeling the earliest future slot "Now". Lets the user see honest "+Nh" offsets instead of a lie on the pill. --- assets/js/propagation_map_hook.ts | 12 +++-- lib/microwaveprop/weather/hrrr_client.ex | 26 +++++++---- .../workers/propagation_grid_worker.ex | 43 ++++++++++++++++- .../weather/hrrr_client_test.exs | 21 +++++++++ .../workers/propagation_grid_worker_test.exs | 46 +++++++++++++++++++ 5 files changed, 132 insertions(+), 16 deletions(-) diff --git a/assets/js/propagation_map_hook.ts b/assets/js/propagation_map_hook.ts index dcc9b347..92fc4535 100644 --- a/assets/js/propagation_map_hook.ts +++ b/assets/js/propagation_map_hook.ts @@ -1333,16 +1333,18 @@ export const PropagationMap: Record & { // "Now" = the latest forecast hour at or before wall-clock. Picking // by absolute distance rounds up past the half-hour and labels a // future slot "Now" (e.g. at 18:35 UTC, 19:00 is 25 min away vs. - // 18:00 at 35 min). Fall back to the earliest slot if they're all - // in the future. + // 18:00 at 35 min). When every available forecast hour is in the + // future (e.g. a pipeline gap where f00-f05 got discarded and only + // f06+ made it onto disk), leave everything as "+Nh" labels rather + // than lying with a "Now" tag on a future slot. const nowMs = now.getTime() const pastOrNow = items.filter(t => t.dt.getTime() <= nowMs) - const closestIdx = pastOrNow.length > 0 + const nowIdx = pastOrNow.length > 0 ? items.indexOf(pastOrNow.reduce((latest, t) => t.dt.getTime() > latest.dt.getTime() ? t : latest)) - : items.reduce((earliest, t, i) => items[earliest].dt.getTime() < t.dt.getTime() ? earliest : i, 0) + : -1 items.forEach((t, i) => { - if (i === closestIdx) { + if (i === nowIdx) { t.label = "Now" } else { const diff = t.offsetH diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 9a68a61e..6de60db7 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -189,19 +189,27 @@ defmodule Microwaveprop.Weather.HrrrClient do @spec parse_idx(String.t()) :: [%{msg: integer(), offset: integer(), var: String.t(), level: String.t()}] def parse_idx(text) do + # Drop lines that don't start with "msg:offset:" — NOAA S3 sometimes + # returns an HTML error page as the idx body during brief outages, + # and the strict String.to_integer/1 path raised ArgumentError and + # killed the whole chain step. Skip unparseable lines and keep + # whatever well-formed rows are present. text |> String.split("\n") |> Enum.reject(&(&1 == "")) - |> Enum.map(fn line -> - parts = String.split(line, ":", parts: 8) + |> Enum.flat_map(&parse_idx_line/1) + end - %{ - msg: String.to_integer(Enum.at(parts, 0)), - offset: String.to_integer(Enum.at(parts, 1)), - var: Enum.at(parts, 3), - level: Enum.at(parts, 4) - } - end) + defp parse_idx_line(line) do + parts = String.split(line, ":", parts: 8) + + with [msg_s, off_s | _] <- parts, + {msg, ""} <- Integer.parse(msg_s), + {offset, ""} <- Integer.parse(off_s) do + [%{msg: msg, offset: offset, var: Enum.at(parts, 3), level: Enum.at(parts, 4)}] + else + _ -> [] + end end @spec byte_ranges_for_messages([map()], [%{var: String.t(), level: String.t()}]) :: [ diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 14eca6bd..beac340a 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -68,15 +68,54 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do @max_forecast_hour 18 @impl Oban.Worker - def perform(%Oban.Job{args: %{"forecast_hour" => fh, "run_time" => run_time_iso}}) do + def perform(%Oban.Job{args: %{"forecast_hour" => fh, "run_time" => run_time_iso}} = job) do {:ok, run_time, _} = DateTime.from_iso8601(run_time_iso) - run_chain_step(run_time, fh) + + try do + run_time + |> run_chain_step(fh) + |> rescue_chain_on_last_attempt!(run_time, fh, job) + rescue + e -> + rescue_chain_on_last_attempt!({:error, {:raised, e}}, run_time, fh, job) + reraise e, __STACKTRACE__ + end end def perform(%Oban.Job{args: args}) when args == %{} do seed_chain() end + @doc """ + Keep the forecast chain alive when a step fails permanently. Oban + will still discard the current job; we just enqueue the next + forecast hour so the rest of the chain can run. A single missing + hour beats the whole cycle going dark. + + Public for testing. + """ + @spec rescue_chain_on_last_attempt!(any(), DateTime.t(), non_neg_integer(), Oban.Job.t()) :: + any() + def rescue_chain_on_last_attempt!(:ok, _run_time, _fh, _job), do: :ok + + def rescue_chain_on_last_attempt!(failure, run_time, fh, %Oban.Job{attempt: attempt, max_attempts: max_attempts}) + when attempt >= max_attempts do + case enqueue_next_step(run_time, fh) do + {:ok, _} -> + Logger.error( + "PropagationGrid: fh=#{fh} failed permanently (#{inspect(failure)}); " <> + "enqueuing fh=#{fh + 1} to keep chain alive" + ) + + :final -> + Logger.error("PropagationGrid: fh=#{fh} (last step) failed permanently (#{inspect(failure)})") + end + + failure + end + + def rescue_chain_on_last_attempt!(failure, _run_time, _fh, _job), do: failure + @doc """ Enqueue the next chain step after a successful forecast-hour run. diff --git a/test/microwaveprop/weather/hrrr_client_test.exs b/test/microwaveprop/weather/hrrr_client_test.exs index e58ad443..ba81a0d5 100644 --- a/test/microwaveprop/weather/hrrr_client_test.exs +++ b/test/microwaveprop/weather/hrrr_client_test.exs @@ -77,6 +77,27 @@ defmodule Microwaveprop.Weather.HrrrClientTest do test "handles empty input" do assert HrrrClient.parse_idx("") == [] end + + test "skips malformed lines instead of raising" do + # NOAA S3 occasionally serves an HTML error page as the idx body + # (e.g. during brief 503 spikes). The old parser would raise + # ArgumentError on String.to_integer of a non-numeric token, + # which killed the whole PropagationGridWorker chain step. + # The tolerant parser returns the well-formed lines and drops + # the garbage. + input = """ + Service Unavailable + 1:0:d=2026032818:TMP:2 m above ground:anl: + bogus:header:line:with:no:numbers + 2:1234567:d=2026032818:DPT:2 m above ground:anl: + """ + + entries = HrrrClient.parse_idx(input) + + assert length(entries) == 2 + assert Enum.map(entries, & &1.msg) == [1, 2] + assert Enum.map(entries, & &1.var) == ["TMP", "DPT"] + end end describe "byte_ranges_for_messages/2" do diff --git a/test/microwaveprop/workers/propagation_grid_worker_test.exs b/test/microwaveprop/workers/propagation_grid_worker_test.exs index beabfb74..854d337d 100644 --- a/test/microwaveprop/workers/propagation_grid_worker_test.exs +++ b/test/microwaveprop/workers/propagation_grid_worker_test.exs @@ -68,4 +68,50 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do end) end end + + describe "chain resilience on permanent failure" do + test "enqueues fh+1 when the final attempt of a mid-chain step fails" do + Oban.Testing.with_testing_mode(:manual, fn -> + run_time = ~U[2026-04-14 16:00:00Z] + + assert {:error, :boom} = + PropagationGridWorker.rescue_chain_on_last_attempt!( + {:error, :boom}, + run_time, + 4, + %Oban.Job{attempt: 5, max_attempts: 5} + ) + + [next] = all_enqueued(worker: PropagationGridWorker) + assert next.args["forecast_hour"] == 5 + assert next.args["run_time"] == DateTime.to_iso8601(run_time) + end) + end + + test "does not enqueue on earlier attempts — Oban retries naturally" do + Oban.Testing.with_testing_mode(:manual, fn -> + PropagationGridWorker.rescue_chain_on_last_attempt!( + {:error, :boom}, + ~U[2026-04-14 16:00:00Z], + 4, + %Oban.Job{attempt: 2, max_attempts: 5} + ) + + assert [] = all_enqueued(worker: PropagationGridWorker) + end) + end + + test "does not enqueue anything past fh=18 even on final-attempt failure" do + Oban.Testing.with_testing_mode(:manual, fn -> + PropagationGridWorker.rescue_chain_on_last_attempt!( + {:error, :boom}, + ~U[2026-04-14 16:00:00Z], + 18, + %Oban.Job{attempt: 5, max_attempts: 5} + ) + + assert [] = all_enqueued(worker: PropagationGridWorker) + end) + end + end end