From 81cb735d6ec1109e11dfe8d41c2253867ef2a2c5 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Tue, 31 Mar 2026 08:12:10 -0500
Subject: [PATCH] Skip corrupt GRIB2 messages instead of failing entire fetch
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Both extract_points and extract_grid now skip messages with missing
data sections instead of halting. Pressure product fetch is now
optional for both grid and single-point modes — if pressure data
has corrupt messages, surface-only profiles are still stored. Fixes
historical HRRR backfill failing on old data with incomplete files.
---
 lib/microwaveprop/weather/grib2/extractor.ex | 24 ++++++++-------
 lib/microwaveprop/weather/hrrr_client.ex     | 31 +++++++++++++++++---
 2 files changed, 40 insertions(+), 15 deletions(-)

diff --git a/lib/microwaveprop/weather/grib2/extractor.ex b/lib/microwaveprop/weather/grib2/extractor.ex
index c4df96d6..6cdefe86 100644
--- a/lib/microwaveprop/weather/grib2/extractor.ex
+++ b/lib/microwaveprop/weather/grib2/extractor.ex
@@ -14,16 +14,17 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
   def extract_points(binary, lat, lon) do
     messages = split_messages(binary)
 
-    results =
-      Enum.reduce_while(messages, {:ok, %{}}, fn msg, {:ok, acc} ->
+    Enum.reduce(messages, {:ok, %{}}, fn
+      msg, {:ok, acc} ->
         case extract_single(msg, lat, lon) do
-          {:ok, key, value} -> {:cont, {:ok, Map.put(acc, key, value)}}
-          {:error, :outside_grid} -> {:halt, {:error, :outside_grid}}
-          {:error, reason} -> {:halt, {:error, reason}}
+          {:ok, key, value} -> {:ok, Map.put(acc, key, value)}
+          {:error, :outside_grid} -> {:error, :outside_grid}
+          {:error, _reason} -> {:ok, acc}
         end
-      end)
 
-    results
+      _msg, error ->
+        error
+    end)
   end
 
   @doc """
@@ -36,7 +37,7 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
     messages = split_messages(binary)
 
     result =
-      Enum.reduce_while(messages, {:ok, init_grid(points)}, fn msg, {:ok, acc} ->
+      Enum.reduce(messages, {:ok, init_grid(points)}, fn msg, {:ok, acc} ->
         case extract_single_grid(msg, points) do
           {:ok, point_values} ->
             merged =
@@ -44,10 +45,11 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
                 Map.update!(grid, point, &Map.put(&1, key, value))
               end)
 
-            {:cont, {:ok, merged}}
+            {:ok, merged}
 
-          {:error, reason} ->
-            {:halt, {:error, reason}}
+          {:error, _reason} ->
+            # Skip messages with missing/corrupt data sections
+            {:ok, acc}
         end
       end)
 
diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex
index 41cad034..2a0a74be 100644
--- a/lib/microwaveprop/weather/hrrr_client.ex
+++ b/lib/microwaveprop/weather/hrrr_client.ex
@@ -30,8 +30,18 @@ defmodule Microwaveprop.Weather.HrrrClient do
     date = DateTime.to_date(hour_dt)
     hour = hour_dt.hour
 
-    with {:ok, sfc_grid} <- fetch_product_grid(date, hour, :surface, points),
-         {:ok, prs_grid} <- maybe_fetch_pressure_grid(date, hour, points, opts) do
+    with {:ok, sfc_grid} <- fetch_product_grid(date, hour, :surface, points) do
+      # Pressure is optional — if it fails, continue with surface-only data
+      prs_grid =
+        case maybe_fetch_pressure_grid(date, hour, points, opts) do
+          {:ok, grid} ->
+            grid
+
+          {:error, reason} ->
+            Logger.warning("HRRR pressure fetch failed (continuing with surface only): #{inspect(reason)}")
+            %{}
+        end
+
       merged = merge_grid_data(sfc_grid, prs_grid)
 
       profiles =
@@ -49,8 +59,21 @@ defmodule Microwaveprop.Weather.HrrrClient do
     date = DateTime.to_date(hour_dt)
     hour = hour_dt.hour
 
-    with {:ok, sfc_data} <- fetch_product(date, hour, :surface, lat, lon),
-         {:ok, prs_data} <- fetch_product(date, hour, :pressure, lat, lon) do
+    with {:ok, sfc_data} <- fetch_product(date, hour, :surface, lat, lon) do
+      # Pressure is optional — old HRRR data may have corrupt messages
+      prs_data =
+        case fetch_product(date, hour, :pressure, lat, lon) do
+          {:ok, data} ->
+            data
+
+          {:error, reason} ->
+            Logger.warning(
+              "HRRR pressure fetch failed for #{lat},#{lon} (continuing with surface only): #{inspect(reason)}"
+            )
+
+            %{}
+        end
+
       merged = Map.merge(sfc_data, prs_data)
       result = build_profile(merged)
       {:ok, Map.put(result, :run_time, hour_dt)}