Refactor: extract magic numbers, add stacktraces, deduplicate code, add tests

- Extract size thresholds in rules.ex into @size_60_gib/@size_40_gib/@size_25_gib module attributes
- Add __STACKTRACE__ to 3 bare rescue blocks in crf_search.ex for production debugging
- Log dashboard query timeouts in media.ex (previously swallowed silently)
- Consolidate duplicate Repo.insert/stale-error handlers in video_upsert.ex into shared do_insert/2
- Extract safe_to_existing_atom/1 helper from duplicate filter patterns in media.ex
- Add crash_reason to Logger metadata config
- Add property-based tests for Rules (vmaf_target monotonicity, build_args deduplication)
- Expand encoder Broadway producer tests (recovery threshold, dispatch_available)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: mjc

config/config.exs

@@ -117,7 +117,8 @@ config :logger, :console,
     :state,
     :exit_code,
     :result,
-    :video_info
+    :video_info,
+    :crash_reason
   ]
 
 # Use Jason for JSON parsing in Phoenix

lib/reencodarr/ab_av1/crf_search.ex

@@ -309,7 +309,10 @@ defmodule Reencodarr.AbAv1.CrfSearch do
       {:noreply, %{state | partial_line_buffer: "", output_buffer: new_output_buffer}}
     rescue
       e ->
-        Logger.error("CrfSearch: Error processing line '#{full_line}': #{Exception.message(e)}")
+        Logger.error("CrfSearch: Error processing line '#{full_line}': #{Exception.message(e)}",
+          crash_reason: {e, __STACKTRACE__}
+        )
+
         {:noreply, %{state | partial_line_buffer: "", output_buffer: [full_line | output_buffer]}}
     end
   end
@@ -352,7 +355,9 @@ defmodule Reencodarr.AbAv1.CrfSearch do
       end
     rescue
       e ->
-        Logger.error("CrfSearch: Error in exit_status=0 handler: #{Exception.message(e)}")
+        Logger.error("CrfSearch: Error in exit_status=0 handler: #{Exception.message(e)}",
+          crash_reason: {e, __STACKTRACE__}
+        )
     end
 
     perform_crf_search_cleanup(state)
@@ -376,7 +381,10 @@ defmodule Reencodarr.AbAv1.CrfSearch do
       handle_crf_search_failure(video, target_vmaf, exit_code, command_line, full_output, state)
     rescue
       e ->
-        Logger.error("CrfSearch: Error in exit_status handler: #{Exception.message(e)}")
+        Logger.error("CrfSearch: Error in exit_status handler: #{Exception.message(e)}",
+          crash_reason: {e, __STACKTRACE__}
+        )
+
         perform_crf_search_cleanup(state)
     end
   end

lib/reencodarr/media.ex

@@ -1130,18 +1130,22 @@ defmodule Reencodarr.Media do
 
   defp apply_sort(query, field, dir), do: from(v in query, order_by: [{^dir, field(v, ^field)}])
 
-  defp maybe_filter_state(query, nil), do: query
-  defp maybe_filter_state(query, ""), do: query
+  defp safe_to_existing_atom(value) when is_atom(value), do: {:ok, value}
 
-  defp maybe_filter_state(query, state) when is_binary(state) do
-    state_atom = String.to_existing_atom(state)
-    from(v in query, where: v.state == ^state_atom)
+  defp safe_to_existing_atom(value) when is_binary(value) do
+    {:ok, String.to_existing_atom(value)}
   rescue
-    ArgumentError -> query
+    ArgumentError -> :error
   end
 
-  defp maybe_filter_state(query, state) when is_atom(state) do
-    from(v in query, where: v.state == ^state)
+  defp maybe_filter_state(query, nil), do: query
+  defp maybe_filter_state(query, ""), do: query
+
+  defp maybe_filter_state(query, state) do
+    case safe_to_existing_atom(state) do
+      {:ok, atom} -> from(v in query, where: v.state == ^atom)
+      :error -> query
+    end
   end
 
   defp maybe_filter_search(query, nil), do: query
@@ -1154,16 +1158,13 @@ defmodule Reencodarr.Media do
 
   defp maybe_filter_service_type(query, nil), do: query
 
-  defp maybe_filter_service_type(query, service) when is_binary(service) do
-    atom = String.to_existing_atom(service)
-    from(v in query, where: v.service_type == ^atom)
-  rescue
-    ArgumentError -> query
+  defp maybe_filter_service_type(query, service) do
+    case safe_to_existing_atom(service) do
+      {:ok, atom} -> from(v in query, where: v.service_type == ^atom)
+      :error -> query
+    end
   end
 
