Add encoder health check and refactor error handling

lib/reencodarr/ab_av1/encode.ex

@@ -10,6 +10,7 @@ defmodule Reencodarr.AbAv1.Encode do
 
   alias Reencodarr.AbAv1.Helper
   alias Reencodarr.AbAv1.ProgressParser
+  alias Reencodarr.Core.Retry
   alias Reencodarr.Dashboard.Events
   alias Reencodarr.Encoder.Broadway.Producer
   alias Reencodarr.{Media, PostProcessor}
@@ -120,14 +121,8 @@ defmodule Reencodarr.AbAv1.Encode do
         {port, {:exit_status, exit_code}},
         %{port: port, vmaf: vmaf, output_file: output_file} = state
       ) do
-    result =
-      case exit_code do
-        0 -> {:ok, :success}
-        _ -> {:error, exit_code}
-      end
-
     # Broadcast encoding completion to Dashboard Events
-    pubsub_result = if result == {:ok, :success}, do: :success, else: {:error, exit_code}
+    pubsub_result = if exit_code == 0, do: :success, else: {:error, exit_code}
 
     Events.broadcast_event(:encoding_completed, %{
       video_id: vmaf.video.id,
@@ -137,91 +132,17 @@ defmodule Reencodarr.AbAv1.Encode do
     # Notify the Broadway producer that encoding is now available
     Producer.dispatch_available()
 
-    # Attempt to notify success/failure, but don't crash if DB is busy — retry later
-    notify_result =
-      try do
-        if result == {:ok, :success} do
-          notify_encoder_success(vmaf.video, output_file)
-        else
-          notify_encoder_failure(vmaf.video, exit_code)
-        end
-
-        :ok
-      rescue
-        e in Exqlite.Error ->
-          Logger.warning(
-            "Encode: Database busy while handling exit_status, scheduling retry: #{inspect(e)}"
-          )
-
-          # Retry after short backoff. Keep current state until retry completes.
-          Process.send_after(self(), {:encoding_exit_retry, exit_code, vmaf, output_file}, 200)
-          :retry_scheduled
+    # Handle success/failure with automatic retry on DB busy
+    Retry.retry_on_db_busy(fn ->
+      if exit_code == 0 do
+        notify_encoder_success(vmaf.video, output_file)
+      else
+        notify_encoder_failure(vmaf.video, exit_code)
       end
+    end)
 
-    case notify_result do
-      :ok ->
-        new_state = %{
-          state
-          | port: :none,
-            video: :none,
-            vmaf: :none,
-            output_file: nil,
-            partial_line_buffer: "",
-            last_progress: nil
-        }
-
-        {:noreply, new_state}
-
-      :retry_scheduled ->
-        # Keep state intact so retry handler has data available
-        {:noreply, state}
-    end
-  end
-
-  @impl true
-  def handle_info({:encoding_exit_retry, exit_code, vmaf, output_file}, state) do
-    Logger.info("Encode: retrying exit_status handling for VMAF #{vmaf.id}")
-
-    retry_result =
-      try do
-        if exit_code == 0 do
-          notify_encoder_success(vmaf.video, output_file)
-        else
-          notify_encoder_failure(vmaf.video, exit_code)
-        end
-
-        :ok
-      rescue
-        e in Exqlite.Error ->
-          Logger.warning(
-            "Encode: retry failed due to DB busy: #{inspect(e)}, scheduling another retry"
-          )
-
-          Process.send_after(self(), {:encoding_exit_retry, exit_code, vmaf, output_file}, 500)
-          :retry_scheduled
-      end
-
-    case retry_result do
-      :ok ->
-        # Clean up state now that notification succeeded
-        cleared_state = %{
-          state
-          | port: :none,
-            video: :none,
-            vmaf: :none,
-            output_file: nil,
-            partial_line_buffer: "",
-            last_progress: nil
-        }
-
-        # Ensure Broadway knows encoder is available
-        Producer.dispatch_available()
-
-        {:noreply, cleared_state}
-
-      :retry_scheduled ->
-        {:noreply, state}
-    end
+    # Always clear state after handling exit (retry succeeded or max attempts reached)
+    {:noreply, clear_state(state)}
   end
 
   # Catch-all for any other port messages
@@ -292,10 +213,18 @@ defmodule Reencodarr.AbAv1.Encode do
 
         port = Helper.open_port(args)
 
-        # Broadcast encoding started to Dashboard Events
+        # Get OS PID from port for health monitoring
+        os_pid =
+          case Port.info(port, :os_pid) do
+            {:os_pid, pid} -> pid
+            _ -> nil
+          end
+
+        # Broadcast encoding started to Dashboard Events (includes OS PID for health check)
         Events.broadcast_event(:encoding_started, %{
           video_id: vmaf.video.id,
-          filename: Path.basename(vmaf.video.path)
+          filename: Path.basename(vmaf.video.path),
+          os_pid: os_pid
         })
 
         # Set up a periodic timer to check if we're still alive and potentially emit progress
@@ -349,13 +278,25 @@ defmodule Reencodarr.AbAv1.Encode do
     def build_encode_args_for_test(vmaf), do: build_encode_args(vmaf)
   end
 
+  defp clear_state(state) do
+    %{
+      state
+      | port: :none,
+        video: :none,
+        vmaf: :none,
+        output_file: nil,
+        partial_line_buffer: "",
+        last_progress: nil
+    }
+  end
+
   defp notify_encoder_success(video, output_file) do
     # Use PostProcessor for cleanup work
     PostProcessor.process_encoding_success(video, output_file)
   end
 
-  defp notify_encoder_failure(_video, _exit_code) do
-    # Failure handling complete - no additional work needed
+  defp notify_encoder_failure(video, exit_code) do
+    PostProcessor.process_encoding_failure(video, exit_code)
   end
 
   # Extract progress data from a line and store it for periodic updates

lib/reencodarr/application.ex

@@ -55,7 +55,8 @@ defmodule Reencodarr.Application do
     broadway_workers = [
       Reencodarr.AbAv1,
       Reencodarr.CrfSearcher.Supervisor,
-      Reencodarr.Encoder.Supervisor
+      Reencodarr.Encoder.Supervisor,
+      Reencodarr.Encoder.HealthCheck
     ]
 
     # Only start Analyzer GenStage in non-test environments

lib/reencodarr/encoder/health_check.ex

@@ -0,0 +1,176 @@
+defmodule Reencodarr.Encoder.HealthCheck do
+  @moduledoc """
+  Monitors the encoder pipeline for stuck states via PubSub events.
+
+  Subscribes to encoding events and detects when ab-av1/ffmpeg hangs silently
+  (no progress events for extended periods). Automatically kills the stuck
+  process after 1 hour of no progress.
+  """
+
+  use GenServer
+  require Logger
+
+  alias Reencodarr.Dashboard.Events
+
+  # Check every 60 seconds
+  @check_interval 60_000
+  # Warn at 30 minutes
+  @warn_threshold 30 * 60_000
+  # Kill at 1 hour
+  @kill_threshold 60 * 60_000
+
+  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+
+  @impl true
+  def init(_opts) do
+    # Subscribe to encoding events instead of polling state
+    Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+    schedule_check()
+
+    {:ok,
+     %{
+       encoding: false,
+       video_id: nil,
+       video_path: nil,
+       last_progress_time: nil,
+       last_progress_percent: nil,
+       warned: false,
+       os_pid: nil
+     }}
+  end
+
+  # Handle encoding lifecycle events
+  @impl true
+  def handle_info({:encoding_started, data}, state) do
+    now = System.monotonic_time(:millisecond)
+
+    {:noreply,
+     %{
+       state
+       | encoding: true,
+         video_id: data[:video_id],
+         video_path: data[:filename],
+         last_progress_time: now,
+         last_progress_percent: nil,
+         warned: false,
+         os_pid: data[:os_pid]
+     }}
+  end
+
+  @impl true
+  def handle_info({:encoding_progress, data}, state) do
+    now = System.monotonic_time(:millisecond)
+    percent = data[:percent]
+
+    {:noreply,
+     %{
+       state
+       | last_progress_time: now,
+         last_progress_percent: percent,
+         warned: false
+     }}
+  end
+
+  @impl true
+  def handle_info({:encoding_completed, _data}, state) do
+    {:noreply,
+     %{
+       state
+       | encoding: false,
+         video_id: nil,
+         video_path: nil,
+         last_progress_time: nil,
+         last_progress_percent: nil,
+         warned: false,
+         os_pid: nil
+     }}
+  end
+
+  # Periodic health check
+  @impl true
+  def handle_info(:health_check, state) do
+    new_state = perform_health_check(state)
+    schedule_check()
+    {:noreply, new_state}
+  end
+
+  # Ignore other PubSub events
+  @impl true
+  def handle_info(_msg, state), do: {:noreply, state}
+
+  defp perform_health_check(%{encoding: false} = state), do: state
+
+  defp perform_health_check(%{encoding: true, last_progress_time: nil} = state) do
+    # Encoding but no progress time set - initialize it
+    %{state | last_progress_time: System.monotonic_time(:millisecond)}
+  end
+
+  defp perform_health_check(%{encoding: true} = state) do
+    now = System.monotonic_time(:millisecond)
+    elapsed = now - state.last_progress_time
+
+    cond do
+      # Stuck for 1 hour - kill the process
+      elapsed > @kill_threshold ->
+        kill_stuck_encoder(state)
+
+        %{
+          state
+          | encoding: false,
+            video_id: nil,
+            video_path: nil,
+            last_progress_time: nil,
+            last_progress_percent: nil,
+            warned: false,
+            os_pid: nil
+        }
+
+      # Stuck for 30 min - warn
+      elapsed > @warn_threshold and not state.warned ->
+        warn_stuck_encoder(state)
+        %{state | warned: true}
+
+      true ->
+        state
+    end
+  end
+
+  defp warn_stuck_encoder(state) do
+    Logger.warning(
+      "Encoder may be stuck - no progress for 30+ minutes. " <>
+        "Video ID: #{state.video_id}, Path: #{state.video_path}"
+    )
+
+    Events.broadcast_event(:encoder_health_alert, %{
+      reason: :stalled_30_min,
+      video_id: state.video_id,
+      video_path: state.video_path
+    })
+  end
+
+  defp kill_stuck_encoder(%{os_pid: nil} = state) do
+    Logger.error(
+      "Could not kill stuck encoder - no OS PID available. " <>
+        "Video ID: #{state.video_id}, Path: #{state.video_path}"
+    )
+  end
+
+  defp kill_stuck_encoder(%{os_pid: os_pid} = state) do
+    Logger.error(
+      "Killing stuck encoder process (PID: #{os_pid}) after 1 hour of no progress. " <>
+        "Video ID: #{state.video_id}, Path: #{state.video_path}"
+    )
+
+    # Run the kill command asynchronously so the HealthCheck GenServer is not blocked
+    Task.start(fn -> System.cmd("kill", [to_string(os_pid)]) end)
+
+    Events.broadcast_event(:encoder_health_alert, %{
+      reason: :killed_stuck_process,
+      video_id: state.video_id,
+      video_path: state.video_path,
+      os_pid: os_pid
+    })
+  end
+
+  defp schedule_check, do: Process.send_after(self(), :health_check, @check_interval)
+end

lib/reencodarr/post_processor.ex

@@ -36,6 +36,8 @@ defmodule Reencodarr.PostProcessor do
     )
 
     # Record detailed failure information with enhanced context
+    # Note: We don't notify Sonarr/Radarr on failure since the original file is unchanged.
+    # Sync notification is only needed when the file has been successfully replaced.
     Reencodarr.FailureTracker.record_process_failure(video, exit_code, context: context)
 
     :ok
@@ -76,8 +78,12 @@ defmodule Reencodarr.PostProcessor do
   defp process_intermediate_success(video, actual_path) do
     case Repo.reload(video) do
       nil ->
-        Logger.error("Failed to reload video #{video.id}: Video not found.")
-        :ok
+        Logger.error(
+          "Failed to reload video #{video.id}: Video not found. Still attempting sync."
+        )
+
+        # File was moved successfully, still try to notify Sonarr with original video data
+        finalize_and_sync(video, actual_path)
 
       reloaded ->
         process_reloaded_video(reloaded, actual_path)
@@ -93,8 +99,12 @@ defmodule Reencodarr.PostProcessor do
         finalize_and_sync(updated_video, actual_path)
 
       {:error, reason} ->
-        Logger.error("Failed to mark video #{video.id} as re-encoded: #{inspect(reason)}")
-        :ok
+        Logger.error(
+          "Failed to mark video #{video.id} as re-encoded: #{inspect(reason)}. Still attempting sync."
+        )
+
+        # DB update failed but file was moved, still try to finalize and notify Sonarr
+        finalize_and_sync(video, actual_path)
     end
   end