Stop scrape chain from dying on a single bad product

A scraped product with nil price_per_round_cents made Prices.create_snapshot
return {:error, changeset}, which blew up the {:ok, _} match in the runner,
crashed the whole perform, and after 3 retries discarded the job. Nothing
reseeds the chain once the job is discarded, so prod scrapes silently died
on Mar 17 and stayed dead until the app was restarted.

- Runner handles snapshot insert errors by logging and continuing instead
  of raising.
- ScrapeJob wraps each Runner.run in try/rescue so one retailer crashing
  doesn't take down the whole cycle.
- ScrapeJob moves schedule_next into an after block so the chain is always
  reseeded, even if the scrape loop crashes.

lib/ammoprices/scraping/runner.ex

@@ -5,6 +5,8 @@ defmodule Ammoprices.Scraping.Runner do
   alias Ammoprices.Prices
   alias Ammoprices.Scraping.HttpClient
 
+  require Logger
+
   def run(scraper, caliber) do
     retailer = Catalog.get_retailer_by_slug!(scraper.retailer_slug())
 
@@ -49,8 +51,20 @@ defmodule Ammoprices.Scraping.Runner do
   defp process_results(parsed, retailer, caliber) do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
-    results =
+    results = Enum.map(parsed, &process_product(&1, retailer, caliber, now))
-      Enum.map(parsed, fn product_data ->
+
+    successful = Enum.count(results, &(&1 == :ok))
+
+    if successful > 0 do
+      Catalog.mark_unseen_products_out_of_stock(retailer.id, caliber.id, now)
+    end
+
+    Catalog.update_retailer(retailer, %{last_scraped_at: now})
+
+    {:ok, %{products_count: successful, snapshots_count: successful}}
+  end
+
+  defp process_product(product_data, retailer, caliber, now) do
     product_attrs = %{
       title: product_data.title,
       url: product_data.url,
@@ -64,31 +78,26 @@ defmodule Ammoprices.Scraping.Runner do
       last_seen_at: now
     }
 
-        case Catalog.upsert_product(retailer.id, caliber.id, product_attrs) do
+    with {:ok, product} <- Catalog.upsert_product(retailer.id, caliber.id, product_attrs),
-          {:ok, product} ->
+         {:ok, _snapshot} <- create_snapshot(product, product_data, now) do
-            snapshot_attrs = %{
+      :ok
+    else
+      {:error, %Ecto.Changeset{} = changeset} ->
+        Logger.warning(
+          "Scrape insert failed for #{retailer.slug} #{product_data.url}: " <>
+            inspect(changeset.errors)
+        )
+
+        :error
+    end
+  end
+
+  defp create_snapshot(product, product_data, now) do
+    Prices.create_snapshot(product.id, %{
       price_cents: product_data.price_cents,
       price_per_round_cents: product_data.price_per_round_cents,
       in_stock: product_data.in_stock,
       recorded_at: now
-            }
+    })
-
-            {:ok, _snapshot} = Prices.create_snapshot(product.id, snapshot_attrs)
-            :ok
-
-          {:error, _changeset} ->
-            :error
-        end
-      end)
-
-    successful = Enum.count(results, &(&1 == :ok))
-
-    if successful > 0 do
-      Catalog.mark_unseen_products_out_of_stock(retailer.id, caliber.id, now)
-    end
-
-    Catalog.update_retailer(retailer, %{last_scraped_at: now})
-
-    {:ok, %{products_count: successful, snapshots_count: successful}}
   end
 end

lib/ammoprices/scraping/scrape_job.ex

@@ -8,6 +8,8 @@ defmodule Ammoprices.Scraping.ScrapeJob do
   alias Ammoprices.Catalog
   alias Ammoprices.Scraping.Runner
 