-  defp maybe_filter_service_type(query, service) when is_atom(service),
-    do: from(v in query, where: v.service_type == ^service)
-
   defp maybe_filter_hdr(query, nil), do: query
   defp maybe_filter_hdr(query, true), do: from(v in query, where: not is_nil(v.hdr))
   defp maybe_filter_hdr(query, false), do: from(v in query, where: is_nil(v.hdr))
@@ -1313,10 +1314,23 @@ defmodule Reencodarr.Media do
 
     Map.merge(video_stats || get_default_video_stats(), vmaf_stats || get_default_vmaf_stats())
   rescue
-    DBConnection.ConnectionError -> get_default_stats()
+    e in DBConnection.ConnectionError ->
+      Logger.warning(
+        "Dashboard stats query failed with connection error: #{Exception.message(e)}"
+      )
+
+      get_default_stats()
   catch
-    :exit, {:timeout, _} -> get_default_stats()
-    :exit, {%DBConnection.ConnectionError{}, _} -> get_default_stats()
+    :exit, {:timeout, _} ->
+      Logger.warning("Dashboard stats query timed out after #{timeout}ms")
+      get_default_stats()
+
+    :exit, {%DBConnection.ConnectionError{} = e, _} ->
+      Logger.warning(
+        "Dashboard stats query failed with connection error: #{Exception.message(e)}"
+      )
+
+      get_default_stats()
   end
 
   defp get_default_video_stats do

lib/reencodarr/media/video_upsert.ex

@@ -242,18 +242,7 @@ defmodule Reencodarr.Media.VideoUpsert do
           {:replace_all_except, [atom()]} | Ecto.Query.t()
         ) :: {:ok, Video.t()} | {:error, Ecto.Changeset.t()}
   defp perform_video_upsert(attrs, on_conflict_query) do
-    result =
-      %Video{}
-      |> Video.changeset(attrs)
-      |> Repo.insert(
-        on_conflict: on_conflict_query,
-        conflict_target: :path,
-        stale_error_field: :updated_at,
-        returning: true
-      )
-
-    # Return the result directly, don't wrap in transaction
-    result
+    do_insert(attrs, on_conflict_query)
   end
 
   @spec perform_single_upsert_in_batch(%{String.t() => any()}) ::
@@ -262,22 +251,12 @@ defmodule Reencodarr.Media.VideoUpsert do
     conflict_except = determine_conflict_except_fields(attrs)
     on_conflict_query = build_on_conflict_query(attrs, conflict_except)
 
-    result =
-      %Video{}
-      |> Video.changeset(attrs)
-      |> Repo.insert(
-        on_conflict: on_conflict_query,
-        conflict_target: :path,
-        stale_error_field: :updated_at,
-        returning: true
-      )
-
-    case result do
+    case do_insert(attrs, on_conflict_query) do
       {:ok, video} ->
         handle_successful_upsert(video)
 
       {:error, %Ecto.Changeset{errors: [updated_at: {"is stale", _}]} = changeset} ->
-        handle_stale_update_error_in_batch(changeset, attrs)
+        handle_stale_update_error(changeset, attrs)
 
       {:error, changeset} ->
         path = Map.get(attrs, "path", "unknown")
@@ -286,23 +265,6 @@ defmodule Reencodarr.Media.VideoUpsert do
     end
   end
 
