Make Hex indexing retries testable

lib/exograph/hex/broadway_pipeline.ex

@@ -154,27 +154,42 @@ defmodule Exograph.Hex.BroadwayPipeline do
   defp index_with_retries(entry, index, opts) do
     retry_count = Keyword.get(opts, :retry_count, 3)
     retry_sleep = Keyword.get(opts, :retry_sleep, 1_000)
+    index_fun = Keyword.get(opts, :index_fun, &Exograph.Hex.Corpus.index_entry/3)
+    sleep_fun = Keyword.get(opts, :sleep_fun, &Process.sleep/1)
 
     Enum.reduce_while(0..retry_count, nil, fn attempt, _last_result ->
-      result = Exograph.Hex.Corpus.index_entry(entry, index, opts)
+      result = safe_index_entry(index_fun, entry, index, opts)
 
       if transient_error?(result) and attempt < retry_count do
-        Process.sleep(retry_sleep * (attempt + 1))
+        sleep_fun.(retry_sleep * (attempt + 1))
         {:cont, result}
       else
         {:halt, result}
       end
     end)
   end
 
-  defp transient_error?({:error, reason}) do
-    text = inspect(reason)
+  defp safe_index_entry(index_fun, entry, index, opts) do
+    index_fun.(entry, index, opts)
+  rescue
+    error -> {:error, {error.__struct__, Exception.message(error)}}
+  catch
+    :exit, reason -> {:error, {:exit, reason}}
+    kind, reason -> {:error, {kind, reason}}
+  end
 
-    String.contains?(text, [
+  defp transient_error?({:error, reason}) do
+    reason
+    |> inspect()
+    |> String.downcase()
+    |> String.contains?([
       "queue_timeout",
       "connection not available",
       "request was dropped from queue",
-      "timed out"
+      "timed out",
+      "time out",
+      "timeout",
+      "transport_error"
     ])
   end
 

test/exograph/hex/broadway_pipeline_test.exs

@@ -4,6 +4,52 @@ defmodule Exograph.Hex.BroadwayPipelineTest do
   alias Broadway.Message
   alias Exograph.Hex.BroadwayPipeline
 
+  test "retries transient time out results" do
+    {:ok, attempts} = Agent.start_link(fn -> 0 end)
+
+    index_fun = fn _entry, _index, _opts ->
+      attempt = Agent.get_and_update(attempts, &{&1 + 1, &1 + 1})
+      if attempt < 3, do: {:error, "commit time out"}, else: :ok
+    end
+
+    sleep_fun = fn _ms -> :ok end
+
+    {result, _elapsed_ms} =
+      BroadwayPipeline.index([%{name: "pkg", version: "1.0.0"}],
+        name: :"test_broadway_retry_#{System.unique_integer([:positive])}",
+        index_fun: index_fun,
+        sleep_fun: sleep_fun,
+        retry_count: 3,
+        retry_sleep: 0
+      )
+
+    assert result.ok == 1
+    assert result.error == 0
+    assert Agent.get(attempts, & &1) == 3
+  end
+
+  test "retries transient raised exceptions" do
+    {:ok, attempts} = Agent.start_link(fn -> 0 end)
+
+    index_fun = fn _entry, _index, _opts ->
+      attempt = Agent.get_and_update(attempts, &{&1 + 1, &1 + 1})
+      if attempt < 2, do: raise("request was dropped from queue"), else: :ok
+    end
+
+    {result, _elapsed_ms} =
+      BroadwayPipeline.index([%{name: "pkg", version: "1.0.0"}],
+        name: :"test_broadway_exception_retry_#{System.unique_integer([:positive])}",
+        index_fun: index_fun,
+        sleep_fun: fn _ms -> :ok end,
+        retry_count: 2,
+        retry_sleep: 0
+      )
+
+    assert result.ok == 1
+    assert result.error == 0
+    assert Agent.get(attempts, & &1) == 2
+  end
+
   test "ack preserves failed message reasons" do
     owner = self()