+  require Logger
+
   @scrapers [
     Ammoprices.Scraping.Retailers.LuckyGunner,
     Ammoprices.Scraping.Retailers.SgAmmo,
@@ -27,16 +29,30 @@ defmodule Ammoprices.Scraping.ScrapeJob do
 
     for scraper <- @scrapers, caliber <- calibers do
       scrape_delay()
-      Runner.run(scraper, caliber)
+      safe_run(scraper, caliber)
     end
 
     prune_stale_products()
 
     Phoenix.PubSub.broadcast(Ammoprices.PubSub, "prices:updated", {:prices_updated, %{}})
 
-    schedule_next()
-
     :ok
+  after
+    # Always reseed the chain, even if something above crashed. Without this,
+    # a single unhandled error kills the chain until the app is restarted.
+    schedule_next()
+  end
+
+  defp safe_run(scraper, caliber) do
+    Runner.run(scraper, caliber)
+  rescue
+    error ->
+      Logger.error(
+        "Scrape crashed for #{inspect(scraper)} / #{caliber.slug}: " <>
+          Exception.format(:error, error, __STACKTRACE__)
+      )
+
+      {:error, :crashed}
   end
 
   defp prune_stale_products do

test/ammoprices/scraping/runner_test.exs

@@ -88,6 +88,55 @@ defmodule Ammoprices.Scraping.RunnerTest do
       assert Ammoprices.Repo.get!(Product, other.id).in_stock == true
     end
 
+    test "continues processing when a single product's snapshot fails validation",
+         %{retailer: retailer, caliber: caliber} do
+      defmodule BrokenScraper do
+        @moduledoc false
+        @behaviour Ammoprices.Scraping.Scraper
+
+        def retailer_slug, do: "lucky-gunner"
+        def category_url(_), do: "/any"
+
+        def parse_products(_body) do
+          [
+            %{
+              title: "Good 9mm",
+              url: "/products/good",
+              brand: "Good",
+              grain_weight: 115,
+              round_count: 50,
+              casing: "brass",
+              in_stock: true,
+              subsonic: false,
+              price_cents: 1599,
+              price_per_round_cents: 32
+            },
+            %{
+              title: "Bad 9mm (no round count)",
+              url: "/products/bad",
+              brand: "Bad",
+              grain_weight: 115,
+              round_count: nil,
+              casing: "brass",
+              in_stock: false,
+              subsonic: false,
+              price_cents: 2167,
+              price_per_round_cents: nil
+            }
+          ]
+        end
+      end
+
+      assert {:ok, stats} = Runner.run(BrokenScraper, caliber)
+      assert stats.products_count == 1
+      assert stats.snapshots_count == 1
+
+      good =
+        Ammoprices.Repo.get_by!(Product, retailer_id: retailer.id, url: "/products/good")
+
+      assert good.in_stock == true
+    end
+
     test "does not prune when scrape returns zero products", %{retailer: retailer, caliber: caliber} do
       Req.Test.stub(HttpClient, fn conn ->
         Req.Test.html(conn, "<html><body>no products here</body></html>")

test/ammoprices/scraping/scrape_job_test.exs

@@ -4,6 +4,7 @@ defmodule Ammoprices.Scraping.ScrapeJobTest do
 
   import Ammoprices.Fixtures
 
+  alias Ammoprices.Scraping.HttpClient
   alias Ammoprices.Scraping.ScrapeJob
 
   @fixture_html File.read!("test/fixtures/lucky_gunner/9mm.html")
@@ -44,7 +45,7 @@ defmodule Ammoprices.Scraping.ScrapeJobTest do
 
     caliber_fixture(%{name: "9mm Luger", slug: "9mm-luger", category: "handgun", aliases: ["9mm", "9x19"]})
 
-    Req.Test.stub(Ammoprices.Scraping.HttpClient, fn conn ->
+    Req.Test.stub(HttpClient, fn conn ->
       Req.Test.html(conn, @fixture_html)
     end)
 
@@ -56,6 +57,27 @@ defmodule Ammoprices.Scraping.ScrapeJobTest do
       assert :ok = perform_job(ScrapeJob, %{})
     end
 
+    test "survives HttpClient crashes and still schedules the next job" do
+      import Ecto.Query
+
+      Req.Test.stub(HttpClient, fn _conn ->
+        raise "simulated network crash"
+      end)
+
+      assert :ok = perform_job(ScrapeJob, %{})
+
+      scheduled_count =
+        Ammoprices.Repo.aggregate(
+          from(j in "oban_jobs",
+            where: j.worker == "Ammoprices.Scraping.ScrapeJob" and j.state == "scheduled"
+          ),
+          :count,
+          :id
+        )
+
+      assert scheduled_count >= 1
+    end
+
     test "hard-deletes products not seen in 30+ days after the scrape cycle" do
       retailer = Ammoprices.Catalog.get_retailer_by_slug!("lucky-gunner")
       caliber = Ammoprices.Catalog.get_caliber_by_slug!("9mm-luger")