-  @spec handle_stale_update_error_in_batch(
-          Ecto.Changeset.t(),
-          %{String.t() => any()}
-        ) :: {:ok, Video.t()} | {:error, Ecto.Changeset.t()}
-  defp handle_stale_update_error_in_batch(changeset, attrs) do
-    # This is expected when dateAdded is not newer than updated_at - treat as success (skip)
-    path = Map.get(attrs, "path")
-    Logger.debug("Skipping update for #{path} in batch - dateAdded not newer than updated_at")
-
-    # Return the existing video instead of an error
-    case Repo.get_by(Video, path: path) do
-      # Shouldn't happen, but handle gracefully
-      nil -> {:error, changeset}
-      existing_video -> {:ok, existing_video}
-    end
-  end
-
   @spec handle_upsert_result(
           {:ok, Video.t()} | {:error, Ecto.Changeset.t()} | {:error, any()},
           %{String.t() => any()}
@@ -334,14 +296,22 @@ defmodule Reencodarr.Media.VideoUpsert do
     {:ok, video}
   end
 
+  defp do_insert(attrs, on_conflict_query) do
+    %Video{}
+    |> Video.changeset(attrs)
+    |> Repo.insert(
+      on_conflict: on_conflict_query,
+      conflict_target: :path,
+      stale_error_field: :updated_at,
+      returning: true
+    )
+  end
+
   defp handle_stale_update_error(changeset, attrs) do
-    # This is expected when dateAdded is not newer than updated_at - treat as success (skip)
     path = Map.get(attrs, "path")
-    Logger.debug("Skipping update for #{path} - dateAdded not newer than updated_at")
+    Logger.debug("Skipping update for #{path} — dateAdded not newer than updated_at")
 
-    # Return the existing video instead of an error
     case Repo.get_by(Video, path: path) do
-      # Shouldn't happen, but handle gracefully
       nil -> {:error, changeset}
       existing_video -> {:ok, existing_video}
     end

lib/reencodarr/rules.ex

@@ -9,6 +9,11 @@ defmodule Reencodarr.Rules do
   alias Reencodarr.Core.Parsers
   alias Reencodarr.Media
 
+  # Size thresholds for VMAF target selection (in bytes)
+  @size_60_gib 60 * 1024 * 1024 * 1024
+  @size_40_gib 40 * 1024 * 1024 * 1024
+  @size_25_gib 25 * 1024 * 1024 * 1024
+
   @doc """
   Build arguments for ab-av1 commands.
 
@@ -404,9 +409,9 @@ defmodule Reencodarr.Rules do
       95
   """
   @spec vmaf_target(map()) :: integer()
-  def vmaf_target(%{size: size}) when is_integer(size) and size > 60 * 1024 * 1024 * 1024, do: 91
-  def vmaf_target(%{size: size}) when is_integer(size) and size > 40 * 1024 * 1024 * 1024, do: 92
-  def vmaf_target(%{size: size}) when is_integer(size) and size > 25 * 1024 * 1024 * 1024, do: 94
+  def vmaf_target(%{size: size}) when is_integer(size) and size > @size_60_gib, do: 91
+  def vmaf_target(%{size: size}) when is_integer(size) and size > @size_40_gib, do: 92
+  def vmaf_target(%{size: size}) when is_integer(size) and size > @size_25_gib, do: 94
   def vmaf_target(_video), do: 95
 
   @doc """

test/reencodarr/encoder/broadway/producer_test.exs

@@ -186,4 +186,73 @@ defmodule Reencodarr.Encoder.Broadway.ProducerTest do
       refute Producer.should_attempt_recovery?(1801)
     end
   end
+
+  describe "recovery threshold boundary" do
+    alias Reencodarr.Encoder.Broadway.Producer
+
+    test "exactly at threshold triggers recovery" do
+      assert Producer.should_attempt_recovery?(900)
+    end
+
+    test "one below threshold does not trigger" do
+      refute Producer.should_attempt_recovery?(899)
+    end
+
+    test "one above threshold does not trigger" do
+      refute Producer.should_attempt_recovery?(901)
+    end
+
+    test "zero never triggers" do
+      refute Producer.should_attempt_recovery?(0)
+    end
+
+    test "negative values never trigger" do
+      refute Producer.should_attempt_recovery?(-1)
+      refute Producer.should_attempt_recovery?(-900)
+    end
+  end
+
+  describe "update_consecutive_count/2 exhaustive" do
+    alias Reencodarr.Encoder.Broadway.Producer
+
+    test "all non-timeout statuses reset counter" do
+      for status <- [:available, :busy] do
+        assert Producer.update_consecutive_count(999, status) == 0,
+               "Expected #{status} to reset counter"
+      end
+    end
+
+    test "timeout from zero increments to one" do
+      assert Producer.update_consecutive_count(0, :timeout) == 1
+    end
+
+    test "timeout accumulates linearly" do
+      result =
+        Enum.reduce(1..10, 0, fn _i, acc ->
+          Producer.update_consecutive_count(acc, :timeout)
+        end)
+
+      assert result == 10
+    end
+
+    test "reset after accumulation returns to zero" do
+      count =
+        0
+        |> Producer.update_consecutive_count(:timeout)
+        |> Producer.update_consecutive_count(:timeout)
+        |> Producer.update_consecutive_count(:timeout)
+        |> Producer.update_consecutive_count(:available)
+
+      assert count == 0
+    end
+  end
+
+  describe "dispatch_available/0" do
+    alias Reencodarr.Encoder.Broadway.Producer
+
+    test "returns error when producer is not running" do
+      # Producer is not started in test environment
+      assert {:error, :producer_not_found} = Producer.dispatch_available()
+    end
+  end
 end