diff --git a/lib/reencodarr/ab_av1/encode.ex b/lib/reencodarr/ab_av1/encode.ex index 842267af..5bed3469 100644 --- a/lib/reencodarr/ab_av1/encode.ex +++ b/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 diff --git a/lib/reencodarr/application.ex b/lib/reencodarr/application.ex index 089bf435..aa81f059 100644 --- a/lib/reencodarr/application.ex +++ b/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 diff --git a/lib/reencodarr/encoder/health_check.ex b/lib/reencodarr/encoder/health_check.ex new file mode 100644 index 00000000..161c4ae4 --- /dev/null +++ b/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 diff --git a/lib/reencodarr/post_processor.ex b/lib/reencodarr/post_processor.ex index 3b7b67a6..c2ca2e6b 100644 --- a/lib/reencodarr/post_processor.ex +++ b/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 diff --git a/lib/reencodarr_web/live/dashboard_live.ex b/lib/reencodarr_web/live/dashboard_live.ex index 8b7214c1..270dc948 100644 --- a/lib/reencodarr_web/live/dashboard_live.ex +++ b/lib/reencodarr_web/live/dashboard_live.ex @@ -195,6 +195,26 @@ defmodule ReencodarrWeb.DashboardLive do {:noreply, assign(socket, :analyzer_progress, progress)} end + # Encoder health alert handler + @impl true + def handle_info({:encoder_health_alert, data}, socket) do + filename = if data.video_path, do: Path.basename(data.video_path), else: "unknown" + + message = + case data.reason do + :stalled_30_min -> + "Encoder may be stuck - no progress for 30+ minutes (#{filename})" + + :killed_stuck_process -> + "Killed stuck encoder after 1 hour (#{filename})" + + _ -> + "Encoder health alert: #{inspect(data.reason)}" + end + + {:noreply, put_flash(socket, :error, message)} + end + # Completion and reset handlers @impl true def handle_info({:encoding_completed, _data}, socket) do